Django遍历大型查询集时调用模型方法的最优方案

你当前代码的核心性能瓶颈是循环内的重复数据库查询，即每次遍历都调用get()获取实例，1万条数据就会产生至少1万次单独的数据库请求，再叠加模型方法内的外键、反向关联查询，耗时自然很高。可按以下优先级逐步优化：

1. 消除N+1查询，预加载关联数据 #

完全放弃先查values()再循环get()的逻辑，直接遍历查询集，同时用select_related（预加载外键关联）、prefetch_related（预加载反向关联、多对多关联）把模型方法中用到的所有关联数据提前一次性查询出来，避免方法内每次访问关联字段都触发新的数据库请求。

# 替换原来的filter，根据方法内用到的关联字段调整select_related、prefetch_related的参数
my_model_qs = MyModel.objects.filter(
    fk_field_1__in=list_of_fk_1_field_ids
).select_related(
    "外键字段1", "外键字段2" # 填写方法内用到的所有外键字段
).prefetch_related(
    "反向关联字段1", "反向关联字段2" # 填写方法内用到的所有反向关联、多对多字段
)

my_args = {"start_date": some_start_date, "end_date": some_end_date}
final_data = []

# 一次循环完成方法调用+数据组装，不需要跑两次循环
for obj in my_model_qs:
    val_1 = obj.my_model_method_1(**my_args)
    val_2 = obj.my_model_method_2(**my_args)
    # 直接组装最终需要的Excel数据结构
    final_data.append({
        "id": obj.id,
        # 其他需要的字段直接从obj取，比如obj.field1, obj.field2
        "val_1": val_1,
        "val_2": val_2,
        # 其他逻辑处理后的字段
    })

这一步优化完成后，通常耗时可以降到数分钟以内。

2. 将计算逻辑下沉到数据库层面 #

如果模型方法内的计算逻辑可以转化为数据库的聚合、条件判断，优先用annotate在查询时直接计算出val_1、val_2，完全不需要在Python层循环调用方法，性能会提升一个数量级。
举个示例（根据实际逻辑调整）：

from django.db.models import Sum, Q, F

my_args = {"start_date": some_start_date, "end_date": some_end_date}
my_model_data = MyModel.objects.filter(
    fk_field_1__in=list_of_fk_1_field_ids
).annotate(
    # 假设val1是关联表符合日期范围的金额求和
    val_1=Sum(
        "关联表名__amount", 
        filter=Q(关联表名__date__range=(my_args["start_date"], my_args["end_date"]))
    ),
    # 假设val2是两个字段的差值计算
    val_2=F("field_a") - F("field_b")
).values("id", "field1", "field2", "val_1", "val_2")

这种方案处理1万条数据通常只需要几秒到几十秒。

3. 增加缓存避免重复计算 #

如果相同参数下同一个实例的方法返回值不会频繁变化，可以给模型方法加缓存，避免重复计算和查询：

from django.core.cache import cache

class MyModel(models.Model):
    # 原有字段
    def my_model_method_1(self, start_date, end_date):
        # 生成唯一缓存key
        cache_key = f"method1_{self.id}_{start_date.strftime('%Y%m%d')}_{end_date.strftime('%Y%m%d')}"
        cached_val = cache.get(cache_key)
        if cached_val is not None:
            return cached_val
        # 原有计算逻辑
        res = 你的计算逻辑
        # 缓存1小时，可根据实际需求调整过期时间
        cache.set(cache_key, res, 3600)
        return res

4. 并行计算提升效率 #

如果计算逻辑无法下沉到数据库，也可以用多线程/多进程并行处理每个实例的计算，注意控制并发数避免数据库连接被占满：

from concurrent.futures import ThreadPoolExecutor

my_model_qs = MyModel.objects.filter(fk_field_1__in=list_of_fk_1_field_ids).select_related(...).prefetch_related(...)
my_args = {"start_date": some_start_date, "end_date": some_end_date}

def process_obj(obj):
    val_1 = obj.my_model_method_1(**my_args)
    val_2 = obj.my_model_method_2(**my_args)
    return {
        "id": obj.id,
        "val_1": val_1,
        "val_2": val_2,
        # 其他字段
    }

# 线程数建议设置为CPU核心数的2-4倍
with ThreadPoolExecutor(max_workers=8) as executor:
    final_data = list(executor.map(process_obj, my_model_qs))

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