如何在Django中跨多个数据库使用select_related？

问题重现 #

数据库配置（settings.py） #

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql', 
        'NAME': 'mysql_db_name',
        'USER': 'mysql_user',
        'PASSWORD': 'mysql_db_pass',
        'HOST': 'localhost',
        'PORT': '3306',
    },
    'postgres_db': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'postgresql_db_name',
        'USER': 'postgresql_user',
        'PASSWORD': 'postgresql_db_pass',
        'HOST': 'localhost',
        'PORT': '5432',
    }
}

模型定义（sales/models.py） #

from django.contrib.auth.models import User  # 存储在MySQL默认库
from django.db import models

# 自定义跨库外键
class ForeignKeyAcrossDb(models.ForeignKey):
    pass

class Batch(models.Model):     # 存储在PostgreSQL的postgres_db库
    class params:
        db = 'postgres_db'
    created_by = ForeignKeyAcrossDb(User, db_constraint=False, on_delete=models.DO_NOTHING)

报错情况 #

执行不带select_related()的查询可正常运行：

data = Batch.objects.filter(created_by=request.user.pk)

但执行带select_related()的查询时触发报错：

data = Batch.objects.select_related('created_by').filter(created_by=request.user.pk)

报错信息：

ProgrammingError at /
relation "auth_user" does not exist
LINE 1: ...r"."date_joined" FROM "sales_batch" INNER JOIN "auth_user...

原因：select_related()会生成数据库层面的JOIN语句，但auth_user表在MySQL中，而Batch查询是在PostgreSQL中执行，跨数据库的JOIN无法被数据库原生支持。

解决方案 #

由于数据库本身不支持跨库JOIN，Django的select_related()无法直接用于跨数据库关联，推荐以下两种替代方案：

1. 手动批量查询关联对象（推荐） #

通过Python层面完成关联，避免数据库JOIN，同时用in_bulk避免N+1查询问题：

# 1. 查询目标Batch数据（从PostgreSQL）
batches = list(Batch.objects.filter(created_by=request.user.pk))

# 2. 提取所有关联的User ID
user_ids = [batch.created_by_id for batch in batches]

# 3. 批量查询User对象（从MySQL默认库）
users = User.objects.in_bulk(user_ids)

# 4. 手动将User关联到对应的Batch实例
for batch in batches:
    batch.created_by = users.get(batch.created_by_id)

2. 封装通用的跨库关联工具函数 #

如果需要在多个场景复用，可以封装一个类似select_related的工具函数：

def cross_db_select_related(queryset, related_field_name):
    # 将查询集转为列表，触发数据库查询
    queryset_results = list(queryset)
    if not queryset_results:
        return queryset_results
    
    # 获取关联字段的信息
    related_field = queryset.model._meta.get_field(related_field_name)
    related_model = related_field.related_model
    # 提取所有关联ID
    related_ids = [getattr(obj, related_field.attname) for obj in queryset_results]
    
    # 批量查询关联模型的数据
    related_objects = related_model.objects.in_bulk(related_ids)
    
    # 手动关联到主模型实例
    for obj in queryset_results:
        setattr(obj, related_field_name, related_objects.get(getattr(obj, related_field.attname)))
    
    return queryset_results

# 使用示例
batches = cross_db_select_related(Batch.objects.filter(created_by=request.user.pk), 'created_by')

注意事项 #

• 跨数据库关联无法依赖数据库层面的外键约束，必须确保数据一致性（比如User被删除时，Batch的created_by_id需要手动处理）；
• 上述方案的性能接近select_related()，仅需两次数据库查询，避免了N+1问题；
• 不要尝试强制让Django生成跨库JOIN语句，这超出了数据库的原生能力范围。

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