如何在Django中为指定类/方法覆盖内部读查询以使用只读副本

问题背景 #

需要实现仅对指定类或单个方法生效的逻辑，让其内部所有Django读查询自动切换到只读副本数据库，不用逐个修改查询语句。试过装饰器、中间件、元类，但都碰了壁——没法直接访问函数内部的读查询。

举个例子，原来的查询是这样：

master_models.CarTransmission.objects.filter(  # pylint: disable=E1101
    variant_id__in=self.variant_ids
).values_list(*self.specifications, named=True)

要改成：

master_models.CarTransmission.objects.using('replica').filter(  # pylint: disable=E1101
    variant_id__in=self.variant_ids
).values_list(*self.specifications, named=True)

还有get_object_or_404的场景：
原代码：

model = get_object_or_404(master_models.Model, slug=slug)

修改后：

model = get_object_or_404(master_models.Model.objects.using('replica'), slug=slug)

要求很明确：只作用于指定类（类下所有方法都生效）或单个方法，拒绝全局/应用级的路由方案。

可行方案：装饰器+上下文管理器（局部替换模型管理器） #

这个方案能精准控制生效范围，不用动内部查询代码，完全符合需求。

1. 先写个上下文管理器，临时换模型的默认管理器 #

这个上下文管理器的作用是：在代码块执行期间，把指定模型的默认管理器换成指向只读副本的版本，执行完再换回去，不影响其他代码。

from contextlib import contextmanager

@contextmanager
def use_replica_for_models(*models):
    original_managers = {}
    try:
        # 先存下原来的管理器，再替换成replica版
        for model in models:
            original_managers[model] = model._default_manager
            model._default_manager = model.objects.using('replica')
        yield
    finally:
        # 不管执行成功还是失败，都把原管理器恢复回去
        for model, manager in original_managers.items():
            model._default_manager = manager

2. 方法级装饰器：给单个方法加开关 #

把上面的上下文管理器包装成装饰器，想用在哪个方法上，直接加装饰器就行，还要指定哪些模型要切换到副本：

def use_replica(*models):
    def decorator(func):
        def wrapper(*args, **kwargs):
            with use_replica_for_models(*models):
                return func(*args, **kwargs)
        return wrapper
    return decorator

用起来很简单：

class CarService:
    # 指定CarTransmission和Model两个模型用副本
    @use_replica(master_models.CarTransmission, master_models.Model)
    def get_transmission_specs(self):
        # 内部查询自动用replica，不用改代码
        return master_models.CarTransmission.objects.filter(
            variant_id__in=self.variant_ids
        ).values_list(*self.specifications, named=True)
    
    def other_method(self):
        # 这个方法不受影响，还是用主库查询
        return master_models.CarTransmission.objects.all()

3. 类级装饰器：给整个类的方法批量加开关 #

如果想让某个类下所有方法都自动用副本，写个类装饰器就行，它会自动给类里的所有方法加上上面的use_replica装饰：

def class_use_replica(*models):
    def decorator(cls):
        # 遍历类的所有方法，跳过__init__这类特殊方法
        for attr_name, attr_value in cls.__dict__.items():
            if callable(attr_value) and not attr_name.startswith('__'):
                setattr(cls, attr_name, use_replica(*models)(attr_value))
        return cls
    return decorator

使用示例：

# 给整个CarService类指定用副本的模型
@class_use_replica(master_models.CarTransmission, master_models.Model)
class CarService:
    def get_transmission_specs(self):
        # 自动用replica查询
        return master_models.CarTransmission.objects.filter(
            variant_id__in=self.variant_ids
        ).values_list(*self.specifications, named=True)
    
    def get_model_by_slug(self, slug):
        # get_object_or_404也会自动用replica的模型管理器
        return get_object_or_404(master_models.Model, slug=slug)

注意点 #

• 线程安全：这个方案是临时替换模型的_default_manager，多线程环境下可能有冲突。如果是异步项目，建议结合threading.local来隔离不同线程的管理器状态。
• 模型范围：一定要明确指定哪些模型要切换，别影响到其他无关模型的查询。
• 原生SQL：如果代码里有用connections['replica'].cursor()的原生SQL查询，这个方案覆盖不到，得单独处理。

替代方案：自定义模型管理器+AST代码替换 #

如果不想用上下文管理器，也可以给模型加个replica专属管理器，再用装饰器自动替换方法里的objects引用：

1. 给模型加replica管理器：

class ReplicaManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().using('replica')

# 在你的模型里添加这个管理器
class CarTransmission(models.Model):
    # ... 你的字段定义
    objects = models.Manager()  # 原有的默认管理器
    replica_objects = ReplicaManager()  # 副本专用管理器

2. 写个装饰器，用AST修改方法代码，把Model.objects换成Model.replica_objects：

import ast
import inspect
import functools

def replace_objects_with_replica(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # 获取函数的源代码
        source_code = inspect.getsource(func)
        # 解析成AST语法树
        tree = ast.parse(source_code)
        # 遍历AST节点，替换objects为replica_objects
        class ObjectReplacer(ast.NodeTransformer):
            def visit_Attribute(self, node):
                if isinstance(node.value, ast.Name) and node.attr == 'objects':
                    node.attr = 'replica_objects'
                return self.generic_visit(node)
        
        tree = ObjectReplacer().visit(tree)
        ast.fix_missing_locations(tree)
        # 编译修改后的代码，替换原函数的代码对象
        code_obj = compile(tree, inspect.getfile(func), 'exec')
        func.__code__ = code_obj.co_consts[0]
        return func(*args, **kwargs)
    return wrapper

使用示例：

class CarService:
    @replace_objects_with_replica
    def get_transmission_specs(self):
        # 代码里的objects会自动被替换成replica_objects
        return master_models.CarTransmission.objects.filter(
            variant_id__in=self.variant_ids
        ).values_list(*self.specifications, named=True)

注意点 #

• AST替换有局限性：如果是复杂的嵌套属性访问（比如some_obj.model.objects），这个替换逻辑可能失效。
• 调试不便：修改了函数的代码对象，调试时看到的代码可能和实际执行的不一致，得注意。

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