在现有Django项目中实现读写更新请求的分库路由（无需改原有查询）

1. 编写自定义数据库路由类 #

在项目中创建db_router.py文件（可放在项目根目录或工具类app下），通过路由逻辑根据操作类型自动分配数据库：

class OperationBasedDBRouter:
    def db_for_read(self, model, **hints):
        # 所有读操作路由到read_db
        return 'read_db'

    def db_for_write(self, model, **hints):
        # 区分创建（新对象）和更新（已有对象）操作
        instance = hints.get('instance')
        if instance is not None and instance.pk is not None:
            # 已有对象的保存属于更新操作，路由到update_db
            return 'update_db'
        # 新对象创建属于写操作，路由到write_db
        return 'write_db'

    def allow_relation(self, obj1, obj2, **hints):
        # 允许同一数据库内的对象建立关联
        db_list = ('read_db', 'write_db', 'update_db')
        if obj1._state.db in db_list and obj2._state.db in db_list:
            return True
        return None

    def allow_migrate(self, db, app_label, model_name=None, **hints):
        # 迁移操作默认允许所有数据库（可根据实际需求调整）
        return True

2. 配置settings.py启用路由 #

在项目的settings.py中添加路由配置，指向刚创建的路由类：

# 路径根据实际文件位置调整，示例为项目根目录下的db_router.py
DATABASE_ROUTERS = ['your_project_name.db_router.OperationBasedDBRouter']

3. 验证效果 #

无需修改现有查询语句，操作会自动路由到对应数据库：

• 读操作MyModel.objects.get(name="abc")自动走read_db
• 创建操作MyModel.objects.create(name="xyz")自动走write_db
• 更新操作（如obj = MyModel.objects.get(name="abc"); obj.name="def"; obj.save()）自动走update_db

注意事项 #

• 需保持三个数据库的表结构一致，可通过Django迁移脚本同步
• 若有跨数据库关联需求，需调整allow_relation方法的逻辑
• 如需针对特定模型做特殊路由，可在路由方法中添加模型判断逻辑

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