Skip to content

Django Database Patterns

This guide covers production-tested patterns for managing databases in Django applications, with focus on multi-tenant architectures, database routing, and optimization strategies.

Multi-Tenant Architecture Overview

Multi-tenant Django applications serve multiple clients (tenants) from a single application instance, each with isolated data storage. The two primary approaches are:

  1. Database-per-tenant: Each tenant has a dedicated database (covered here)
  2. Schema-per-tenant: All tenants share a database but use separate schemas
  3. Row-level isolation: All tenants share tables with tenant discriminator columns

The database-per-tenant model provides the strongest isolation guarantees and simplest backup/restore operations, at the cost of increased connection overhead and infrastructure complexity.

When to Use Database-Per-Tenant

Choose this model when:

  • You need strong data isolation for compliance or security
  • Tenants have significantly different sizes or resource requirements
  • You need per-tenant backup and restore capabilities
  • You're comfortable managing multiple database connections

Database Configuration Patterns

Development Configuration

Development environments typically use a single database host with multiple databases for simplicity:

# settings/development_databases.py
"""Database configuration settings for development environment."""

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "poseidon",
        "HOST": "db",
        "USER": "root",
        "PASSWORD": "Vanadium23",
        "OPTIONS": {
            "charset": "utf8mb4",
        },
    },
    "poseidon": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "poseidon",
        "HOST": "db",
        "USER": "root",
        "PASSWORD": "Vanadium23",
        "OPTIONS": {
            "charset": "utf8mb4",
        },
    },
    "accounts": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "accounts",
        "HOST": "db",
        "USER": "root",
        "PASSWORD": "Vanadium23",
        "OPTIONS": {
            "charset": "utf8mb4",
        },
    },
    "accounts_RO": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "accounts",
        "HOST": "db",
        "USER": "root",
        "PASSWORD": "Vanadium23",
        "OPTIONS": {
            "charset": "utf8mb4",
        },
    },
    # Tenant databases...
    "TENANT_NAME": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "TENANT_NAME",
        "HOST": "db",
        "USER": "root",
        "PASSWORD": "Vanadium23",
        "OPTIONS": {
            "charset": "utf8mb4",
        },
    },
    "TENANT_NAME_RO": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "TENANT_NAME",
        "HOST": "db",
        "USER": "root",
        "PASSWORD": "Vanadium23",
        "OPTIONS": {
            "charset": "utf8mb4",
        },
    },
}

Read Replicas in Development

Development environments typically point read replicas (suffix _RO) to the same database as write connections to simplify local setup.

Production Configuration

Production environments use RDS Proxy for connection pooling and separate read replicas:

# settings/production_databases.py
from poseidon.commons.config.ps_config import pconfig

POSEIDON_DATABASE_USERNAME = pconfig.get_param("POSEIDON_DATABASE_USERNAME")
POSEIDON_DATABASE_PASSWORD = pconfig.get_param("POSEIDON_DATABASE_PASSWORD")
DATABASE_RO_CONNECTION_STRING = "planion-proxy-1-read-only.endpoint.proxy-chrrlhnvdwuh.us-east-1.rds.amazonaws.com"
DATABASE_RW_CONNECTION_STRING = "planion-proxy-1.proxy-chrrlhnvdwuh.us-east-1.rds.amazonaws.com"
TENANTS = pconfig.get_param("PLANSTONE_TENANTS_LIST").split(",")

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "poseidon",
        "USER": POSEIDON_DATABASE_USERNAME,
        "PASSWORD": POSEIDON_DATABASE_PASSWORD,
        "HOST": DATABASE_RW_CONNECTION_STRING,
    },
    "poseidon": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "poseidon",
        "USER": POSEIDON_DATABASE_USERNAME,
        "PASSWORD": POSEIDON_DATABASE_PASSWORD,
        "HOST": DATABASE_RW_CONNECTION_STRING,
    },
    "poseidon_RO": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "poseidon",
        "USER": POSEIDON_DATABASE_USERNAME,
        "PASSWORD": POSEIDON_DATABASE_PASSWORD,
        "HOST": DATABASE_RO_CONNECTION_STRING,
    },
    "accounts": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "accounts",
        "USER": POSEIDON_DATABASE_USERNAME,
        "PASSWORD": POSEIDON_DATABASE_PASSWORD,
        "HOST": DATABASE_RW_CONNECTION_STRING,
    },
    "accounts_RO": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "accounts",
        "USER": POSEIDON_DATABASE_USERNAME,
        "PASSWORD": POSEIDON_DATABASE_PASSWORD,
        "HOST": DATABASE_RO_CONNECTION_STRING,
    },
}

# Dynamically create tenant databases from configuration
for tenant in TENANTS:
    tenant = tenant.strip().upper()
    if tenant:
        DATABASES[f"{tenant}"] = {
            "ENGINE": "django.db.backends.mysql",
            "NAME": tenant,
            "USER": POSEIDON_DATABASE_USERNAME,
            "PASSWORD": POSEIDON_DATABASE_PASSWORD,
            "HOST": DATABASE_RW_CONNECTION_STRING,
            "OPTIONS": {
                "charset": "utf8mb4",
            },
        }

        DATABASES[f"{tenant}_RO"] = {
            "ENGINE": "django.db.backends.mysql",
            "NAME": tenant,
            "USER": POSEIDON_DATABASE_USERNAME,
            "PASSWORD": POSEIDON_DATABASE_PASSWORD,
            "HOST": DATABASE_RO_CONNECTION_STRING,
            "OPTIONS": {
                "charset": "utf8mb4",
            },
        }

Dynamic Configuration Security

  • Load credentials from secure parameter stores (AWS SSM, HashiCorp Vault)
  • Never commit credentials to version control
  • Use environment-specific parameter paths
  • Implement parameter caching with appropriate TTLs

Testing Configuration

Test environments require careful isolation to prevent data leakage between tests:

# settings/testing_databases.py
"""
Simplified test database configuration for Poseidon.

This configuration defines two test databases:
1. test_poseidon - The main database for most tests and the default database
2. test_testclient - A separate database for tenant-specific tests

Most database aliases mirror the default database to:
- Reduce database creation overhead
- Simplify test isolation
- Make it clear which databases are for testing
"""

from .development_databases import DATABASES as DEV_DATABASES

# Create a new dictionary for test databases
DATABASES = {}

# Define consistent test database names
DEFAULT_TEST_DB = "test_poseidon"
TENANT_TEST_DB = "test_testclient"

# Configure each database to use the appropriate test database
for alias, config in DEV_DATABASES.items():
    DATABASES[alias] = config.copy()

    # Special handling for TESTCLIENT - it gets its own database
    if alias in ("TESTCLIENT", "TESTCLIENT_RO"):
        DATABASES[alias].setdefault("TEST", {})["NAME"] = TENANT_TEST_DB
    elif alias == "default":
        # The default database gets the default test database name
        DATABASES[alias].setdefault("TEST", {})["NAME"] = DEFAULT_TEST_DB
    else:
        # All other databases mirror the default test database
        DATABASES[alias].setdefault("TEST", {})["MIRROR"] = "default"

    # Ensure all test databases use transactions for isolation
    DATABASES[alias].setdefault("TEST", {})["SERIALIZE"] = False

    # Force Django to create and destroy test databases properly
    DATABASES[alias].setdefault("TEST", {})["CREATE_DB"] = True

Test Database Optimization

Use TEST.MIRROR to point multiple database aliases to the same physical test database. This dramatically reduces test setup time when you have many tenant databases but only need to test a few.

Database Routing

Database routers determine which database Django uses for each operation. The router must implement four methods: db_for_read, db_for_write, allow_relation, and allow_migrate.

Simple Router Implementation

# db_routers.py

class DBRouter:
    """
    Database router that delegates database selection to models.

    This router checks for a _DATABASE attribute on models and uses
    that to determine which database to use for read/write operations.
    If no _DATABASE attribute exists, Django falls back to the 'default' database.
    """

    def db_for_read(self, model, **hints):
        """
        Route read operations to the database specified by the model.

        Args:
            model: The model class being queried
            **hints: Additional routing hints (instance, etc.)

        Returns:
            Database alias string or None to use default routing
        """
        # Let models specify their own database if needed
        return getattr(model, "_DATABASE", None)

    def db_for_write(self, model, **hints):
        """
        Route write operations to the database specified by the model.

        Args:
            model: The model class being saved
            **hints: Additional routing hints (instance, etc.)

        Returns:
            Database alias string or None to use default routing
        """
        # Let models specify their own database if needed
        return getattr(model, "_DATABASE", None)

    def allow_migrate(self, db, app_label, model_name=None, **hints):
        """
        Determine if migration should run on this database.

        Args:
            db: Database alias
            app_label: Application label
            model_name: Name of the model (lowercase)
            **hints: Additional routing hints

        Returns:
            True if migration should run, False if not, None for default behavior
        """
        # Skip migrations for legacy models
        if model_name == "abstract" or model_name == "abstractauthors":
            return False
        return True

Router Limitations

Database routers only affect Django ORM operations. Raw SQL queries, management commands, and third-party packages may bypass router logic entirely.

Registering the Router

Configure the router in your settings module:

# settings/base.py or settings/development.py

DATABASE_ROUTERS = ["poseidon.db_routers.DBRouter"]

Django evaluates routers in order, using the first non-None response. Place more specific routers before general ones.

Advanced Router Patterns

For more complex scenarios, implement additional logic:

class TenantRouter:
    """
    Advanced router with tenant-aware routing and read replica support.
    """

    def db_for_read(self, model, **hints):
        """Route reads to read replicas when available."""
        database = getattr(model, "_DATABASE", None)

        if database:
            # Check if model prefers read replicas
            if getattr(model, "_PREFER_READ_REPLICA", True):
                replica = f"{database}_RO"
                if replica in settings.DATABASES:
                    return replica
            return database

        return None

    def db_for_write(self, model, **hints):
        """Route writes to primary databases only."""
        database = getattr(model, "_DATABASE", None)

        # Never write to read replicas
        if database and database.endswith("_RO"):
            return database[:-3]  # Remove _RO suffix

        return database

    def allow_relation(self, obj1, obj2, **hints):
        """
        Allow relations only between models in the same database.
        """
        db1 = getattr(obj1._meta.model, "_DATABASE", "default")
        db2 = getattr(obj2._meta.model, "_DATABASE", "default")

        # Remove _RO suffix for comparison
        db1 = db1[:-3] if db1.endswith("_RO") else db1
        db2 = db2[:-3] if db2.endswith("_RO") else db2

        return db1 == db2

    def allow_migrate(self, db, app_label, model_name=None, **hints):
        """
        Control which databases receive migrations.
        """
        # Don't migrate read replicas
        if db.endswith("_RO"):
            return False

        # Skip legacy models
        if model_name in ("abstract", "abstractauthors"):
            return False

        # Only migrate app models to default database
        if app_label == "poseidon":
            return db == "default"

        return None

Read Replica Routing

Automatically routing reads to replicas can cause issues with read-after-write consistency. Consider explicit .using() calls for operations that must read their own writes.

Model Patterns

Models for Legacy Databases

When integrating with existing databases, use managed = False to prevent Django from managing the schema:

from django.db import models

class Accounts(models.Model):
    """
    Legacy accounts table from existing database.

    This model represents a table that existed before Django and should
    not be modified by Django migrations.
    """
    accountname = models.CharField(
        db_column="ACCOUNTNAME",
        max_length=75,
        blank=True,
        null=True
    )
    acronym = models.CharField(
        db_column="ACRONYM",
        primary_key=True,
        max_length=20
    )
    active = models.BooleanField(
        db_column="ACTIVE",
        blank=True,
        null=True
    )
    defaultconf = models.CharField(
        db_column="DEFAULTCONF",
        max_length=20,
        blank=True,
        null=True
    )

    class Meta:
        db_table = "accounts"
        managed = False
        ordering = ["acronym"]
        app_label = "poseidon"
        verbose_name_plural = "accounts"

    _DATABASE = "accounts"

Key patterns for legacy models:

  • managed = False: Prevents Django from creating/modifying the table
  • db_table: Explicitly maps to existing table name
  • db_column: Maps to existing column names (especially for uppercase/unusual naming)
  • _DATABASE: Custom attribute for database routing
  • app_label: Required for models in multi-app projects
  • blank=True, null=True: Matches existing schema constraints

Migration Conflicts

Even with managed = False, Django creates migration files tracking field changes. Use allow_migrate in your router to prevent these migrations from running.

Conference Model Example

from django.db import models
from django.utils import timezone

class PlanionContainer(models.Model):
    """
    Legacy conference container from Planion system.

    Represents a conference event with all associated metadata.
    This table existed before Django and should not be managed by migrations.
    """
    created_at = models.DateTimeField(blank=True, null=True)
    updated_at = models.DateTimeField(blank=True, null=True)
    code = models.CharField(max_length=8, primary_key=True, db_column="cfcnf")
    name = models.CharField(max_length=100, blank=True, null=True, db_column="cfname")
    sponsor = models.CharField(max_length=50, blank=True, null=True)
    date = models.DateField(blank=True, null=True)
    dateend = models.DateField(blank=True, null=True)
    confattend = models.DecimalField(max_digits=5, decimal_places=0, blank=True, null=True)
    cityid = models.CharField(max_length=6, blank=True, null=True)

    # ... many more fields ...

    scheduled_start = models.DateField(blank=True, null=True, db_column="schedstart")
    scheduled_end = models.DateField(blank=True, null=True, db_column="schedend")

    class Meta:
        managed = False
        db_table = "confrnce"
        app_label = "poseidon"
        verbose_name = "Conference"
        verbose_name_plural = "Conferences"

Pythonic Field Names

You can use Python-friendly field names while mapping to database columns with db_column. This improves code readability while maintaining compatibility: scheduled_start in Python maps to schedstart in the database.

Explicitly Selecting Databases

Override automatic routing with the .using() method:

# Read from read replica
demo_person_exists = People.objects.using("TESTCLIENT_RO").filter(pid=123823).exists()

# Write to primary database
People.objects.using("TESTCLIENT").create(
    pid=123823,
    first_name="John",
    last_name="Doe"
)

# Load fixtures into specific database
from django.core.management import call_command

call_command(
    "loaddata",
    "poseidon/fixtures/test/embed_demo_people.json",
    database="TESTCLIENT"
)

Consistency Considerations

When using read replicas, be aware of replication lag. Critical read-after-write operations should use .using("TENANT_NAME") to query the primary database.

Migration Management

Running Migrations

Django applies migrations based on DATABASE_ROUTERS configuration and allow_migrate logic:

# Migrate default database
python manage.py migrate

# Migrate specific database
python manage.py migrate --database=accounts

# Show migration status for all databases
python manage.py migrate --database=default --plan
python manage.py migrate --database=accounts --plan

Multi-Database Migration Strategy

  1. Identify managed databases: Only databases with managed = True models should receive migrations
  2. Configure allow_migrate: Prevent migrations from running on wrong databases
  3. Order migrations carefully: Create shared models before dependent models
  4. Test migration rollback: Ensure migrations can be safely reversed
# Example allow_migrate implementation
def allow_migrate(self, db, app_label, model_name=None, **hints):
    """
    Control migration application across databases.
    """
    # Application models go to default database only
    if app_label == "poseidon":
        return db == "default"

    # Don't migrate legacy/unmanaged models
    if model_name in ("abstract", "abstractauthors"):
        return False

    # Don't migrate read replicas
    if db.endswith("_RO"):
        return False

    # Auth/session models go to default
    if app_label in ("auth", "contenttypes", "sessions", "admin"):
        return db == "default"

    # Default to True for unknown cases
    return True

Migration Safety

Always test migrations in a staging environment first. Multi-database migrations can fail in complex ways, and rolling back partial failures is difficult.

Generating Migrations for Unmanaged Models

When you need to track model changes without applying migrations:

# Generate migration file
python manage.py makemigrations

# Edit migration to set managed = False
# Then run:
python manage.py migrate --fake

This creates migration history without modifying the database schema.

Connection Pooling and Optimization

Using RDS Proxy

AWS RDS Proxy provides connection pooling and improves application resilience:

# Production configuration with RDS Proxy
DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "poseidon",
        "USER": POSEIDON_DATABASE_USERNAME,
        "PASSWORD": POSEIDON_DATABASE_PASSWORD,
        "HOST": "planion-proxy-1.proxy-chrrlhnvdwuh.us-east-1.rds.amazonaws.com",
        "OPTIONS": {
            "charset": "utf8mb4",
            "connect_timeout": 5,
            "init_command": "SET sql_mode='STRICT_TRANS_TABLES'",
        },
        "CONN_MAX_AGE": 0,  # RDS Proxy handles pooling
    },
}

Benefits of RDS Proxy:

  • Connection reuse: Dramatically reduces connection overhead
  • IAM authentication: Eliminates password management
  • Automatic failover: Handles primary database failures transparently
  • Connection limits: Protects database from connection exhaustion

Connection Configuration

When using RDS Proxy, set CONN_MAX_AGE = 0 to disable Django's connection pooling. RDS Proxy handles pooling more efficiently at the infrastructure layer.

Connection Pooling with PgBouncer

For PostgreSQL deployments, PgBouncer provides similar benefits:

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": "poseidon",
        "USER": "poseidon_app",
        "PASSWORD": os.environ["DB_PASSWORD"],
        "HOST": "pgbouncer.internal",
        "PORT": "6432",
        "OPTIONS": {
            "connect_timeout": 5,
            "options": "-c statement_timeout=30000",
        },
        "CONN_MAX_AGE": 0,
    },
}

Connection Settings

# Conservative settings for many small requests
DATABASES = {
    "default": {
        # ... other settings ...
        "CONN_MAX_AGE": 60,  # Keep connections alive for 1 minute
        "CONN_HEALTH_CHECKS": True,  # Django 4.1+
        "OPTIONS": {
            "connect_timeout": 10,
            "read_timeout": 30,
            "write_timeout": 30,
        },
    },
}

# Aggressive settings for connection pools
DATABASES = {
    "default": {
        # ... other settings ...
        "CONN_MAX_AGE": 600,  # Keep connections for 10 minutes
        "CONN_HEALTH_CHECKS": True,
        "OPTIONS": {
            "pool_size": 20,
            "max_overflow": 10,
        },
    },
}

Connection Leaks

Long-lived connections can accumulate stale state or hold locks. Always use CONN_HEALTH_CHECKS = True in Django 4.1+ to validate connections before reuse.

Query Optimization

Reduce query count with proper prefetching:

# Bad: N+1 queries
conferences = PlanionContainer.objects.all()
for conf in conferences:
    print(conf.sponsor.name)  # Hits database each iteration

# Good: 2 queries with select_related
conferences = PlanionContainer.objects.select_related("sponsor").all()
for conf in conferences:
    print(conf.sponsor.name)  # Uses cached data

# Good: 2 queries with prefetch_related for reverse relations
accounts = Accounts.objects.prefetch_related("conferences").all()
for account in accounts:
    print(account.conferences.count())  # Uses cached data

Rules of thumb:

  • select_related: Use for ForeignKey and OneToOneField (forward relations)
  • prefetch_related: Use for ManyToManyField and reverse ForeignKey relations
  • Chain both: queryset.select_related("author").prefetch_related("tags")

Query Expression Examples

from django.db.models import Count, Q, F, Prefetch

# Annotate with related counts
accounts_with_conf_count = Accounts.objects.annotate(
    conference_count=Count("conferences")
).filter(conference_count__gt=5)

# Complex filtering with Q objects
active_conferences = PlanionContainer.objects.filter(
    Q(date__gte=timezone.now()) | Q(dateend__gte=timezone.now()),
    active=True
)

# F expressions for field comparisons
long_conferences = PlanionContainer.objects.filter(
    dateend__gt=F("date") + timezone.timedelta(days=30)
)

# Custom prefetch queries
recent_conferences = PlanionContainer.objects.filter(
    date__gte=timezone.now() - timezone.timedelta(days=365)
)
accounts_with_recent = Accounts.objects.prefetch_related(
    Prefetch("conferences", queryset=recent_conferences)
)

Indexing Strategy

Create indexes for frequently queried columns:

class PlanionContainer(models.Model):
    code = models.CharField(max_length=8, primary_key=True, db_column="cfcnf")
    name = models.CharField(max_length=100, blank=True, null=True, db_column="cfname")
    date = models.DateField(blank=True, null=True, db_index=True)
    dateend = models.DateField(blank=True, null=True, db_index=True)

    class Meta:
        managed = False
        db_table = "confrnce"
        indexes = [
            models.Index(fields=["date", "dateend"], name="conf_dates_idx"),
            models.Index(fields=["sponsor", "date"], name="conf_sponsor_date_idx"),
        ]

Legacy Database Indexes

For managed = False models, Django won't create indexes. Use raw SQL migrations or database admin tools to add indexes to existing tables.

Database Transactions

Transaction Basics

Django provides transaction control at multiple levels:

from django.db import transaction

# Decorator for entire view
@transaction.atomic
def create_conference(request):
    conference = PlanionContainer.objects.create(...)
    registration = Registration.objects.create(conference=conference)
    return HttpResponse("Created")

# Context manager for specific blocks
def update_conference(conference_id):
    with transaction.atomic():
        conference = PlanionContainer.objects.select_for_update().get(
            code=conference_id
        )
        conference.confattend += 1
        conference.save()

Multi-Database Transactions

Each database connection has independent transactions:

from django.db import transaction

# Atomic across multiple databases
def create_account_and_conference(account_data, conf_data):
    with transaction.atomic(using="accounts"):
        account = Accounts.objects.using("accounts").create(**account_data)

    with transaction.atomic(using="TENANT_NAME"):
        conference = PlanionContainer.objects.using("TENANT_NAME").create(
            sponsor=account.acronym,
            **conf_data
        )

    return account, conference

Two-Phase Commit Not Supported

Django doesn't support distributed transactions across databases. If the second transaction fails, the first remains committed. Implement compensating transactions or idempotency to handle partial failures.

Transaction Isolation Levels

Configure isolation levels per database:

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "poseidon",
        "OPTIONS": {
            "isolation_level": "read committed",  # PostgreSQL
            "init_command": "SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED",  # MySQL
        },
    },
}

Isolation levels (from least to most strict):

  • Read uncommitted: Can see uncommitted changes (rarely used)
  • Read committed: Only sees committed changes (common default)
  • Repeatable read: Same query returns same results within transaction
  • Serializable: Full isolation, slowest performance

Read Replicas

Routing Reads to Replicas

Use naming conventions and explicit routing:

# Model with read replica preference
class Accounts(models.Model):
    # ... fields ...

    class Meta:
        managed = False
        db_table = "accounts"

    _DATABASE = "accounts"
    _PREFER_READ_REPLICA = True  # Custom attribute for router

# Explicit read from replica
accounts = Accounts.objects.using("accounts_RO").all()

# Write must go to primary
account = Accounts.objects.using("accounts").create(...)

Handling Replication Lag

Replication lag can cause read-after-write inconsistencies:

def create_and_display_account(account_data):
    # Write to primary
    account = Accounts.objects.using("accounts").create(**account_data)

    # Force read from primary to see our write
    # (Don't use replica - replication lag might hide new record)
    fresh_account = Accounts.objects.using("accounts").get(
        acronym=account.acronym
    )

    return fresh_account

Replication Lag Mitigation

  • Use primary database for critical read-after-write scenarios
  • Implement retry logic with exponential backoff
  • Monitor replication lag in production
  • Consider eventual consistency in UI design

Read-Write Splitting Pattern

class ReplicaRouter:
    """
    Router that splits reads and writes automatically.
    """

    def db_for_read(self, model, **hints):
        """Send reads to replicas by default."""
        database = getattr(model, "_DATABASE", None)

        if database:
            replica = f"{database}_RO"
            # Fall back to primary if replica doesn't exist
            if replica in settings.DATABASES:
                return replica
            return database

        return None

    def db_for_write(self, model, **hints):
        """All writes go to primary."""
        database = getattr(model, "_DATABASE", None)

        # Strip _RO suffix if present
        if database and database.endswith("_RO"):
            return database[:-3]

        return database

Testing with Multiple Databases

Test Database Configuration

# settings/testing_databases.py
from .development_databases import DATABASES as DEV_DATABASES

DATABASES = {}

DEFAULT_TEST_DB = "test_poseidon"
TENANT_TEST_DB = "test_testclient"

for alias, config in DEV_DATABASES.items():
    DATABASES[alias] = config.copy()

    if alias in ("TESTCLIENT", "TESTCLIENT_RO"):
        DATABASES[alias].setdefault("TEST", {})["NAME"] = TENANT_TEST_DB
    elif alias == "default":
        DATABASES[alias].setdefault("TEST", {})["NAME"] = DEFAULT_TEST_DB
    else:
        # All other databases mirror default to reduce overhead
        DATABASES[alias].setdefault("TEST", {})["MIRROR"] = "default"

    DATABASES[alias].setdefault("TEST", {})["SERIALIZE"] = False

Writing Multi-Database Tests

import pytest
from django.test import TestCase, TransactionTestCase

class MultiDatabaseTestCase(TransactionTestCase):
    """
    Base class for tests requiring multiple databases.

    Use TransactionTestCase (not TestCase) when testing multiple databases
    to ensure proper cleanup.
    """

    databases = {"default", "accounts", "TESTCLIENT"}  # Declare used databases

    def test_cross_database_operation(self):
        """Test operation spanning multiple databases."""
        # Create account in accounts database
        account = Accounts.objects.using("accounts").create(
            acronym="TEST",
            accountname="Test Account"
        )

        # Create conference in tenant database
        conference = PlanionContainer.objects.using("TESTCLIENT").create(
            code="TEST2025",
            name="Test Conference 2025",
            sponsor=account.acronym
        )

        # Verify both records exist
        assert Accounts.objects.using("accounts").filter(acronym="TEST").exists()
        assert PlanionContainer.objects.using("TESTCLIENT").filter(
            code="TEST2025"
        ).exists()

# Pytest fixture approach
@pytest.fixture
def test_account(db):
    """Fixture providing test account."""
    return Accounts.objects.using("accounts").create(
        acronym="PYTEST",
        accountname="PyTest Account"
    )

@pytest.mark.django_db(databases=["default", "accounts", "TESTCLIENT"])
def test_with_fixtures(test_account):
    """Test using pytest fixtures with multiple databases."""
    conference = PlanionContainer.objects.using("TESTCLIENT").create(
        code="PY2025",
        name="PyTest Conference",
        sponsor=test_account.acronym
    )

    assert conference.sponsor == test_account.acronym

TestCase vs TransactionTestCase

  • TestCase: Fast, uses transactions for cleanup, but only supports single database
  • TransactionTestCase: Slower, truncates tables for cleanup, supports multiple databases
  • Use TransactionTestCase or pytest with databases parameter for multi-database tests

Loading Fixtures in Tests

from django.core.management import call_command

class FixtureTestCase(TransactionTestCase):
    databases = {"default", "TESTCLIENT"}

    def setUp(self):
        """Load test fixtures before each test."""
        # Load fixtures into specific database
        call_command(
            "loaddata",
            "poseidon/fixtures/test/embed_demo_people.json",
            database="TESTCLIENT"
        )
        call_command(
            "loaddata",
            "poseidon/fixtures/test/embed_demo_cstmflds.json",
            database="TESTCLIENT"
        )

    def test_fixture_data(self):
        """Test using loaded fixture data."""
        person = People.objects.using("TESTCLIENT_RO").get(pid=123823)
        assert person.first_name == "Sarah"

Common Pitfalls

1. Forgetting to Specify Database

Problem: Query uses wrong database because router returns None

# Bad: Might query wrong database if router returns None
account = Accounts.objects.get(acronym="TEST")

# Good: Explicitly specify database
account = Accounts.objects.using("accounts").get(acronym="TEST")

Solution: Always use .using() for multi-database operations or ensure router handles all cases.

2. Cross-Database Relations

Problem: Django can't enforce foreign keys across databases

# Bad: Cross-database ForeignKey won't work
class Conference(models.Model):
    sponsor = models.ForeignKey(Accounts, on_delete=models.CASCADE)

    class Meta:
        db_table = "conferences"

    _DATABASE = "TENANT_NAME"  # Different database from Accounts!

# Good: Use CharField for cross-database references
class Conference(models.Model):
    sponsor_acronym = models.CharField(max_length=20)

    def get_sponsor(self):
        """Manual join across databases."""
        return Accounts.objects.using("accounts").get(
            acronym=self.sponsor_acronym
        )

Solution: Use character fields for cross-database references and implement manual joins.

3. Migration on Wrong Database

Problem: Migration runs on wrong database, creating unwanted tables

# Bad: No migration control
class DBRouter:
    def allow_migrate(self, db, app_label, model_name=None, **hints):
        return True  # Allows migration everywhere!

# Good: Explicit migration control
class DBRouter:
    def allow_migrate(self, db, app_label, model_name=None, **hints):
        # Only migrate poseidon app on default database
        if app_label == "poseidon":
            return db == "default"

        # Don't migrate legacy models
        if model_name in ("abstract", "abstractauthors"):
            return False

        # Don't migrate read replicas
        if db.endswith("_RO"):
            return False

        return None

Solution: Implement comprehensive allow_migrate logic in router.

4. Read-After-Write with Replicas

Problem: Reading from replica immediately after write doesn't see new data

# Bad: Replication lag causes inconsistency
account = Accounts.objects.using("accounts").create(acronym="NEW")
# Immediately read from replica - might not exist yet!
exists = Accounts.objects.using("accounts_RO").filter(acronym="NEW").exists()

# Good: Read from primary for consistency
account = Accounts.objects.using("accounts").create(acronym="NEW")
exists = Accounts.objects.using("accounts").filter(acronym="NEW").exists()

Solution: Use primary database for reads that must see recent writes.

5. Transaction Scope with Multiple Databases

Problem: Assuming transactions span databases

# Bad: Second transaction can commit even if first fails
with transaction.atomic():  # Only affects default database!
    Accounts.objects.using("accounts").create(...)
    PlanionContainer.objects.using("TENANT_NAME").create(...)
    # If second create fails, first remains committed

# Good: Separate atomic blocks per database
with transaction.atomic(using="accounts"):
    account = Accounts.objects.using("accounts").create(...)

try:
    with transaction.atomic(using="TENANT_NAME"):
        PlanionContainer.objects.using("TENANT_NAME").create(...)
except Exception:
    # Implement compensating transaction if needed
    account.delete()
    raise

Solution: Use separate atomic() blocks per database and handle partial failures.

6. Test Isolation with Multiple Databases

Problem: Tests fail due to insufficient database declarations

# Bad: TestCase doesn't know about multiple databases
class MyTest(TestCase):
    def test_cross_database(self):
        Accounts.objects.using("accounts").create(...)  # May fail!

# Good: Declare all used databases
class MyTest(TransactionTestCase):
    databases = {"default", "accounts", "TESTCLIENT"}

    def test_cross_database(self):
        Accounts.objects.using("accounts").create(...)  # Works!

Solution: Use TransactionTestCase with databases attribute, or pytest with @pytest.mark.django_db(databases=[...]).

7. Connection Pooling with CONN_MAX_AGE

Problem: Stale connections accumulate state between requests

# Bad: Long-lived connections without health checks
DATABASES = {
    "default": {
        # ...
        "CONN_MAX_AGE": 600,  # 10 minutes
        # No health checks!
    },
}

# Good: Enable connection health checks (Django 4.1+)
DATABASES = {
    "default": {
        # ...
        "CONN_MAX_AGE": 600,
        "CONN_HEALTH_CHECKS": True,  # Validate before reuse
    },
}

Solution: Always enable CONN_HEALTH_CHECKS when using CONN_MAX_AGE > 0.

8. Unmanaged Model Migrations

Problem: Django creates migrations for managed = False models

# Problem: Migration file created even though model is unmanaged
class PlanionContainer(models.Model):
    # ... fields ...

    class Meta:
        managed = False  # But Django still tracks changes!

# Solution: Prevent migrations in router
class DBRouter:
    def allow_migrate(self, db, app_label, model_name=None, **hints):
        # Skip migrations for specific models
        if model_name == "planioncontainer":
            return False
        return True

Solution: Use allow_migrate to prevent migration generation, or use --fake when running migrations.

Summary

Multi-database Django applications require careful attention to:

  1. Configuration: Separate development, production, and testing database configurations
  2. Routing: Implement robust database routers with clear allow_migrate logic
  3. Model Design: Use managed = False and _DATABASE attributes for legacy databases
  4. Query Patterns: Explicit .using() calls for cross-database operations
  5. Transactions: Separate atomic blocks per database, handle partial failures
  6. Read Replicas: Balance performance with consistency requirements
  7. Testing: Use TransactionTestCase and declare all databases
  8. Optimization: Leverage connection pooling, proper indexing, and query optimization

By following these patterns, you can build robust multi-tenant applications that scale effectively while maintaining data isolation and query performance.