Django模型管理器中访问请求实现默认租户过滤的可行性探讨

我的代码库中有一个模型被300多个地方调用，要求该模型默认必须通过company_filter进行过滤，无例外情况。我需要从请求中获取tenant_id和tenant_type这两个值（每个请求都能拿到），先通过tenant_id过滤Company模型，再把company、tenant_id和tenant_type传入过滤器来过滤目标模型的查询集。当前方案看起来可行，但我担心会出现内存泄漏或者不同用户请求混淆的问题，想问下这个方案是否有效，有没有更优的实现方式？

中间件代码 #

import threading

_thread_local = threading.local()


class RequestMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        _thread_local.request = request

        response = self.get_response(request)

        # 视图返回后释放内存
        del _thread_local.request

        return response


def get_request():
    return getattr(_thread_local, 'request', None)

模型管理器代码 #

class TariffrateCompanyManger(models.Manager):
    # 默认按公司过滤费率
    def get_queryset(self):
        from config.middleware import get_request

        request = get_request()

        company = Company.objects.get_company_tenant_id(request.tenant_id)

        company_filter = tariff_company_filter(company, request.tenant_id, request.tenant_type)

        if request:
            qs = super().get_queryset().filter(company_filter)
        else:
            qs = super().get_queryset()

        return qs

当前方案的有效性 #

你的方案不会出现请求混淆问题：threading.local()是线程隔离的，每个请求在独立线程中处理，各自的request对象只会绑定到当前线程的本地存储，不同用户的请求数据完全隔离，不会串用。

关于内存泄漏：你在中间件中做了del _thread_local.request的清理操作，但存在一个漏洞——如果请求处理过程中抛出异常，del语句会因为代码中断而无法执行，线程本地存储里的request可能无法及时清理。可以把清理逻辑放到finally块中，确保无论请求是否成功都会执行：

import threading

_thread_local = threading.local()


class RequestMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        _thread_local.request = request
        try:
            response = self.get_response(request)
        finally:
            # 无论请求是否正常完成，都清理线程本地存储的request
            del _thread_local.request
        return response


def get_request():
    return getattr(_thread_local, 'request', None)

另外，模型管理器代码有潜在风险：如果get_request()返回None（比如非请求上下文调用，如后台任务、命令行），直接访问request.tenant_id会抛出AttributeError。同时你要求无例外过滤，所以必须处理参数缺失或非请求场景，避免返回未过滤的数据：

class TariffrateCompanyManger(models.Manager):
    # 默认按公司过滤费率
    def get_queryset(self):
        from config.middleware import get_request

        request = get_request()
        qs = super().get_queryset()

        if request:
            # 先校验请求是否包含必要参数
            if hasattr(request, 'tenant_id') and hasattr(request, 'tenant_type'):
                company = Company.objects.get_company_tenant_id(request.tenant_id)
                company_filter = tariff_company_filter(company, request.tenant_id, request.tenant_type)
                qs = qs.filter(company_filter)
            else:
                # 参数缺失时抛出异常，避免未过滤数据泄露
                raise ValueError("请求缺少tenant_id或tenant_type，无法应用公司过滤")
        else:
            # 非请求上下文强制抛出异常，确保只有过滤后的查询集能被使用
            raise RuntimeError("非请求上下文调用，请手动指定过滤条件")

        return qs

更优实现方式 #

1. 减少重复查询，提升性能 #

每次调用get_queryset都查询Company模型会增加数据库开销，可以在中间件中提前查询并缓存到线程本地存储：

import threading
from django.core.exceptions import ObjectDoesNotExist

_thread_local = threading.local()


class RequestMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        _thread_local.request = request
        try:
            # 提前查询并缓存company
            if hasattr(request, 'tenant_id'):
                try:
                    _thread_local.company = Company.objects.get_company_tenant_id(request.tenant_id)
                except ObjectDoesNotExist:
                    _thread_local.company = None
            response = self.get_response(request)
        finally:
            # 清理所有线程本地存储的变量
            for attr in ['request', 'company']:
                if hasattr(_thread_local, attr):
                    delattr(_thread_local, attr)
        return response


def get_request():
    return getattr(_thread_local, 'request', None)

def get_current_company():
    return getattr(_thread_local, 'company', None)

修改模型管理器直接使用缓存：

class TariffrateCompanyManger(models.Manager):
    # 默认按公司过滤费率
    def get_queryset(self):
        from config.middleware import get_request, get_current_company

        request = get_request()
        qs = super().get_queryset()

        if request:
            if hasattr(request, 'tenant_id') and hasattr(request, 'tenant_type'):
                company = get_current_company()
                if not company:
                    raise ValueError("不存在对应tenant_id的公司")
                company_filter = tariff_company_filter(company, request.tenant_id, request.tenant_type)
                qs = qs.filter(company_filter)
            else:
                raise ValueError("请求缺少tenant_id或tenant_type，无法应用公司过滤")
        else:
            raise RuntimeError("非请求上下文调用，请手动指定过滤条件")

        return qs

2. 降低耦合度（可选） #

模型管理器属于数据层，依赖请求上下文会增加代码耦合度。如果后续有精力优化300个调用点，可以考虑：

• 定义全局过滤函数，在视图/服务层统一应用过滤逻辑
• 使用自定义查询集结合上下文管理器，将过滤参数与请求生命周期绑定

但考虑到你现有调用点数量，当前基于线程本地存储的方案是成本最低的选择。

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