Database Design Patterns Every Python Backend Needs in Production
A customer called our support team. “My order is gone. I placed it yesterday and now it’s not showing up.” Our support engineer checked the database. The order wasn’t there. Not in any table. Not in any log. Just… gone.
Thirty minutes of digging revealed what happened. An admin user was cleaning up test orders from the staging environment and accidentally ran the delete query against production. The WHERE clause had a typo. Instead of deleting orders with status = 'test', it deleted orders with status = 'pending'. Twenty-three real customer orders, permanently erased from the database.
We had no soft deletes. No audit trail. No way to recover the data except from a backup taken 18 hours earlier. We restored the backup to a separate database, manually extracted the 23 orders, and re-inserted them. It took four hours.
That incident drove me to add five database patterns to every project I build. Each one prevents a specific category of data disaster that I’ve either experienced firsthand or seen in a code review. None of them are complicated. All of them are worth the 30 minutes of setup.
Pattern 1: Soft Deletes — Never Lose Data Again
Soft Delete Without Losing Data Database Design Pattern
Hard deletes (DELETE FROM orders WHERE id = 42) are permanent. Once the row is gone, it's gone. Soft deletes mark a row as deleted without removing it:
# models/mixins.py
from django.db import models
from django.utils import timezone
class SoftDeleteQuerySet(models.QuerySet):
def delete(self):
"""Override delete to soft-delete instead."""
return self.update(deleted_at=timezone.now())
def hard_delete(self):
"""Actually delete from the database."""
return super().delete()
def alive(self):
"""Only non-deleted records."""
return self.filter(deleted_at__isnull=True)
def dead(self):
"""Only soft-deleted records."""
return self.filter(deleted_at__isnull=False)
class SoftDeleteManager(models.Manager):
def get_queryset(self):
return SoftDeleteQuerySet(self.model, using=self._db).alive()
class SoftDeleteModel(models.Model):
"""Mixin that adds soft-delete behavior to any model."""
deleted_at = models.DateTimeField(null=True, blank=True, db_index=True)
objects = SoftDeleteManager() # Default: only alive records
all_objects = models.Manager() # Includes soft-deleted records
class Meta:
abstract = True
def delete(self, *args, **kwargs):
self.deleted_at = timezone.now()
self.save(update_fields=['deleted_at'])
def hard_delete(self, *args, **kwargs):
super().delete(*args, **kwargs)
def restore(self):
self.deleted_at = None
self.save(update_fields=['deleted_at'])
@property
def is_deleted(self):
return self.deleted_at is not None
Now any model can use soft deletes by inheriting from this mixin:
class Order(SoftDeleteModel):
customer = models.ForeignKey(Customer, on_delete=models.CASCADE)
total = models.DecimalField(max_digits=10, decimal_places=2)
status = models.CharField(max_length=20)
created_at = models.DateTimeField(auto_now_add=True)
Usage is transparent:
# Soft delete — sets deleted_at, row stays in database
order = Order.objects.get(id=42)
order.delete()
# Default queries exclude soft-deleted records
orders = Order.objects.all() # Only alive orders
# Explicitly include soft-deleted records
all_orders = Order.all_objects.all() # Everything, including deleted
# Restore a soft-deleted order
order = Order.all_objects.get(id=42)
order.restore()
For SQLAlchemy / FastAPI:
from sqlalchemy import Column, DateTime, event
from sqlalchemy.orm import Query
from datetime import datetime, timezone
class SoftDeleteMixin:
deleted_at = Column(DateTime, nullable=True, index=True)
def soft_delete(self):
self.deleted_at = datetime.now(timezone.utc)
def restore(self):
self.deleted_at = None
@property
def is_deleted(self):
return self.deleted_at is not None
# Apply a default filter to exclude soft-deleted records
@event.listens_for(Query, "before_compile", retval=True)
def soft_delete_filter(query):
for desc in query.column_descriptions:
entity = desc.get("entity")
if entity and hasattr(entity, "deleted_at"):
query = query.filter(entity.deleted_at.is_(None))
return query
The Partial Index That Keeps Queries Fast
Soft deletes mean your table grows forever — deleted rows accumulate. Add a partial index so queries only scan alive records:
class Meta:
indexes = [
models.Index(
fields=['status', 'created_at'],
condition=models.Q(deleted_at__isnull=True),
name='idx_orders_active'
),
]
PostgreSQL only indexes rows where deleted_at IS NULL. The index stays small and fast regardless of how many rows are soft-deleted.
Pattern 2: Audit Trail — Know Who Changed What, When
Database Audit Trail — Know Who Changed What, When
When something goes wrong in production — data corruption, unauthorized changes, unexplained state transitions — the first question is always “what happened?” Without an audit trail, the answer is “we don’t know.”
# models/audit.py
import json
from django.db import models
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
class AuditLog(models.Model):
class Action(models.TextChoices):
CREATE = 'create'
UPDATE = 'update'
DELETE = 'delete'
# What was changed
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
object_id = models.PositiveIntegerField()
content_object = GenericForeignKey('content_type', 'object_id')
# Who changed it
user_id = models.IntegerField(null=True)
user_email = models.CharField(max_length=255, null=True)
# What happened
action = models.CharField(max_length=10, choices=Action.choices)
changes = models.JSONField(default=dict)
timestamp = models.DateTimeField(auto_now_add=True, db_index=True)
# Request context
ip_address = models.GenericIPAddressField(null=True)
user_agent = models.TextField(blank=True, default='')
class Meta:
indexes = [
models.Index(fields=['content_type', 'object_id', '-timestamp']),
models.Index(fields=['user_id', '-timestamp']),
]
ordering = ['-timestamp']
Recording Changes Automatically
Instead of manually logging in every view, use a mixin that captures changes on save:
# models/mixins.py
from django.contrib.contenttypes.models import ContentType
class AuditableMixin:
"""Mixin that automatically logs changes to the audit trail."""
_audit_user = None # Set this before saving
def save(self, *args, **kwargs):
if self.pk:
self._log_update()
else:
# Schedule create log after the object has a PK
is_new = True
super().save(*args, **kwargs)
if 'is_new' in dir() and is_new:
self._log_create()
def _log_create(self):
AuditLog.objects.create(
content_type=ContentType.objects.get_for_model(self),
object_id=self.pk,
user_id=getattr(self._audit_user, 'id', None),
user_email=getattr(self._audit_user, 'email', None),
action=AuditLog.Action.CREATE,
changes=self._serialize_fields(),
)
def _log_update(self):
try:
old = type(self).objects.get(pk=self.pk)
except type(self).DoesNotExist:
return
changes = {}
for field in self._meta.fields:
old_value = getattr(old, field.attname)
new_value = getattr(self, field.attname)
if old_value != new_value:
changes[field.name] = {
'old': str(old_value),
'new': str(new_value),
}
if changes:
AuditLog.objects.create(
content_type=ContentType.objects.get_for_model(self),
object_id=self.pk,
user_id=getattr(self._audit_user, 'id', None),
user_email=getattr(self._audit_user, 'email', None),
action=AuditLog.Action.UPDATE,
changes=changes,
)
def _serialize_fields(self):
return {
field.name: str(getattr(self, field.attname))
for field in self._meta.fields
}
Setting the Audit User from the Request
Use middleware to capture the current user and make it available to models:
# middleware/audit.py
import threading
_thread_local = threading.local()
def get_current_user():
return getattr(_thread_local, 'user', None)
class AuditMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
_thread_local.user = getattr(request, 'user', None)
response = self.get_response(request)
_thread_local.user = None
return response
Update the mixin to use it automatically:
class AuditableMixin:
@property
def _audit_user(self):
from middleware.audit import get_current_user
return get_current_user()
Now every model change is automatically logged:
class Order(AuditableMixin, SoftDeleteModel):
customer = models.ForeignKey(Customer, on_delete=models.CASCADE)
total = models.DecimalField(max_digits=10, decimal_places=2)
status = models.CharField(max_length=20)
# Querying the audit trail
audit_entries = AuditLog.objects.filter(
content_type=ContentType.objects.get_for_model(Order),
object_id=42,
)
for entry in audit_entries:
print(f"{entry.timestamp} | {entry.user_email} | {entry.action} | {entry.changes}")
# Output:
# 2026-06-10 14:23 | admin@company.com | update | {"status": {"old": "pending", "new": "shipped"}}
# 2026-06-09 10:15 | alice@company.com | create | {"customer_id": "42", "total": "99.99", ...}
When that accidental delete happened, an audit trail would have told us exactly who ran the query, when, and what data was affected — in seconds instead of hours.
Pattern 3: Multi-Tenancy with Row-Level Security
Multi-Tenancy with Row-Level Security
If you’re building a SaaS product where multiple customers share the same database, tenant isolation is critical. One missing WHERE tenant_id = ? and you're showing Customer A's data to Customer B.
The Application-Level Approach
# models/mixins.py
class TenantMixin(models.Model):
"""Mixin that adds tenant isolation to any model."""
tenant = models.ForeignKey(
'tenants.Tenant',
on_delete=models.CASCADE,
db_index=True,
)
class Meta:
abstract = True
class TenantManager(models.Manager):
def get_queryset(self):
from middleware.tenant import get_current_tenant
qs = super().get_queryset()
tenant = get_current_tenant()
if tenant:
return qs.filter(tenant=tenant)
return qs
# middleware/tenant.py
import threading
_thread_local = threading.local()
def get_current_tenant():
return getattr(_thread_local, 'tenant', None)
class TenantMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
if hasattr(request, 'user') and request.user.is_authenticated:
_thread_local.tenant = request.user.tenant
else:
_thread_local.tenant = None
response = self.get_response(request)
_thread_local.tenant = None
return response
class Order(TenantMixin):
customer = models.ForeignKey(Customer, on_delete=models.CASCADE)
total = models.DecimalField(max_digits=10, decimal_places=2)
objects = TenantManager()
# This automatically filters by the current user's tenant
orders = Order.objects.all() # Only shows orders for the current tenant
The Database-Level Approach (PostgreSQL Row-Level Security)
Application-level filtering relies on every query including the tenant filter. One bug — one forgotten filter in a management command or admin query — and data leaks across tenants. PostgreSQL’s Row-Level Security (RLS) enforces isolation at the database level:
-- Enable RLS on the orders table
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;
-- Create a policy that only shows rows matching the current tenant
CREATE POLICY tenant_isolation ON orders
USING (tenant_id = current_setting('app.current_tenant_id')::integer);
# Set the tenant context at the start of each request
class RLSTenantMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
if hasattr(request, 'user') and request.user.is_authenticated:
from django.db import connection
with connection.cursor() as cursor:
cursor.execute(
"SET LOCAL app.current_tenant_id = %s",
[request.user.tenant_id]
)
return self.get_response(request)
Now even raw SQL queries, admin panels, and management commands are isolated by tenant. The database itself enforces the policy — your application code can’t accidentally bypass it.
Indexing for Multi-Tenancy
Every index on a multi-tenant table should include tenant_id as the first column:
class Order(TenantMixin):
status = models.CharField(max_length=20)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
indexes = [
# BAD — scans across all tenants
# models.Index(fields=['status', 'created_at']),
# GOOD — scoped to tenant first
models.Index(fields=['tenant', 'status', '-created_at']),
]
Without tenant_id leading the index, a query like WHERE tenant_id = 5 AND status = 'pending' has to scan the full index and filter. With tenant_id first, PostgreSQL jumps directly to tenant 5's entries.
Pattern 4: Optimistic Locking — Prevent Concurrent Update Conflicts
Version Check Prevents Silent Overwrite
When two users edit the same record simultaneously, the last save wins — silently overwriting the first user’s changes. Optimistic locking detects this conflict:
# models/mixins.py
class OptimisticLockMixin(models.Model):
version = models.IntegerField(default=0)
class Meta:
abstract = True
def save(self, *args, **kwargs):
if self.pk:
# Update only if the version matches
updated = type(self).objects.filter(
pk=self.pk,
version=self.version,
).update(
version=models.F('version') + 1,
**{
field.attname: getattr(self, field.attname)
for field in self._meta.fields
if field.attname not in ('id', 'version')
}
)
if updated == 0:
raise ConflictError(
"This record was modified by another user. "
"Please reload and try again."
)
self.version += 1
else:
super().save(*args, **kwargs)
For SQLAlchemy / FastAPI:
from sqlalchemy import Column, Integer
from sqlalchemy.orm import validates
class OptimisticLockMixin:
version = Column(Integer, default=0, nullable=False)
def update_with_lock(db: Session, model_class, record_id: int, version: int, updates: dict):
"""Update a record with optimistic locking."""
updated = (
db.query(model_class)
.filter(
model_class.id == record_id,
model_class.version == version,
)
.update({
**updates,
"version": model_class.version + 1,
})
)
if updated == 0:
raise ConflictError("Record was modified by another user")
db.commit()
In your API endpoint:
@app.put("/api/orders/{order_id}")
def update_order(
order_id: int,
data: OrderUpdate,
db: Session = Depends(get_db),
):
# Client sends the version they last saw
update_with_lock(
db=db,
model_class=Order,
record_id=order_id,
version=data.version,
updates={"status": data.status, "notes": data.notes},
)
return {"success": True}
The frontend includes the version field in every update request. If two users load the same order (version 3), the first update succeeds (version becomes 4), and the second update fails because version 3 no longer exists. The second user gets a clear error message instead of silently losing their changes.
Pattern 5: Timestamp Columns on Every Table
The simplest pattern and the one with the highest ROI. Every table should have created_at and updated_at:
# models/mixins.py
class TimestampMixin(models.Model):
created_at = models.DateTimeField(auto_now_add=True, db_index=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
abstract = True
For SQLAlchemy:
from sqlalchemy import Column, DateTime, func
class TimestampMixin:
created_at = Column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
index=True,
)
updated_at = Column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
nullable=False,
)
These two columns cost nothing to maintain and provide enormous value:
- Debugging: “When was this order created? When was it last modified?” Without timestamps, you’re guessing.
- Data sync: External systems can ask “give me all records updated since my last sync” using
WHERE updated_at > ?. - Ordering: Most list endpoints default to newest-first. Without
created_at, you're sorting byid, which is less reliable for display order. - Compliance: GDPR, SOC 2, and other frameworks require audit capabilities. Timestamps are the minimum baseline.
Putting It All Together: The Base Model
The Production Base Model
Combine all five patterns into a base model that every table in your project inherits from:
# models/base.py
from django.db import models
from django.utils import timezone
class BaseModel(SoftDeleteModel, TimestampMixin, AuditableMixin):
"""
Base model with:
- Soft deletes (deleted_at)
- Timestamps (created_at, updated_at)
- Audit trail (automatic change logging)
"""
class Meta:
abstract = True
class TenantBaseModel(TenantMixin, BaseModel):
"""
Base model for multi-tenant tables. Adds:
- Tenant isolation (tenant_id)
"""
class Meta:
abstract = True
Now every model in your project:
class Order(TenantBaseModel, OptimisticLockMixin):
customer = models.ForeignKey(Customer, on_delete=models.CASCADE)
total = models.DecimalField(max_digits=10, decimal_places=2)
status = models.CharField(max_length=20)
This order model has soft deletes, timestamps, audit logging, tenant isolation, and optimistic locking — without a single line of that logic in the model itself. The mixins handle everything.
Bottom Line
Five Patterns, Five Data Disasters Prevented
Database design patterns aren’t about making your schema “clean.” They’re about preventing data disasters that cost hours, days, or customer trust to recover from.
Soft deletes prevent permanent data loss from accidental or malicious deletions. Audit trails tell you exactly who changed what, when, and why. Multi-tenancy ensures customers never see each other’s data. Optimistic locking prevents silent overwrites from concurrent edits. Timestamps give you the minimum observability every table needs.
Each pattern takes about 30 minutes to implement as a mixin. Once implemented, every model in your project inherits the protection automatically. The base model approach means new developers don’t need to remember to add these patterns — they’re the default.
Those 23 orders that were accidentally deleted? With soft deletes, they’d have been recoverable in seconds — Order.all_objects.filter(deleted_at__isnull=False).update(deleted_at=None). With an audit trail, we'd have known exactly what happened within minutes. Five patterns. Thirty minutes of setup. And you never write "we lost the data" in a post-mortem again.
Which of these patterns do you use in your projects? I’m curious about soft deletes specifically — do you implement them at the application level or use database triggers? And has anyone implemented Row-Level Security in PostgreSQL for multi-tenancy? Share your approach in the comments.
