A Django integration package for Electric SQL, enabling real-time data synchronization between Django applications and client devices.
Electric SQL is a sync engine that enables local-first development with real-time synchronization. Instead of traditional request-response patterns, Electric provides efficient delta-based syncing between your backend database and client applications.
- Seamless Django Integration: Works naturally with Django ORM and existing models
- Real-time Sync: Bidirectional synchronization between Django and Electric SQL
- Shape-based Sync: Sync specific data subsets using SQL-like filters
- Offline-First: Build applications that work offline and sync when connected
- Type-Safe: Full type hints and mypy support
- Battle-Tested: Comprehensive test suite with pytest
- Developer-Friendly: Management commands, admin integration, and decorators
- Signals Support: React to sync events with Django signals
- Caching: Built-in caching for sync operations
pip install django-electric# settings.py
INSTALLED_APPS = [
...
'django_electric',
...
]# settings.py
# Electric service URL (required)
ELECTRIC_SERVICE_URL = 'http://localhost:5133'
# Optional settings
ELECTRIC_AUTH_TOKEN = 'your-auth-token' # If using authentication
ELECTRIC_AUTO_SYNC = True # Auto-sync on model save
ELECTRIC_SYNC_BATCH_SIZE = 100 # Batch size for sync operations
ELECTRIC_TIMEOUT = 30 # Request timeout in secondsfrom django.db import models
from django_electric.models import ElectricSyncMixin
from django_electric.managers import ElectricManager
class Article(ElectricSyncMixin, models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
published = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
# Use Electric manager for sync capabilities
objects = ElectricManager()
class Meta:
electric_sync = True # Enable sync for this model
electric_where = "published = true" # Only sync published articles# Sync to Electric
result = Article.electric_sync()
print(f"Shape ID: {result['shape_id']}")
# Pull data from Electric
stats = Article.electric_pull()
print(f"Created: {stats['created']}, Updated: {stats['updated']}")
# Get synced data
articles = Article.electric_get_data(limit=50)# Check Electric status
python manage.py electric_status
# Sync all models
python manage.py electric_sync --all
# Sync specific model
python manage.py electric_sync --model myapp.Article
# Pull data from Electric
python manage.py electric_sync --model myapp.Article --pullElectric SQL uses "shapes" to define what data to sync. A shape is a subset of your data defined by:
- Table: Which model/table to sync
- Where clause: SQL filter for the data
- Columns: Specific fields to include
- Include: Related models to sync
# Sync only published articles
Article.electric_sync(where="published = true")
# Sync with specific columns
shape = Article.get_electric_shape(
columns=['id', 'title', 'created_at']
)Configure sync behavior in your model's Meta class:
class Meta:
electric_sync = True # Enable sync
electric_where = "status = 'active'" # Default filter
electric_columns = ['id', 'name', 'email'] # Specific columnsModel.electric_sync(where=None, force=False)
Sync model to Electric SQL.
Article.electric_sync(where="published = true", force=True)Model.electric_pull(where=None, update_existing=True)
Pull data from Electric to local database.
stats = Article.electric_pull(where="created_at > '2024-01-01'")Model.electric_get_data(where=None, offset=0, limit=100)
Get synced data from Electric.
articles = Article.electric_get_data(limit=50)@electric_cached(timeout=300)
Cache Electric sync results.
from django_electric.decorators import electric_cached
@electric_cached(timeout=600)
def get_articles():
return Article.electric_get_data()@electric_retry(max_attempts=3, delay=1.0)
Retry sync operations on failure.
from django_electric.decorators import electric_retry
@electric_retry(max_attempts=5)
def sync_all_data():
Article.electric_sync()
Comment.electric_sync()For advanced usage, use the ElectricClient directly:
from django_electric.client import ElectricClient
with ElectricClient() as client:
# Create a shape
shape = client.create_shape(
table="articles",
where="published = true",
columns=["id", "title", "content"]
)
# Sync the shape
result = client.sync_shape(shape)
# Get shape data
data = client.get_shape_data(result['shape_id'])Check Electric SQL connection and configuration.
python manage.py electric_statusSync models with Electric SQL.
# Sync all models
python manage.py electric_sync --all
# Sync specific model
python manage.py electric_sync --model blog.Post
# Sync with custom filter
python manage.py electric_sync --model blog.Post --where "published = true"
# Force sync (ignore cache)
python manage.py electric_sync --model blog.Post --force
# Pull data from Electric
python manage.py electric_sync --model blog.Post --pullAdd Electric sync to your admin actions:
from django.contrib import admin
from .models import Article
@admin.register(Article)
class ArticleAdmin(admin.ModelAdmin):
actions = ['sync_to_electric', 'pull_from_electric']
def sync_to_electric(self, request, queryset):
result = Article.electric_sync()
self.message_user(request, f"Synced: {result}")
def pull_from_electric(self, request, queryset):
stats = Article.electric_pull()
self.message_user(
request,
f"Created: {stats['created']}, Updated: {stats['updated']}"
)React to sync events:
from django_electric.signals import (
electric_sync_started,
electric_sync_completed,
electric_sync_failed,
)
@receiver(electric_sync_completed)
def on_sync_complete(sender, **kwargs):
print(f"Sync completed for {sender.__name__}")
@receiver(electric_sync_failed)
def on_sync_failed(sender, error, **kwargs):
print(f"Sync failed: {error}")See the examples/demo_project directory for a complete working Django application demonstrating:
- Model configuration with Electric sync
- Management command usage
- Admin panel integration
- API views with sync operations
- Decorator usage
# Clone the repository
git clone https://github.com/Apoorvgarg-Creator/dj-electric.git
cd django-electric
# Create virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install in development mode
pip install -e ".[dev]"# Run all tests
pytest
# Run with coverage
pytest --cov=django_electric
# Run specific test file
pytest tests/test_client.py
# Run in watch mode
pytest-watch# Format code
black .
isort .
# Lint
flake8 django_electric
mypy django_electric
# Type checking
mypy django_electricContributions are welcome! Please see CONTRIBUTING.md for details.
- Fork the repository
- Create a feature branch (
git checkout -b feature/amazing-feature) - Make your changes
- Add tests for your changes
- Ensure all tests pass (
pytest) - Commit your changes (
git commit -m 'Add amazing feature') - Push to the branch (
git push origin feature/amazing-feature) - Open a Pull Request
# Build the package
python -m build
# Upload to PyPI (requires credentials)
twine upload dist/*- Python >= 3.9
- Django >= 4.2
- Electric SQL service running
Apache License 2.0 - see LICENSE for details.
- Issues: GitHub Issues
- Discussions: GitHub Discussions
- Documentation: Wiki
- Built for Electric SQL
- Inspired by local-first and offline-first architectures
- Thanks to the Django and Electric SQL communities
- WebSocket support for real-time updates
- Conflict resolution strategies
- Multi-tenancy support
- GraphQL integration
- Django REST Framework integration
- Async support with Django 4.2+
- Schema migration tools
- Performance monitoring and metrics
Made with ❤️ for the Django and Electric SQL communities