Bài 29: Working with Multiple Databases

DATABASES Configuration

Django hỗ trợ sử dụng nhiều databases trong một project.

Basic Configuration

# settings.pyDATABASES = {    'default': {        'ENGINE': 'django.db.backends.postgresql',        'NAME': 'main_db',        'USER': 'postgres',        'PASSWORD': 'password',        'HOST': 'localhost',        'PORT': '5432',    },    'users_db': {        'ENGINE': 'django.db.backends.postgresql',        'NAME': 'users_db',        'USER': 'postgres',        'PASSWORD': 'password',        'HOST': 'localhost',        'PORT': '5432',    },    'analytics_db': {        'ENGINE': 'django.db.backends.mysql',        'NAME': 'analytics',        'USER': 'root',        'PASSWORD': 'password',        'HOST': 'localhost',        'PORT': '3306',        'OPTIONS': {            'charset': 'utf8mb4',        },    },    'legacy_db': {        'ENGINE': 'django.db.backends.sqlite3',        'NAME': BASE_DIR / 'legacy.sqlite3',    },}

Read Replica Configuration

# settings.py - Primary with read replicasDATABASES = {    'default': {        'ENGINE': 'django.db.backends.postgresql',        'NAME': 'main_db',        'USER': 'postgres',        'PASSWORD': 'password',        'HOST': 'primary.example.com',        'PORT': '5432',    },    'replica1': {        'ENGINE': 'django.db.backends.postgresql',        'NAME': 'main_db',        'USER': 'postgres',        'PASSWORD': 'password',        'HOST': 'replica1.example.com',        'PORT': '5432',    },    'replica2': {        'ENGINE': 'django.db.backends.postgresql',        'NAME': 'main_db',        'USER': 'postgres',        'PASSWORD': 'password',        'HOST': 'replica2.example.com',        'PORT': '5432',    },}

Connection Pooling

# settings.py - Connection pooling with django-db-poolDATABASES = {    'default': {        'ENGINE': 'django_db_pool.backends.postgresql',        'NAME': 'main_db',        'USER': 'postgres',        'PASSWORD': 'password',        'HOST': 'localhost',        'PORT': '5432',        'POOL_OPTIONS': {            'POOL_SIZE': 10,            'MAX_OVERFLOW': 20,        }    }} # Using pgbouncer (recommended for production):DATABASES = {    'default': {        'ENGINE': 'django.db.backends.postgresql',        'NAME': 'main_db',        'USER': 'postgres',        'PASSWORD': 'password',        'HOST': 'localhost',        'PORT': '6432',  # pgbouncer port        'OPTIONS': {            'connect_timeout': 10,        }    }}

Database Routing

Basic Database Router

# routers.pyclass AuthRouter:    """Route auth models to users_db"""        route_app_labels = {'auth', 'contenttypes', 'sessions'}        def db_for_read(self, model, **hints):        """Read from users_db"""        if model._meta.app_label in self.route_app_labels:            return 'users_db'        return None        def db_for_write(self, model, **hints):        """Write to users_db"""        if model._meta.app_label in self.route_app_labels:            return 'users_db'        return None        def allow_relation(self, obj1, obj2, **hints):        """Allow relations if both in same db"""        if (obj1._meta.app_label in self.route_app_labels or            obj2._meta.app_label in self.route_app_labels):            return True        return None        def allow_migrate(self, db, app_label, model_name=None, **hints):        """Ensure auth models only in users_db"""        if app_label in self.route_app_labels:            return db == 'users_db'        return None # settings.pyDATABASE_ROUTERS = ['myproject.routers.AuthRouter']

Advanced Routing

# routers.pyclass PrimaryReplicaRouter:    """Route reads to replicas, writes to primary"""        def db_for_read(self, model, **hints):        """Distribute reads across replicas"""        import random        return random.choice(['replica1', 'replica2'])        def db_for_write(self, model, **hints):        """All writes to primary"""        return 'default'        def allow_relation(self, obj1, obj2, **hints):        """Allow relations within same database"""        db_set = {'default', 'replica1', 'replica2'}        if obj1._state.db in db_set and obj2._state.db in db_set:            return True        return None        def allow_migrate(self, db, app_label, model_name=None, **hints):        """Only migrate on primary"""        return db == 'default' class AnalyticsRouter:    """Route analytics models to analytics_db"""        def db_for_read(self, model, **hints):        if model._meta.app_label == 'analytics':            return 'analytics_db'        return None        def db_for_write(self, model, **hints):        if model._meta.app_label == 'analytics':            return 'analytics_db'        return None        def allow_relation(self, obj1, obj2, **hints):        if (obj1._meta.app_label == 'analytics' or            obj2._meta.app_label == 'analytics'):            return obj1._meta.app_label == obj2._meta.app_label        return None        def allow_migrate(self, db, app_label, model_name=None, **hints):        if app_label == 'analytics':            return db == 'analytics_db'        return None # settings.pyDATABASE_ROUTERS = [    'myproject.routers.PrimaryReplicaRouter',    'myproject.routers.AuthRouter',    'myproject.routers.AnalyticsRouter',]

Model-Specific Routing

# routers.pyclass ModelRouter:    """Route specific models to specific databases"""        model_db_map = {        'blog.Post': 'default',        'blog.Comment': 'default',        'analytics.PageView': 'analytics_db',        'analytics.UserActivity': 'analytics_db',        'logs.ErrorLog': 'logs_db',        'logs.AccessLog': 'logs_db',    }        def db_for_read(self, model, **hints):        key = f'{model._meta.app_label}.{model._meta.model_name}'        return self.model_db_map.get(key)        def db_for_write(self, model, **hints):        key = f'{model._meta.app_label}.{model._meta.model_name}'        return self.model_db_map.get(key)        def allow_relation(self, obj1, obj2, **hints):        db1 = self.db_for_read(obj1.__class__)        db2 = self.db_for_read(obj2.__class__)        if db1 and db2:            return db1 == db2        return None        def allow_migrate(self, db, app_label, model_name=None, **hints):        if model_name:            key = f'{app_label}.{model_name}'            target_db = self.model_db_map.get(key)            if target_db:                return db == target_db        return None

Multi-Database Queries

Explicit Database Selection

# Query specific databasefrom blog.models import Post # Read from default databaseposts = Post.objects.all() # Read from specific databaseposts = Post.objects.using('replica1').all() # Write to specific databasepost = Post(title='Test', content='Content')post.save(using='default') # Update in specific databasePost.objects.using('default').filter(id=1).update(title='Updated') # Delete from specific databasePost.objects.using('default').filter(id=1).delete()

Cross-Database Queries

# models.pyclass Post(models.Model):    title = models.CharField(max_length=200)    author = models.ForeignKey(User, on_delete=models.CASCADE)        class Meta:        app_label = 'blog' class PageView(models.Model):    post_id = models.IntegerField()  # Not ForeignKey - different DB    user_id = models.IntegerField()    viewed_at = models.DateTimeField(auto_now_add=True)        class Meta:        app_label = 'analytics' # queries.pydef get_post_with_views(post_id):    """Get post from default DB and views from analytics DB"""    post = Post.objects.get(id=post_id)    views = PageView.objects.using('analytics_db').filter(        post_id=post_id    ).count()        return {        'post': post,        'views': views,    } # Can't do direct join across databases:# This will FAIL:# Post.objects.select_related('pageview_set')  # ERROR! # Instead, fetch separately and combine in Python:def get_popular_posts():    """Get posts with view counts"""    # Get view counts from analytics DB    view_counts = PageView.objects.using('analytics_db').values(        'post_id'    ).annotate(        count=Count('id')    ).order_by('-count')[:10]        # Get post IDs    post_ids = [item['post_id'] for item in view_counts]        # Get posts from default DB    posts = Post.objects.filter(id__in=post_ids)        # Combine results    view_dict = {item['post_id']: item['count'] for item in view_counts}        results = []    for post in posts:        results.append({            'post': post,            'views': view_dict.get(post.id, 0),        })        return sorted(results, key=lambda x: x['views'], reverse=True)

Transactions Across Databases

from django.db import transaction # Transaction in single database@transaction.atomicdef create_post(title, content):    post = Post.objects.create(title=title, content=content)    return post # Transaction in specific database@transaction.atomic(using='default')def create_post_primary(title, content):    post = Post.objects.using('default').create(        title=title,        content=content    )    return post # Multiple transactions (different databases)def create_post_with_analytics(title, content, user_id):    """Create post and log in analytics - separate transactions"""        # Transaction 1: Create post in default DB    with transaction.atomic(using='default'):        post = Post.objects.create(title=title, content=content)        # Transaction 2: Log in analytics DB    with transaction.atomic(using='analytics_db'):        PageView.objects.using('analytics_db').create(            post_id=post.id,            user_id=user_id        )        return post # WARNING: No distributed transactions!# If second transaction fails, first is already committed!def unsafe_multi_db_operation():    """This is NOT atomic across databases"""    try:        with transaction.atomic(using='default'):            post = Post.objects.create(title='Test')                        # If this fails, post is still created!            with transaction.atomic(using='analytics_db'):                PageView.objects.using('analytics_db').create(                    post_id=post.id                )    except Exception as e:        # post already exists in default DB!        pass # Better approach with manual rollback:def safer_multi_db_operation():    """Manual cleanup on failure"""    post = None    try:        with transaction.atomic(using='default'):            post = Post.objects.create(title='Test')                try:            with transaction.atomic(using='analytics_db'):                PageView.objects.using('analytics_db').create(                    post_id=post.id                )        except Exception as e:            # Rollback first operation            if post:                post.delete(using='default')            raise    except Exception as e:        # Handle error        pass

Raw SQL with Multiple Databases

from django.db import connections def execute_raw_query(db_alias='default'):    """Execute raw SQL on specific database"""    with connections[db_alias].cursor() as cursor:        cursor.execute("SELECT * FROM blog_post WHERE published = true")        rows = cursor.fetchall()        return rows def get_database_stats():    """Get stats from multiple databases"""    stats = {}        # Default database    with connections['default'].cursor() as cursor:        cursor.execute("SELECT COUNT(*) FROM blog_post")        stats['total_posts'] = cursor.fetchone()[0]        # Analytics database    with connections['analytics_db'].cursor() as cursor:        cursor.execute("SELECT COUNT(*) FROM analytics_pageview")        stats['total_views'] = cursor.fetchone()[0]        return stats

Database Migrations

Running Migrations

# Migrate all databasespython manage.py migrate # Migrate specific databasepython manage.py migrate --database=users_dbpython manage.py migrate --database=analytics_db # Show migration statuspython manage.py showmigrations --database=users_db # Fake migration (mark as applied without running)python manage.py migrate --fake --database=analytics_db

Creating Migrations

# Create migration for specific apppython manage.py makemigrations blog # The migration will be applied to databases based on router rules # routers.py - Control which migrations run whereclass AnalyticsRouter:    def allow_migrate(self, db, app_label, model_name=None, **hints):        """Only migrate analytics app on analytics_db"""        if app_label == 'analytics':            return db == 'analytics_db'                # Don't migrate analytics on other databases        if db == 'analytics_db':            return app_label == 'analytics'                return None

Migration Dependencies

# blog/migrations/0002_add_fields.pyfrom django.db import migrations, models class Migration(migrations.Migration):        dependencies = [        ('blog', '0001_initial'),        # Can reference migrations from other apps        ('auth', '0012_alter_user_first_name_max_length'),    ]        operations = [        migrations.AddField(            model_name='post',            name='featured',            field=models.BooleanField(default=False),        ),    ] # But be careful with cross-database dependencies!# If auth is in users_db and blog is in default,# this dependency might cause issues

Data Migration Across Databases

# blog/migrations/0003_copy_data.pyfrom django.db import migrations def copy_old_posts(apps, schema_editor):    """Copy posts from legacy database"""    Post = apps.get_model('blog', 'Post')        # Get from legacy DB    old_posts = Post.objects.using('legacy_db').all()        # Save to default DB    for old_post in old_posts:        Post.objects.using('default').create(            title=old_post.title,            content=old_post.content,            created=old_post.created,        ) def reverse_copy(apps, schema_editor):    """Reverse migration"""    Post = apps.get_model('blog', 'Post')    Post.objects.using('default').all().delete() class Migration(migrations.Migration):        dependencies = [        ('blog', '0002_add_fields'),    ]        operations = [        migrations.RunPython(copy_old_posts, reverse_copy),    ]

Database Sharding

Simple Sharding by User ID

# routers.pyclass UserShardRouter:    """Shard users across multiple databases by ID"""        def db_for_read(self, model, **hints):        if model._meta.app_label == 'accounts':            if 'instance' in hints:                user_id = hints['instance'].id                return self._get_shard(user_id)        return None        def db_for_write(self, model, **hints):        if model._meta.app_label == 'accounts':            if 'instance' in hints:                user_id = hints['instance'].id                return self._get_shard(user_id)        return None        def _get_shard(self, user_id):        """Determine shard by user ID"""        shard_count = 4        shard_id = user_id % shard_count        return f'shard_{shard_id}' # settings.pyDATABASES = {    'default': {'ENGINE': '...', 'NAME': 'main'},    'shard_0': {'ENGINE': '...', 'NAME': 'users_shard_0'},    'shard_1': {'ENGINE': '...', 'NAME': 'users_shard_1'},    'shard_2': {'ENGINE': '...', 'NAME': 'users_shard_2'},    'shard_3': {'ENGINE': '...', 'NAME': 'users_shard_3'},}

Hash-Based Sharding

# routers.pyimport hashlib class HashShardRouter:    """Shard by hash of key field"""        shard_count = 4        def db_for_read(self, model, **hints):        if model._meta.app_label == 'data':            if 'shard_key' in hints:                return self._get_shard(hints['shard_key'])        return None        def db_for_write(self, model, **hints):        if model._meta.app_label == 'data':            if 'shard_key' in hints:                return self._get_shard(hints['shard_key'])        return None        def _get_shard(self, key):        """Hash-based sharding"""        hash_value = int(hashlib.md5(str(key).encode()).hexdigest(), 16)        shard_id = hash_value % self.shard_count        return f'data_shard_{shard_id}' # Usagefrom data.models import Record # Specify shard key in hintsRecord.objects.using(    HashShardRouter()._get_shard('user@example.com')).filter(email='user@example.com')

Bài Tập

Bài 1: Multi-Database Setup

Task: Configure and use multiple databases:

# Requirements:# 1. Configure 3 databases: default (PostgreSQL), users (PostgreSQL), analytics (MySQL)# 2. Create AuthRouter to route auth models to users database# 3. Create AnalyticsRouter to route analytics models to analytics database# 4. Create User model in users database# 5. Create PageView model in analytics database# 6. Create Post model in default database# 7. Write queries using explicit database selection# 8. Run migrations on all databases# 9. Create management command to show database stats# 10. Test all database operations

Bài 2: Read Replicas

Task: Implement primary-replica setup:

# Requirements:# 1. Configure 1 primary + 2 replica databases# 2. Create router to send reads to replicas# 3. Create router to send writes to primary# 4. Implement read/write splitting# 5. Add replica load balancing (round-robin or random)# 6. Handle replica lag (force primary read when needed)# 7. Create decorator for forcing primary reads# 8. Monitor query distribution# 9. Test failover scenarios# 10. Document replication setup

Bài 3: Database Sharding

Task: Implement user sharding:

# Requirements:# 1. Configure 4 shard databases# 2. Create UserShardRouter (shard by user ID)# 3. Distribute users across shards evenly# 4. Query users from correct shard# 5. Handle user creation (assign to shard)# 6. Create management command to rebalance shards# 7. Implement shard lookup cache# 8. Handle user migration between shards# 9. Monitor shard usage# 10. Test shard failure handling

Bài 4: Complete Multi-Database System

Task: Full multi-database architecture:

# Requirements:# 1. 5+ databases (primary, replicas, analytics, logs, cache)# 2. Complete routing system# 3. Read/write splitting# 4. Sharding implementation# 5. Cross-database queries# 6. Transaction management# 7. Migration strategy# 8. Backup/restore procedures# 9. Monitoring and alerting# 10. Performance optimization# 11. Failover handling# 12. Data synchronization# 13. Audit logging# 14. Documentation# 15. Load testing

Tài Liệu Tham Khảo

Previous: Bài 28: Template Advanced Features | Next: Bài 30: Deployment Basics

#database-routing

Django Multiple Databases