Label Studio 大规模 Django QuerySet 安全迭代指南:使用 iterate_queryset 替代 .iterator() Label Studio 大规模 Django QuerySet 安全迭代指南使用 iterate_queryset 替代 .iterator()【免费下载链接】label-studioLabel Studio is a multi-type data labeling and annotation tool with standardized output format项目地址: https://gitcode.com/GitHub_Trending/la/label-studio导读Label Studio 后端Django在批量处理任务Task、视图View等大数据集时会频繁遍历 QuerySet直接使用 Django 自带的.iterator()虽然内存开销小但存在长事务、服务器游标、连接占用等隐患且无法按主键分块复用索引优化。本文以仓库内.cursor/rules/iterate_queryset.md编码规范为骨架结合 核心实现 与 单元测试 源码完整讲解iterate_queryset()的用法、pk 分块与流式回退的双路径实现原理、聚合查询的边界处理、keyset 批量迭代辅助函数以及它们在数据迁移、Data Manager 批量更新、FSM 状态回填中的真实落地场景。读完你不仅能写出合规的大数据集遍历代码还能理解其底层机制并正确规避顺序丢失聚合被拆分等陷阱。一、为什么不要直接调用.iterator()Django 官方为大型 QuerySet 提供了.iterator()它按需从数据库取行避免一次性把全部对象加载进内存。但 Label Studio 的编码规范明确指出遍历大型 QuerySet 时应优先使用iterate_queryset()而不是.iterator()from core.utils.iterators import iterate_queryset # 推荐推荐写法 for obj in iterate_queryset(queryset): process(obj) # 避免不推荐写法 for obj in queryset.iterator(): process(obj)从源码结构看直接使用.iterator()主要带来三类问题长事务与游标占用.iterator()依赖服务端游标在事务未结束时连接一直被占用对 Postgres/MySQL 连接池压力较大Label Studio 的异步任务、存储同步等场景并发度高容易被拖垮。无法按块重建查询.iterator()只是一条 SQL 的流式执行无法利用按主键切片 pk__in过滤这种可走索引、可中断续跑的分块策略。难以统一收口当项目需要为所有大数据集遍历统一调整分块大小、增加特性开关或修复边界 bug 时分散的.iterator()调用无法集中治理。因此仓库在 label_studio/core/utils/iterators.py 中提供了统一的迭代入口全项目所有大表遍历都收口到这两个函数上。二、iterate_queryset()的 API 与默认分块配置函数签名如下见 iterators.pydef iterate_queryset(queryset, chunk_sizeNone):参数queryset任意 Django QuerySet已有过滤器filter/exclude、注解annotate、select_related、prefetch_related、only/defer优化都会保留。参数chunk_size每次从数据库加载的对象数量上限。传None时使用全局默认值settings.QS_ITERATOR_DEFAULT_CHUNK_SIZE。返回值一个生成器generator逐条产出模型实例。默认分块大小定义在 label_studio/core/settings/base.py#L1111# QuerySet iterator settings QS_ITERATOR_DEFAULT_CHUNK_SIZE int(get_env(QS_ITERATOR_DEFAULT_CHUNK_SIZE, 1000))也就是说默认每块加载 1000 条记录可通过环境变量QS_ITERATOR_DEFAULT_CHUNK_SIZE覆盖。若传入的chunk_size 0函数会直接抛出ValueErrorif chunk_size 0: raise ValueError(fchunk_size must be positive, got {chunk_size})这一行为有对应测试约束见 test_iterators.pydef test_iterate_queryset_in_keyset_batches_rejects_invalid_chunk_size(): queryset UserFactory._meta.model.objects.all() with pytest.raises(ValueError, matchchunk_size must be positive): list(iterate_queryset_in_keyset_batches(queryset, chunk_size0))三、核心实现原理pk 分块pk chunking双路径iterate_queryset()的实现并非总是分块它根据是否开启特性开关以及查询是否包含显式 GROUP BY走两条完全不同的路径源码见 iterators.py#L24-L50if not flag_set(fflag_fix_back_plt_863_remove_iterator_27082025_short, userauto) or _has_explicit_group_by(queryset): for obj in queryset.iterator(chunk_sizechunk_size): yield obj return model queryset.model pk_field model._meta.pk.name all_ids list(queryset.values_list(pk_field, flatTrue)) if not all_ids: return for i in range(0, len(all_ids), chunk_size): chunk_ids all_ids[i : i chunk_size] # 基于原始 queryset 重建新查询保留 annotations、select_related、 # prefetch_related、only/defer 等所有优化 chunk_qs queryset.filter(**{f{pk_field}__in: chunk_ids}) for obj in chunk_qs: yield obj3.1 路径一流式迭代回退路径当特性开关fflag_fix_back_plt_863_remove_iterator_27082025_short未开启或查询包含显式 GROUP BY 时直接退化为带chunk_size的.iterator()逐条产出。该特性开关当前在 label_studio/feature_flags.json#L4986 中on: true同时登记在 stale_feature_flags.py 中说明新代码默认走 pk 分块路径。3.2 路径二pk 分块默认路径当开关开启且无显式 GROUP BY 时执行主键分块先用queryset.values_list(pk_field, flatTrue)一次性收集全部主键到内存只取主键列内存占用远小于取整行对象空结果直接返回按chunk_size切片主键对每一片用queryset.filter(pk__inchunk_ids)基于原始 queryset 重建子查询并遍历。这样做的收益非常明显每个子查询都走主键索引可以按块限流chunk_qs逐块执行而非一条大 SQL 流式且原始 queryset 上的所有优化annotations、select_related、prefetch_related、only/defer都被完整保留——这正是源码注释强调的 preserving all optimizations。3.3 顺序不保证重要约束原规则文档明确警示iterate_queryset()不保证保持顺序。原因在 pk 分块路径下显而易见主键列表切片本身不保证与查询自然顺序一致每个pk__in子查询的返回顺序由数据库决定通常按主键而非原始排序分块之间是串行拼接块与块之间天然没有全局顺序概念。因此在用户可见列表、导出、报表等顺序有语义的场景不要使用iterate_queryset()这正是规则文档if order matters, dont use it的落地点。需要有序遍历时请使用下文第四节的 keyset 批量迭代方案或显式order_by后按需自行分页。四、聚合查询的边界处理_has_explicit_group_bypk 分块对聚合查询有一个隐蔽的坑对.values(...).annotate(...)的查询做pk__in重写会把显式 GROUP BY 悄悄替换成按主键分组导致聚合被拆分——每个分组只统计到落在当前块内的行出现重复行 少计总数。源码通过_has_explicit_group_by()守卫来规避def _has_explicit_group_by(queryset): group_by getattr(queryset.query, group_by, None) return bool(group_by) and group_by is not True注释解释了关键语义.values(...).annotate(...)会产生显式 GROUP BY 字段列表而普通.annotate()group_by 为True表示按每个选中字段分组一行对应一个对象与 pk 分块兼容所以只有显式分组才走流式回退。对应测试 test_iterate_queryset_preserves_aggregate_rows 验证了该行为构造 3 个用户各 3 个项目的聚合查询chunk_size2故意不能整除断言每个分组恰好产出一次且计数为 3证明聚合行没有被拆分或重复。def test_iterate_queryset_preserves_aggregate_rows(): A .values().annotate() queryset must yield one row per group, not one per table row. ... queryset ( Project.objects.filter(created_by__inusers) .values(created_by_id) .annotate(project_countCount(id)) .order_by(created_by_id) ) rows list(iterate_queryset(queryset, chunk_size2)) assert [row[created_by_id] for row in rows] sorted(user.id for user in users) assert [row[project_count] for row in rows] [3, 3, 3]普通模型查询则仍走 pk 分块路径由 test_iterate_queryset_still_chunks_plain_querysets 保证分块后每个对象恰好返回一次。五、进阶iterate_queryset_in_keyset_batches()keyset 批量迭代当需要按批次而不是逐条处理对象且要求稳定的键序游标keyset pagination时同一模块还提供了iterate_queryset_in_keyset_batches()见 iterators.py#L53-L108。它与iterate_queryset()的关键差异如下维度iterate_queryset()iterate_queryset_in_keyset_batches()产出单位逐条 yield 模型实例批量 yield每批最多chunk_size个实例排序不保证pk 分块会打乱强制order_by(key_field)覆盖原查询排序分页机制主键切片 pk__in键集游标key_field__gt翻页断点续传不支持支持start_after/stop_at适用场景顺序无关的大表逐条处理需按单调键顺序分块扫描、可恢复的任务队列函数签名与参数def iterate_queryset_in_keyset_batches( queryset, chunk_sizeNone, key_fieldpk, start_afterNone, stop_atNone, ):key_field用于分页的单调字段通常为pk或id要求取值唯一性足够保证key_field__gt翻页不跳过同值行。start_after可选跳过key_field start_after的行实现断点续跑。stop_at可选包含上界过滤key_field stop_at的行。文档字符串明确警告该函数总是施加order_by(key_field)会覆盖并破坏原 queryset 的任何排序因此不要用于用户可见列表、导出、报表等排序有意义的场景与规则文档不保证顺序的警示一脉相承。测试覆盖了按键排序分块test_iterate_queryset_in_keyset_batches_orders_by_key_field、start_after/stop_at边界test_iterate_queryset_in_keyset_batches_respects_start_after_and_stop_at以及空查询集返回空test_iterate_queryset_in_keyset_batches_accepts_empty_queryset。六、仓库内的真实使用场景规则并非空谈两个函数已广泛落地于 Label Studio 后端的关键路径可直接作为参考范式。6.1 Data Manager 批量更新 Task 数据列data_manager/actions/data_columns.py#L59-L75 在批量改写任务数据列如data字段时用iterate_queryset配合.only()只加载必要字段再与batched_iterator组合分批写回if settings.DJANGO_DB settings.DJANGO_DB_SQLITE: updated_count 0 task_iterator iterate_queryset(queryset.only(id, data), chunk_sizesettings.UPDATE_COLUMN_BATCH_SIZE) for task_batch in batched_iterator(task_iterator, settings.UPDATE_COLUMN_BATCH_SIZE): for task in task_batch: task.data task.data or {} ...同样的模式还出现在 data_columns.py#L317 的range命令处理中iterate_queryset(queryset.only(id, data), chunk_sizesettings.UPDATE_COLUMN_BATCH_SIZE)可见.only()瘦身 iterate_queryset 分块是官方推荐的内存友好组合。6.2 数据迁移中的全表扫描data_manager/migrations/0018_remove_allow_skip.py#L38-L40 在正向/反向数据迁移中遍历全部View对象时直接使用默认分块views iterate_queryset(View.objects.all())迁移任务由start_job_async_or_sync驱动借助 pk 分块避免一次性把全部 View 加载进内存这正是大数据迁移安全迭代的标准姿势。6.3 FSM 状态回填避免 OOMfsm/functions.py#L20-L64 中backfill_fsm_states为存储同步创建的任务回填初始 FSM 状态源码注释与文档字符串都明确写着Tasks are processed in chunks viaiterate_querysetto avoid OOM issues.实现上先收集task_ids列表再iterate_queryset(Task.objects.filter(id__intask_ids))逐条处理fsm/functions.py#L60-L64并配合CurrentContext.get_user()逐条初始化状态。这是存储同步批量建任务 大批量状态回填场景下防止进程内存暴涨的典型处理。七、使用规范小结综合规则文档与源码在 Label Studio 中遍历大型 QuerySet 时应遵守以下约定默认统一用iterate_queryset(queryset)不要散落.iterator()调用分块大小由QS_ITERATOR_DEFAULT_CHUNK_SIZE默认 1000统一控制可用环境变量调整。顺序敏感场景禁用iterate_queryset()不保证顺序iterate_queryset_in_keyset_batches()会强制按键排序并覆盖原排序用户可见列表、导出、报表请另选有序分页方案。尽量配合字段瘦身只需部分列时用.only(id, ...)能显著降低每块内存占用参考 data_columns.py 的写法。聚合查询交给流式回退values().annotate()这类显式 GROUP BY 查询会自动走.iterator()路径不要手动干预也不要担心被 pk 分块拆坏——有测试兜底。批量处理用 keyset 版本需要按批处理、断点续跑时选用iterate_queryset_in_keyset_batches()并传key_field、start_after、stop_at。chunk_size必须为正数传 0 或负数会抛ValueError编写通用代码时注意参数校验。如需深入验证或复现可重点阅读 核心实现、单元测试、默认配置 三处文件并对照 data_columns.py 与 fsm/functions.py 两个真实落地案例。【免费下载链接】label-studioLabel Studio is a multi-type data labeling and annotation tool with standardized output format项目地址: https://gitcode.com/GitHub_Trending/la/label-studio创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考