Database Routing

Django-CFG provides automatic database routing based on app labels with zero manual .using() calls.

Automatic Routing

The DatabaseRouter automatically routes 100% of queries to the correct database based on app labels—no manual .using() calls needed!

How It Works

The DatabaseRouter automatically routes database operations based on DATABASE_ROUTING_RULES:

# settings.py
DATABASE_ROUTERS = ['django_cfg.routing.DatabaseRouter']

DATABASE_ROUTING_RULES = {
    'blog': 'blog_db',
    'shop': 'shop_db',
    # Apps not listed → default database
}

Router Implementation

The DatabaseRouter provides four key methods:

Read Operations

def db_for_read(self, model, **hints):
    """Route read operations to appropriate database."""
    rules = getattr(settings, 'DATABASE_ROUTING_RULES', {})
    app_label = model._meta.app_label

    return rules.get(app_label, 'default')

Use Case: Routes SELECT queries to the correct database based on model's app_label.

Write Operations

def db_for_write(self, model, **hints):
    """Route write operations to appropriate database."""
    rules = getattr(settings, 'DATABASE_ROUTING_RULES', {})
    app_label = model._meta.app_label

    return rules.get(app_label, 'default')

Use Case: Routes INSERT, UPDATE, DELETE operations to the correct database.

Relationships

def allow_relation(self, obj1, obj2, **hints):
    """
    Allow relations between objects.

    - Both routed to SAME db → allow
    - One or both NOT routed → allow (enables cross-db FK)
    - Both routed to DIFFERENT dbs → deny
    """
    rules = getattr(settings, 'DATABASE_ROUTING_RULES', {})
    db1 = rules.get(obj1._meta.app_label)
    db2 = rules.get(obj2._meta.app_label)

    # If both are routed, they must be in the same database
    if db1 and db2:
        return db1 == db2

    # If one or both are not routed, allow the relation
    # This enables cross-database ForeignKeys to shared models (e.g., User)
    return True

Use Case: Controls whether ForeignKey relationships are allowed between models in different databases.

Migrations

def allow_migrate(self, db, app_label, **hints):
    """Control which apps can migrate on which databases."""
    rules = getattr(settings, 'DATABASE_ROUTING_RULES', {})
    target_db = rules.get(app_label)

    if target_db:
        # App IS routed → allow only on target db
        return db == target_db

    # Check if this db is used by other apps
    if db in rules.values():
        # This db is used by other apps → don't migrate default apps here
        return False

    # Default apps → migrate on default database
    return db == 'default'

Use Case: Controls which migrations run on which databases. Prevents accidental migrations to wrong database.

Configuration

Setup

Advanced Feature

Most applications don't need database routing. Use this only if you have:

• Multiple physical databases for different modules
• Read replicas for performance
• Separate analytics/reporting database

# api/config.py - ONLY use if you need multiple databases!
from django_cfg import DjangoConfig, DatabaseConfig
from .environment import env

class MyDjangoConfig(DjangoConfig):
    # Single database (default for most SaaS apps)
    databases = {
        "default": DatabaseConfig.from_url(url=env.database.url),
    }

# Generate Django settings
config = MyDjangoConfig()

For advanced multi-database setup (rare), see Multi-Database Guide.

Environment Configuration

# .env or system environment variables
DATABASE__URL="postgresql://localhost/mydb"

# api/environment/loader.py
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict

class DatabaseConfig(BaseSettings):
    url: str = Field(default="sqlite:///db/default.sqlite3")

    model_config = SettingsConfigDict(
        env_prefix="DATABASE__",
        env_nested_delimiter="__",
    )

Single Database for SaaS

Most SaaS applications use a single database with multiple apps/modules. Database routing is typically only needed for:

• Read replicas (performance)
• Analytics warehouses (separate reporting DB)
• Legacy system integration

---

Advanced: Read Replica Setup

For performance scaling with read replicas:

Configuration

# api/environment/loader.py
class DatabaseConfig(BaseSettings):
    url: str = Field(default="sqlite:///db/default.sqlite3")
    url_replica: str | None = Field(default=None)

    model_config = SettingsConfigDict(
        env_prefix="DATABASE__",
        env_nested_delimiter="__",
    )

# Production ENV
DATABASE__URL="postgresql://app:pass@db-primary.example.com/mydb"
DATABASE__URL_REPLICA="postgresql://readonly:pass@db-replica.example.com/mydb"

# api/config.py
class MyDjangoConfig(DjangoConfig):
    databases = {
        "default": DatabaseConfig.from_url(url=env.database.url),
    }

    if env.database.url_replica:
        databases["replica"] = DatabaseConfig.from_url(url=env.database.url_replica)
        database_routers = ["myapp.routers.ReadReplicaRouter"]

Custom Router

# myapp/routers.py
import random

class ReadReplicaRouter:
    """Route 80% of reads to replica, all writes to primary."""

    def db_for_read(self, model, **hints):
        """Most reads go to replica."""
        return 'replica' if random.random() < 0.8 else 'default'

    def db_for_write(self, model, **hints):
        """All writes to primary."""
        return 'default'

    def allow_relation(self, obj1, obj2, **hints):
        """Allow relations between same database."""
        return True

    def allow_migrate(self, db, app_label, **hints):
        """Only migrate on primary."""
        return db == 'default'

Usage Examples

from apps.profiles.models import User

# Writes always go to primary
user = User.objects.create(email="user@example.com")
# → Routed to 'default' (primary)

# Reads mostly go to replica (80%)
users = User.objects.all()
# → Routed to 'replica' (80%) or 'default' (20%)

# Force specific database when needed
fresh_user = User.objects.using('default').get(id=user.id)
# → Always reads from primary (latest data)

heavy_report = User.objects.using('analytics').aggregate(
    total=Count('*'),
    revenue=Sum('purchases__amount')
)
# → Use analytics warehouse for heavy queries

Manual Database Selection

# Force read from primary (when you need latest data)
latest_user = User.objects.using('default').get(id=123)

# Force read from replica (when eventual consistency is ok)
cached_users = User.objects.using('replica').filter(is_active=True)

# Use analytics database for reports
from django.db import connections

with connections['analytics'].cursor() as cursor:
    cursor.execute("SELECT * FROM user_stats WHERE created_at > NOW() - INTERVAL '30 days'")
    results = cursor.fetchall()

Routing Flow - Read Replica Pattern

Read Operation

Flow Steps:

1. User calls: User.objects.all()
2. Django calls: router.db_for_read(User)
3. Router randomizes: 80% replica, 20% primary
4. Django executes query on selected database

Write Operation

Flow Steps:

1. User calls: User.objects.create(...)
2. Django calls: router.db_for_write(User)
3. Router always returns: 'default' (primary)
4. Django executes INSERT on primary
5. Data replicates to replica asynchronously

Migration

Flow Steps:

1. User runs: python manage.py migrate
2. Django calls: router.allow_migrate(db='default', app_label='users')
3. Router checks: db == 'default' → True
4. Django migrates: only on primary
5. Replica syncs schema via replication

Debugging Routing

Check Which Database Will Be Used

from django.db import router
from apps.blog.models import Post

# For reading
db = router.db_for_read(Post)
print(f"Post reads from: {db}")  # blog_db

# For writing
db = router.db_for_write(Post)
print(f"Post writes to: {db}")  # blog_db

# Check if migration is allowed
allowed = router.allow_migrate('blog_db', 'blog')
print(f"Migrate blog on blog_db: {allowed}")  # True

View Active Router

from django.conf import settings

print("Active routers:", settings.DATABASE_ROUTERS)
print("Routing rules:", settings.DATABASE_ROUTING_RULES)

Test Routing

# Test automatic routing
post = Post.objects.create(title="Test")
print(post._state.db)  # blog_db

user = User.objects.create_user(email="test@example.com")
print(user._state.db)  # default

Advanced Patterns

📚 Read Replicas - Scale Read Operations

Read Replicas Setup

# settings.py
DATABASE_ROUTING_RULES = {
    'blog': 'blog_db',
    'blog_replica': 'blog_replica_db',
}

Custom Router for Load Distribution

class ReplicaRouter:
    def db_for_read(self, model, **hints):
        if model._meta.app_label == 'blog':
            # 80% of reads go to replica
            import random
            return 'blog_replica_db' if random.random() < 0.8 else 'blog_db'
        return None  # Fall through to next router

    def db_for_write(self, model, **hints):
        if model._meta.app_label == 'blog':
            return 'blog_db'  # All writes to primary
        return None

Enable Multi-Router Chain

# settings.py
DATABASE_ROUTERS = [
    'myapp.routers.ReplicaRouter',  # Custom read replica logic
    'django_cfg.routing.DatabaseRouter',  # Fallback routing
]

Performance Gain

Read replicas can handle 80% of database load, reducing primary database pressure and improving response times.

🔀 Sharding by User - Horizontal Scaling

User-Based Sharding

def get_user_shard(user_id):
    """Determine database shard for user."""
    shard_count = 4
    shard_id = user_id % shard_count
    return f"shard_{shard_id}"

# Use in queries
shard = get_user_shard(user.id)
user_posts = Post.objects.using(shard).filter(author=user)

Advanced Sharding Router

class ShardingRouter:
    def db_for_read(self, model, **hints):
        if 'user_id' in hints:
            return get_user_shard(hints['user_id'])
        return None

    def db_for_write(self, model, **hints):
        if 'user_id' in hints:
            return get_user_shard(hints['user_id'])
        return None

Complexity Trade-off

Sharding adds complexity. Consider read replicas first—only shard when you exceed 100M+ records or 10K+ writes/sec.

Best Practices

1. Use Automatic Routing

# ✅ Good: Automatic routing
post = Post.objects.create(title="My Post")

# ❌ Bad: Manual routing (unless necessary)
post = Post.objects.using('blog_db').create(title="My Post")

2. Avoid Cross-Database JOINs

# ✅ Good: Query from the correct side
orders = Order.objects.filter(user_id=user.id)

# ❌ Bad: Cross-database relationship traversal
orders = user.order_set.all()  # May not work across databases

3. Document Routing Rules

# settings.py
DATABASE_ROUTING_RULES = {
    # Blog app → blog_db (PostgreSQL on db-blog.example.com)
    'blog': 'blog_db',

    # Shop app → shop_db (PostgreSQL on db-shop.example.com)
    'shop': 'shop_db',

    # Analytics app → analytics (PostgreSQL read-only warehouse)
    'analytics': 'analytics',
}

Troubleshooting

Common Issue: Routing Not Working

Symptom: Queries still go to default database

Solution:

# 1. Check if router is configured
from django.conf import settings
print(settings.DATABASE_ROUTERS)
# Should show: ['django_cfg.routing.DatabaseRouter']

# 2. Check routing rules exist
print(settings.DATABASE_ROUTING_RULES)
# Should show: {'blog': 'blog_db', ...}

# 3. Verify app_label matches
from apps.blog.models import Post
print(Post._meta.app_label)  # Should be 'blog', not 'apps.blog'

Wrong Database Being Used

Symptom: Model saves to unexpected database

Debug Steps:

# 1. Check which database is actually being used
from django.db import connections
print(connections['blog_db'].queries)  # Show recent queries

# 2. Inspect model after save
post = Post.objects.create(title="Test")
print(post._state.db)  # Should show 'blog_db'

# 3. Test router directly
from django.db import router
db = router.db_for_write(Post)
print(f"Router selected: {db}")  # Should be 'blog_db'

Migration Issues - Quick Fix

Symptom: Migrations applied to wrong database

Solution:

# 1. Check migration status for all databases
python manage.py showmigrations

# 2. Migrate specific database
python manage.py migrate --database=blog_db

# 3. Use migrate_all for all databases (Django-CFG helper)
python manage.py migrate_all

# 4. Rollback if needed
python manage.py migrate blog 0003 --database=blog_db

🔍 Advanced Debugging Techniques

Enable Query Logging

# settings.py
LOGGING = {
    'version': 1,
    'handlers': {
        'console': {'class': 'logging.StreamHandler'},
    },
    'loggers': {
        'django.db.backends': {
            'handlers': ['console'],
            'level': 'DEBUG',  # Shows all SQL queries with database alias
        },
    },
}

Test Routing Programmatically

# test_routing.py
from django.test import TestCase
from django.db import router
from apps.blog.models import Post

class RoutingTestCase(TestCase):
    def test_read_routing(self):
        db = router.db_for_read(Post)
        self.assertEqual(db, 'blog_db')

    def test_write_routing(self):
        db = router.db_for_write(Post)
        self.assertEqual(db, 'blog_db')

    def test_actual_database_used(self):
        post = Post.objects.create(title="Test")
        self.assertEqual(post._state.db, 'blog_db')

See Also

• Multi-Database - Multi-database setup
• Migrations - Migration commands
• Cross-Database Relations - ForeignKeys across databases
• Sample Project - Complete example