如何在Django ORM中跨两个不同数据库执行子查询

问题原因 #

当在主查询（Model2，对应db2）中嵌套子查询（Model1，对应db1）时，Django ORM默认会使用主查询的数据库连接执行整个SQL语句，导致子查询中的Model1被解析为db2中的表，而db2中不存在该表，从而抛出错误。此外你代码中存在语法错误：model2_qs.objects.using('db1')，model2_qs已是QuerySet实例，不能再调用.objects属性。

解决方案 #

通过自定义CrossDBSubquery类，重写子查询的SQL生成逻辑，为跨库表添加数据库名前缀，让ORM生成正确的跨库查询SQL。该方案全程使用ORM语法，无需编写完整原生SQL。

自定义跨库子查询类 #

from django.db.models import Subquery
from django.db import connections

class CrossDBSubquery(Subquery):
    def __init__(self, queryset, using, **kwargs):
        self.target_db = connections[using].settings_dict['NAME']
        self.table_name = queryset.model._meta.db_table
        super().__init__(queryset, **kwargs)
    
    def as_sql(self, compiler, connection, template=None, **kwargs):
        # 生成子查询SQL并替换表名为带数据库前缀的完整名称
        sql, params = super().as_sql(compiler, connection, template, **kwargs)
        # 根据数据库类型调整引号（示例适配MySQL，PostgreSQL可替换为""）
        quoted_table = connection.ops.quote_name(self.table_name)
        quoted_full_table = f"{connection.ops.quote_name(self.target_db)}.{quoted_table}"
        sql = sql.replace(quoted_table, quoted_full_table)
        return sql, params

修正后的查询代码 #

from django.db.models import OuterRef
from app1.models import Model1
from app2.models import Model2

# 构建Model1的子查询
model1_subquery = Model1.objects.filter(id=OuterRef('context_id')).values('value')[:1]

# 主查询使用Model2的db2连接，通过自定义类注入跨库子查询
res = Model2.objects.using('db2').annotate(
    model1_value=CrossDBSubquery(model1_subquery, using='db1')
)

注意事项 #

• 确保你的数据库支持跨库查询（如MySQL、PostgreSQL均支持同一实例下的跨库查询），且数据库用户拥有访问两个库的权限。
• 若使用非MySQL数据库，需调整as_sql方法中的引号规则（比如PostgreSQL使用双引号包裹表名和库名）。

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