Django多数据库同步与跨库访问问询：同步cl_db至default数据库

问题描述 #

1. 如何在Django框架中实现不同数据库与应用之间的同步？
2. 如何在Django应用的数据库中访问其他数据库的表？具体场景：default是Django主应用数据库，cl_db是ASP.NET应用的数据库（同服务器），需将cl_db的数据同步至default。

现有配置 #

DATABASES 配置 #

DATABASES = {
    'default': {
            'ENGINE': 'django.db.backends.postgresql_psycopg2', 
            'NAME': 'fail_over',                    
            'USER': 'SomeUser',                      
            'PASSWORD': 'SomePassword',                  
            'HOST': '127.0.0.1',                     
            'PORT': '',                      
    },

    'cl_db': {
            'ENGINE': 'django.db.backends.postgresql_psycopg2', 
            'NAME': 'cl_dev',                    
            'USER': 'SomeUser',                      
            'PASSWORD': 'SomePassword',                  
            'HOST': '127.0.0.1',                     
            'PORT': '',                      
    },
}

数据库路由配置 #

DATABASE_ROUTERS = ['app.dbrouters.AuthRounter']

AuthRouter 实现 #

class AuthRouter:
    """
    A router to control all database operations on models in the
    auth and contenttypes applications.
    """
    route_app_labels = {'auth', 'contenttypes'}

    def db_for_read(self, model, **hints):
        """
        Attempts to read auth and contenttypes models go to auth_db.
        """
        if model._meta.app_label in self.route_app_labels:
            return 'default'
        return None

    def db_for_write(self, model, **hints):
        """
        Attempts to write auth and contenttypes models go to auth_db.
        """
        if model._meta.app_label in self.route_app_labels:
            return 'default'
        return None

    def allow_relation(self, obj1, obj2, **hints):
        """
        Allow relations if a model in the auth or contenttypes apps is
        involved.
        """
        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):
        """
        Make sure the auth and contenttypes apps only appear in the
        'auth_db' database.
        """
        if app_label in self.route_app_labels:
            return db == 'default'
        return None

现有模型（models.py） #

class usermaster(models.Model):
  userid = models.AutoField(primary_key=True, unique=True)
  class Meta:
    app_label = 'user_master'

解决方案 #

一、访问其他数据库的表 #

要读取cl_db中的现有表，需创建未托管模型，让Django直接映射到外部数据库表，而非生成新表。

1. 创建外部表映射模型

假设cl_db中有一张名为cl_usermaster的表（需与实际表名一致），在models.py中添加：

class CLUserMaster(models.Model):
    userid = models.IntegerField(primary_key=True)
    # 按cl_db中实际表结构添加其他字段，如username、email等
    username = models.CharField(max_length=100)
    email = models.EmailField(null=True)

    class Meta:
        managed = False  # 关键：告知Django不管理该表的迁移/创建
        db_table = 'cl_usermaster'  # 对应cl_db中的实际表名
        app_label = 'user_master'

2. 配置数据库路由指向cl_db

新增专门处理跨库模型的路由类，或修改现有AuthRouter：

class CLDBRouter:
    route_models = {'user_master.CLUserMaster'}

    def db_for_read(self, model, **hints):
        if f"{model._meta.app_label}.{model.__name__}" in self.route_models:
            return 'cl_db'
        return None

    def db_for_write(self, model, **hints):
        # 不建议直接写入外部应用数据库，如需写入可调整返回值
        return None

    def allow_migrate(self, db, app_label, model_name=None, **hints):
        # 禁止该模型生成迁移文件
        if model_name.lower() == 'cl_usermaster':
            return False
        return None

更新DATABASE_ROUTERS配置，加入新路由：

DATABASE_ROUTERS = ['app.dbrouters.AuthRouter', 'app.dbrouters.CLDBRouter']

现在可通过ORM读取cl_db数据：

# 从cl_db读取全部用户数据
cl_users = CLUserMaster.objects.using('cl_db').all()

二、实现数据库数据同步 #

根据需求选择批量定时同步或实时同步方案：

方案1：批量同步脚本（适合定时执行）

编写Django自定义管理命令，实现数据批量同步：

1. 在应用目录下创建management/commands文件夹，添加sync_cl_db.py：

from django.core.management.base import BaseCommand
from django.db import transaction
from user_master.models import CLUserMaster, usermaster

class Command(BaseCommand):
    help = 'Sync data from cl_db to default database'

    def handle(self, *args, **options):
        # 开启事务确保数据一致性
        with transaction.atomic():
            cl_users = CLUserMaster.objects.using('cl_db').all()
            sync_count = 0

            for cl_user in cl_users:
                # 用update_or_create避免主键冲突
                _, created = usermaster.objects.update_or_create(
                    userid=cl_user.userid,
                    defaults={
                        # 如需同步其他字段，在这里添加映射关系
                    }
                )
                if created:
                    sync_count += 1

        self.stdout.write(self.style.SUCCESS(f"Sync completed: {sync_count} new records added, {len(cl_users)-sync_count} records updated"))

2. 执行同步命令：

python manage.py sync_cl_db

可配合Linux cron或Windows任务计划，实现定时自动同步。

方案2：实时同步（基于数据库触发器）

若需实时同步，可在PostgreSQL中创建触发器，当cl_db的表发生变更时自动同步到default：

-- 在cl_db中创建同步函数
CREATE OR REPLACE FUNCTION sync_to_default_usermaster()
RETURNS TRIGGER AS $$
BEGIN
    -- 插入/更新到default库的usermaster表
    INSERT INTO fail_over.public.usermaster(userid)
    VALUES(NEW.userid)
    ON CONFLICT(userid) DO UPDATE SET userid = NEW.userid;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- 为cl_usermaster表创建触发器
CREATE TRIGGER trigger_sync_usermaster
AFTER INSERT OR UPDATE ON cl_dev.public.cl_usermaster
FOR EACH ROW EXECUTE FUNCTION sync_to_default_usermaster();

注意：需确保数据库用户拥有跨库操作权限，该方案与Django解耦，依赖数据库原生功能。

三、通用多数据库同步注意事项 #

• 字段一致性：确保源表与目标表的字段类型、约束匹配，避免数据同步失败。
• 性能优化：批量同步时优先使用bulk_create/bulk_update，减少数据库交互次数。
• 冲突处理：明确主键重复时的处理逻辑，避免数据覆盖或丢失。
• 日志记录：同步过程中添加日志，便于排查异常。

内容的提问来源于stack exchange，提问作者Vas