Saleor 开发规范深度解读横向扩展架构下的并发、测试与数据迁移实战指南【免费下载链接】saleorSaleor Core: the high performance, composable, headless commerce API.项目地址: https://gitcode.com/gh_mirrors/sa/saleor本文基于仓库根目录下的 CLAUDE.md 系统梳理 Saleor Core高性能、可组合的 headless 商务 API的工程开发规范覆盖部署模型心智模型、多租户站点解析、GraphQL 约定、测试规范、并发与线程安全模式、数据迁移最佳实践、性能与数据库访问原则等核心主题。读完本文你将理解为什么 Saleor 要求把所有操作都当作与自身并发执行并掌握F()表达式、select_for_update、on_commit、update_or_create等并发原语在真实项目中的落地用法以及一套可直接复制的 pytest 断言与 GraphQL 授权测试范式。一、部署模型一切规范的第一性原理Saleor 面向**水平扩展horizontal scale-out**部署大量完全相同的 Kubernetes PodWeb 服务 Celery worker并发运行在负载均衡器之后共享一个 Postgres带读副本与共享的 Redis/Valkey 消息代理。任何一个请求或任务可能落在任意 Pod 上同一份代码同时运行在多个进程中。这个部署模型是阅读本文所有规范的起点CLAUDE.md 强调写代码时要时刻牢记这一点进程内/本地状态不能作为事实来源source of truth。永远不要依赖模块级全局变量、内存缓存或上一个请求设置的值会在下一个请求存在这样的假设——下一个调用可能落在不同的 Pod 上。共享状态只存在于 Postgres 与 Valkey/Redis 中。假设每个操作都会与自身并发执行。使用原子数据库操作F()、update_or_create、select_for_update替代先检查后执行check-then-act。绝不阻塞请求/worker 线程。请求处理代码中禁止time.sleep()和无界while True重试循环——一个被阻塞的线程会浪费 Pod 上有限的 worker 槽位。使用有界重试并把等待逻辑交给后台任务。在事务提交之后调度后台工作。使用on_commit/call_event而不是在开启的事务里裸调.delay()否则任务可能在别的 Pod 上先于事务提交运行从而读不到它需要的行。保护共享数据库。它是扩展的瓶颈避免 N1 查询和循环内的逐行写入用 bulk保持select_for_update事务简短只取需要的列不要在热路径上增加每请求查询——复用 dataloader。让任务幂等且可安全重试——broker 可能多次投递同一个任务Pod 可能在运行途中被杀掉或重新调度。二、多租户永远不要依赖settings.SITE_ID一个部署可能用一个进程服务多个站点且SITE_ID可能未设置或与开发环境不同。禁止直接读取它来识别当前站点也禁止假设只有一个站点。用Site.objects.get_current()解析站点GraphQL 层则用get_site_promisedataloader。两者在配置了SITE_ID时按它选站点未配置时回退到请求主机——直接依赖settings.SITE_ID会把所有租户静默折叠成一个。任何按站点隔离的东西都用解析出的站点来作为 key而不是常量。缓存 key、模块级 dict 或查询过滤器若把单个站点写死会让一个租户读到另一个租户的数据。不要假设只有一行。Site.objects.get()在第二个站点出现时就会崩溃要按解析出的站点过滤。从源码看Site.objects.get_current()读取的是打补丁后的进程级全局缓存THREADED_SITE_CACHE见 saleor/site/patch_sites.py该文件将django.contrib.sites的SiteManager.get_current、clear_cache、get_by_natural_key替换为线程安全实现threading.Lock保护按SITE_ID或请求 host 查库并缓存。关键细节该缓存永远不会被Site/SiteSettings的保存操作失效因此同一进程中写入后通过它读到的值可能是陈旧的——这就是下文编写测试一节强调断言方式的原因。测试中需要显式Site.objects.clear_cache()才能破坏缓存。三、GraphQL 约定版本化与权限API 版本化先查看最近的 git tag 确定当前分支基于哪个版本例如 3.22 tag在此基础上新字段的描述需要加上ADDED_IN_{VERSION}子句。GraphQL 权限用PermissionsField描述字段的访问限制而不是在解析器内部手写权限判断。四、架构原则显式调用拒绝隐式信号不要使用 Django signals如post_save、pre_delete、receiver。应当从触发逻辑的代码处显式调用相关逻辑而不是通过信号处理器串联。唯一例外是用于触发数据迁移任务的post_migrate信号见数据迁移一节。五、代码风格使用全局 import 语句所有 import 放在文件顶部不要放在函数、方法或其他局部作用域内。六、测试规范6.1 在 git worktree 中运行测试当 Saleor 被检出在 git worktree 中时优先在该 worktree 的容器内运行操作测试、管理命令等而不是在宿主机上。宿主机的服务可能属于另一个 worktree直接在宿主机运行会打到错误的数据库/缓存或与其他 worktree 的栈冲突。使用.worktree-container/compose.sh包装脚本它指向当前 worktree 的隔离栈在saleor服务内运行命令.worktree-container/compose.sh up -d # 启动当前 worktree 的栈 .worktree-container/compose.sh exec saleor pytest --reuse-db saleor/path/to/test_file.py .worktree-container/compose.sh exec saleor python manage.py migrate在 worktree 中工作时不要直接在宿主机上运行这些操作。6.2 运行测试用pytest运行测试。附加--reuse-db参数以复用测试数据库、加速测试。通过传测试文件路径作为参数来选择要运行的测试。在开发容器内依赖是系统级安装的直接运行pytest即可——无需激活虚拟环境。在宿主机容器外上先进入虚拟环境再执行测试。6.3 编写测试的通用规则使用 given/when/then 结构清晰表达给定条件-执行动作-断言结果。使用pytestfixtures 做 setup 和 teardown测试套件在文件中扁平声明不要包在类里。优先用 fixture 而不是 mock。fixtures 通常在tests/fixtures目录下是pytest.fixture装饰的函数当测试对象没有现成 fixture 时新建一个并可用 factory 向 fixture 传参。断言实际返回值而非不为 None。例如assert email is ab.com而不是assert email is not None。断言 GraphQL 错误时同时断言错误消息。断言 Enum 时在测试文件中 import 枚举用assert error.code MyEnum.SOME_ERROR.name而不是字符串比较。如果已有对值的引用避免断言裸值。例如创建了实体并断言响应中返回该实体时用实体的字段与响应字段比较而不是写死字面量。构建 GraphQL 查询变量时使用枚举值而非普通字符串{field: TransactionSortField.CREATED_AT.name}而不是{field: CREATED_AT}。不要给测试名写多余前缀文件叫test_my_mutation.py时测试应命名test_with_x_y而不是test_my_mutation_with_x_y。测试 GraphQL 预期错误场景时总是断言 errors 列表的期望长度。准备测试数据时把值提取到变量并在断言中复用不要在 setup 和 assertion 之间重复字面量。比较 JSON payload 时用json.loads()转成 dict 再比较不要用json.dumps()比较序列化字符串——key 顺序变化会让字符串比较失效。期望恰好一行时用Model.objects.get()而不是.first()加is not None断言。get()同时完成取数与恰好存在一行的断言可以省掉单独的 not-None 检查。存在性检查优先用qs.exists()而不是qs.count() 0/! 0——exists()比COUNT便宜。示例assert checkout.lines.exists() is False。相似场景用pytest.mark.parametrize参数化而不是复制粘贴几乎相同的测试体。用带_case前缀的字符串参数标识每个用例而不是用ids参数——描述紧挨着数据更易读易改pytest.mark.parametrize( (_case, value), [ (omitted, {}), (empty_list, {lines: []}), ], ) def test_something(_case, value): ...当 mutation 输入是可选的/nullable 时除了省略该字段还要显式测试null值——两者是不同的输入可能被区别处理。永远不要假设缓存是冷的——或热的。缓存是进程级全局的跨测试存活因此缓存由谁预热取决于测试顺序。要断言写入已落地就回读行obj.refresh_from_db(fields(...))而不是通过Site.objects.get_current()这类缓存访问器要统计查询数先预热缓存得到稳态否则显式破坏缓存Site.objects.clear_cache()。一个只因缓存恰好为空才通过的测试会在某个无关调用者预热缓存后立刻失败。6.4 精确断言review 评论的第一大来源CLAUDE.md 专门开辟一节强调精确断言要点如下断言精确的值与数量绝不断言边界或仅仅不存在。不要写len(x) 0、 1、assert not errors、call_count 1——断言精确的计数、枚举或对象才能捕获意外多余的返回结果。期望一个错误时断言恰好一个、完整错误消息以及它的path/字段。在否定/权限拒绝测试中还要断言副作用确实没有发生——余额未变、SomeEvent.objects.exists() is False、对象未被修改。仅有错误断言并不能证明操作被拦截。绝不断言测试自身刚设置并保存的值——那验证的是 ORM 而不是被测代码。要么设置状态要么断言它二者取其一。绝不要把可变输入在传给可能修改它的代码之后又对它自身做断言——这种比较永远不可能失败。应设计为期望值独立于输入用字面量或单独 fixture 构造。copy.deepcopy仅在确有修改风险且没有更干净方案时作为最后手段——它很昂贵涉及 pickle无法扩展到大型套件。优先用真实 fixture/token 而不是 mock 内部机制。用真实的create_access_token*工厂或 pyjwt 生成真正有效/无效的 token而不是 mockget_decoded_token返回固定 payload——否则真实故障不会被发现。修复必须附带回归测试端到端复现实际报告的 bug并针对真正失败的字段/path而不是只测新 helper 的 happy path。把前置条件编码为断言而非注释在调用前写assert flag is True这样默认值一旦变化测试会快速失败。期望恰好一行时用Model.objects.get()它断言唯一性而不是.first() not-None 检查用qs.exists()而不是qs.count() 0。不要硬编码不存在的主键如99999——--reuse-db可能重新分配它。改用负数 id 如-1Postgres 没有无符号整数类型负数永远不可能匹配真实行还省掉一次max(pk)查询。测试中的 URL 使用保留测试域名example.com、*.test绝不使用真实域名。IP 使用网络地址使其无法触达真实服务器例如8.8.8.0或1.1.1.0。参数化近似相同的测试体保持每个测试最小化删掉无关 setup按断言的确切行为命名避免双重否定。6.5 GraphQL 授权测试什么时候应该写授权测试新增受限字段restricted field时新增受限类型restricted type时新增受限 mutation 或 query 时新增公开mutation 或 query 时——这确保我们显式确认新 query/mutation 不需要任何授权。写授权测试时必须全部做到以下几点覆盖所有客户端类型未认证Unauthenticated无权限用户非 staff 用户缺少所需权限的 staff 用户拥有正确权限的 staff 用户使用参数化测试不要创建多个独立测试。绝不仅仅依赖错误消息判断访问被拒绝——因为即使报错仍可能返回了受限数据或执行了操作。因此总是确保受限数据没有出现在 HTTP 响应中若是 GraphQL mutation总是确保数据没有被修改若是 GraphQL mutation总是确保外部调用webhooks、API 调用、邮件等没有被触发。标准模板如下可直接复用pytest.mark.parametrize( (_case, client_fixture, permission_fixture, is_allowed), [ ( Unauthenticated user should be rejected, api_client, None, False, ), ( Authenticated unprivileged user (non-staff) should be rejected, user_api_client, None, False, ), ( Authenticated user w/o the permission should be rejected, staff_api_client, None, False, ), ( Authenticated user w/ the correct permission should be allowed, staff_api_client, permission_manage_settings, True, ), ] ) def test_authorization( request, client_fixture: str, permission_fixture: str | None, is_allowed: bool, ): client request.getfixturevalue(client_fixture) if permission_fixture: perm request.getfixturevalue(permission_fixture) if client.app: client.app.permissions.add(perm) elif client.user: client.user.user_permissions.add(perm) else: raise AssertionError(Couldnt add the permission) # shouldnt occur response client.post_graphql(query, variables) if is_allowed: content get_graphql_content(response) assert content[data] ... else: assert_no_permission(response) content get_graphql_content_from_response(response) assert content[data] ...七、代码结构与分层业务逻辑放在领域层domain layerGraphQL mutation 只做输入校验与编排。token 如何构造/剥离、余额如何变化等属于领域模块而不是 mutation。非 GraphQL 代码绝不能从graphql层 import。如果领域层的任务或模型需要某个 helper把 helper 移到领域模块中由 GraphQL 层从领域层导入——绝不允许反向。在数据产生点做校验fail-fast子实体在与父实体同一个 clean 步骤中清理——不要把同一实体的清理拆到不同阶段。只传函数真正需要的窄值单个 flag 或已取到的 setting而不是让函数去拆解宽泛的info/SiteSettings上下文对象。宽参数会隐藏真实依赖迫使调用方反复穿线上下文。优先用类型化结构dataclass/NamedTuple而不是即兴的dict集合并保持函数返回形状同质不要在同一个列表中混入异构实体。把分支多或冗长的逻辑提取成具名 helper而不是用注释叙述一个约 100 行的函数。先执行短路/全局检查再去取数据。删除重构产生的死代码未使用的校验器、仅为满足 mypy 而加的不可达守卫。要从源头修正类型而不是在错误的上游注解外面包一层cast()。八、命名规范禁止晦涩缩写gc_brand应写成gift_card_brand新字段名与已有的同类字段对齐。使用正向布尔命名——delivery_changed而不是not delivery_unchanged。任何门控废弃deprecated行为的 flag 或字段加legacy_前缀让废弃从名字上一眼可见。名字要反映返回值/含义与正确的领域而不是偶然的调用点不要把格式假设写进名字code 可能含字母时用last_chars而不是last_digits。九、错误处理捕获具体异常绝不捕获裸Exception或DatabaseError。需要时捕获整个 provider 异常族BotoCoreError和ClientErrorGoogleAPIError同时覆盖 4xx和5xx以及最窄的数据库异常OperationalError而不是基类。区分可重试失败网络错误、HTTP 5xx与终态失败4xx、坏数据、ValueError——绝不重试无法自愈的失败终态失败立即返回/中止让无效工作快速排空。读取 HTTP 响应体之前先检查状态码出站网络调用总是传显式 timeout。错误消息必须说明实际原因not found 与 wrong type 不同对列表输入要归因到具体的行/字段配置/店铺级失败用通用非字段错误。传播捕获到的原始消息而不是重写它。校验引用的 GraphQL ID 并返回干净的错误——不要让坏/不存在的 id 暴露出晦涩的 Cannot return null for non-nullable field 崩溃。十、Webhooks 与事件分发分发 webhook 事件时总是用saleor.core.utils.events中的call_event而不是直接调用 manager 方法。Bad:manager.product_variant_discounted_price_updated(price_info, webhookswebhooks)Good:from saleor.core.utils.events import call_event call_event(manager.product_variant_discounted_price_updated, price_info, webhookswebhooks)从实现看saleor/core/utils/events.pycall_event的核心逻辑是检查当前连接是否处于原子事务块connection.in_atomic_block若在事务中则通过transaction.on_commit延迟到提交后执行否则立即执行——这正好呼应调度后台工作在事务提交之后的部署模型要求确保 webhook 不会在数据提交前被其他 Pod 消费。十一、并发与线程安全模式重点章节Saleor 运行在多个并发执行的 Python 服务中。以下是必须遵循的线程安全模式其中多数对应 CLAUDE.md 末尾的Saleor 使用的模式汇总表。11.1 原子递增模式Atomic Increment永远不要用instance.count 1; instance.save()——这不是原子操作会造成丢失更新lost updates。Bad:existing.count 1 existing.save(update_fields[count])Good - 使用 F() 表达式from django.db.models import F Model.objects.filter(pkexisting.pk).update(countF(count) 1)仓库中的真实案例见 saleor/discount/utils/voucher.pyincrease_voucher_code_usage_value使用code.used F(used) 1再save(update_fields[used])完成优惠券码使用次数递增递减则用Case/When守卫下限used__gt0时才F(used) - 1否则置 0。11.2 避免 Check-Then-Act 竞态永远不要在独立操作中先检查记录是否存在再创建/更新它。Bad:existing Model.objects.filter(appapp, keykey).first() if not existing: Model.objects.create(appapp, keykey, ...) # Race condition: duplicate may be created else: existing.update(...)Good - 使用带唯一约束的 update_or_createobj, created Model.objects.update_or_create( appapp, keykey, # Lookup fields defaults{message: message, updated_at: now} # Fields to update )注意update_or_create要真正安全lookup 字段上必须存在唯一约束。11.3 行级锁select_for_update对于需要在同一行上进行多次读/写的复杂操作使用select_for_updatefrom saleor.core.tracing import traced_atomic_transaction with traced_atomic_transaction(): obj Model.objects.select_for_update().get(pkpk) # Perform operations - row is locked until transaction ends obj.save()模式创建 lock_objects.py 模块参考 saleor/payment/lock_objects.pydef my_model_qs_select_for_update() - QuerySet[MyModel]: return MyModel.objects.order_by(pk).select_for_update(of[self])真实实现中transaction_item_qs_select_for_update正是TransactionItem.objects.order_by(pk).select_for_update(of[self])并且组合出get_order_and_transaction_item_locked_for_update、get_checkout_and_transaction_item_locked_for_update这样的高层 helper一次性锁定 ordertransaction item 或 checkouttransaction item保证支付/退款场景的读写一致性。select_for_update 的坑Queryset 是惰性的——未求值的锁 queryset 不会执行 SQL因此不会加锁而且会静默失败。必须在事务内强制求值list(...)、.get()/.first()或把它作为写操作的__in子查询。with transaction.atomic(): # Bad: lazy - no lock _locked model_qs_select_for_update().filter(pk__inpks) # Good: rows locked, only pks fetched _locked list( model_qs_select_for_update().filter(pk__inpks).values_list(pk, flatTrue) ) Model.objects.bulk_update(objs, [field])用assert_locks_rows_before_writefixturesaleor/tests/fixtures.py测试锁——它捕获代码块的查询并断言在UPDATE之前发出了带确定性ORDER BY的FOR UPDATE查询。从源码看该 fixture 从捕获的 SQL 中筛出含FOR UPDATE的查询和UPDATE开头的查询断言恰好两条、且锁查询在写查询之前、FOR UPDATE确实存在。它直接对 SQL 断言是刻意为之因为漏加锁很难用行为测试发现。11.4 数据库事务用traced_atomic_transaction包裹相关操作from saleor.core.tracing import traced_atomic_transaction with traced_atomic_transaction(): # All operations here are atomic obj1.save() obj2.save()从 saleor/core/tracing.py 的实现看它本质是transaction.atomic()外包一层 OpenTelemetry span名为transaction组件属性为orm因此每个事务都会自动产生可观测的追踪跨度。11.5 在提交后调度后台工作永远不要在开启的事务里用裸.delay()调度 Celery 任务——任务可能在别的 Pod 上先于事务提交启动读到它还看不到的数据而失败。用on_commit或call_event它对 webhook 事件就是这么做的让任务只在提交后触发from django.db.transaction import on_commit with traced_atomic_transaction(): obj.save() on_commit(lambda: my_task.delay(obj.pk))多写/读-改-写操作要包在事务里做到全有或全无并在同一个事务里发出状态变更事件。11.6 绝不阻塞请求/worker 线程请求处理或 worker 代码中禁止time.sleep()和无界while True重试循环——阻塞的线程浪费 Pod 上有限的 worker 槽位。改用有界for attempt in range(n)重试把等待卸载给后台任务。11.7 其他并发规则把allow_writer()的作用域限制到恰好需要它的那条语句with allow_writer(): ...而不是整个任务/函数。在每个 Promise.then()回调内重新建立 DB/writer 上下文——它在上一个上下文退出之后才运行类似 async所以在.then()之前打开的 writer/context 在回调内部已经失效。绝不要用独立的empty()/非空检查去 gate 阻塞式Queue.get()——检查与 get 之间最后一个元素可能被取走导致消费者挂起。用非阻塞 get 或哨兵sentinel。把重负载扇出任务大下载、数千对象路由到专门的、有界队列避免单个 bulk mutation 饿死 worker 或让 Pod OOM。11.8 模式汇总表CLAUDE.md 给出的速查表模块示例均为仓库真实路径PatternModule ExampleWhen to UseF()atomic incrementsaleor/discount/utils/voucher.py计数器更新select_for_updatesaleor/payment/lock_objects.py复杂读-改-写update_or_createsaleor/graphql/attribute/utils/type_handlers.pyUpsert 操作traced_atomic_transactionsaleor/core/tracing.py多操作原子性on_committask scheduling任何分发任务的 mutation仅在提交后触发任务十二、数据迁移Data Migrations12.1 将搜索索引标记为 dirty当新增或修改搜索向量索引逻辑时创建一个数据迁移把所有现存行标记为 dirty让它们被重新索引。结构Task 文件—— 放在saleor/app/migrations/tasks/saleorversion.pyMigration 文件—— 放在saleor/app/migrations/number_name.pyTask 规则用app.task(queuesettings.DATA_MIGRATIONS_TASKS_QUEUE_NAME)和allow_writer()装饰分批处理例如每批 1000 行避免长事务通过 app 的lock_objects模块用select_for_update锁定行再更新使用子查询模式先锁行再在单独查询中 filterupdatewith transaction.atomic(): pks ( model_qs_select_for_update() .filter(pk__inbatch_pks) .values_list(pk, flatTrue) ) Model.objects.filter(pk__inpks).update(search_index_dirtyTrue)末尾还有更多行时用.delay()自链self-chain当批次无返回时停止链式调度。Migration 规则用post_migrate信号在全部迁移完成后触发 Celery 任务不要内联触发在RunPython操作内部连接post_migrate用registry.get_app_config(app)作为 sender总是提供migrations.RunPython.noop作为反向操作。参考示例Task 见 saleor/giftcard/migrations/tasks/saleor3_22.pyMigration 见 saleor/giftcard/migrations/0023_mark_gift_cards_search_vector_as_dirty.py。后者在RunPython中定义on_migrations_complete以registry.get_app_config(giftcard)为 sender 连接post_migrateweakFalse迁移完成后.delay()触发批量打 dirty 标记的任务。12.2 并发添加索引 / 唯一约束同步构建索引默认的AddIndex/AddConstraint在构建期间会对整张表持有ACCESS EXCLUSIVE锁阻塞读写。在大表上这意味着显著的停机时间。应改用CONCURRENTLY构建索引不阻塞并发流量。规则CREATE INDEX CONCURRENTLY不能在事务内运行所以创建它的迁移必须设置atomic False。让并发索引独占一个迁移。快速的原子 schema 变更AddField、CheckConstraint等留在常规atomic True迁移中只有慢速、非原子的索引构建单独成迁移。这保证大多数 schema 变更仍是事务性的并在并发构建失败需要重试时限制爆炸半径。要让UniqueConstraint以并发构建的索引为支撑用SeparateDatabaseAndState包裹原始 SQLdatabase_operations用CONCURRENTLY创建索引并通过ALTER TABLE ... ADD CONSTRAINT ... UNIQUE USING INDEX挂上约束state_operations持有对应的AddConstraint(UniqueConstraint(...))让 Django 的模型状态保持同步。总是提供reverse_sql用DROP INDEX CONCURRENTLY IF EXISTS/DROP CONSTRAINT IF EXISTS。参考示例saleor/page/migrations/0030_slug_translation_unique_constraint.pyatomic FalseSeparateDatabaseAndState先CREATE UNIQUE INDEX CONCURRENTLY uniq_lang_slug_pagetransl ON page_pagetranslation (language_code, slug)再ALTER TABLE ... ADD CONSTRAINT ... UNIQUE USING INDEXstate 侧配UniqueConstraint(fields(language_code, slug))。saleor/app/migrations/0040_appextension_identifier_unique_constraint.py从原子的 0039_appextension_identifier_and_more.py 中拆分出来的独立并发迁移为app_appextension (app_id, identifier)构建并发唯一索引。十三、代码风格补充pk与 docstring13.1 Django 正确性用pk而不是id不要使用 Django 模型的id数据库字段引用对象 ID 时用pk。Dont:book Book.objects.get(id1) id book.idDo:book Book.objects.get(pk1) id book.pk13.2 优先使用 docstring 而不是注释描述行为时用 docstring 而不是注释def foo(): Doc string十四、Agent 技能约定docs/agentsIssue tracker本项目已选择退出 issue 跟踪——skills 不得在任何地方创建 issues、PRDs 或 triage 记录把发布到 issue tracker的步骤视为不适用。见 docs/agents/issue-tracker.md。Triage labels不适用——没有 issue 队列因此 triage 状态机与标签词汇表未使用。见 docs/agents/issue-tracker.md。Domain docs单上下文——仓库根只有一个CONTEXT.md加 docs/adr 目录。见 docs/agents/domain.md。十五、性能原则避免不必要地宽泛调用Model.save()和Model.refresh_from_db()。改用只选特定字段的元组例如invoice.refresh_from_db(fields(id,))或invoice.save(update_fields(number,))。避免对同一批对象多次迭代多个 O(N) 操作。Dont:assigned_ids [giftcard.pk for giftcards in assigned_cards] deactivated_ids [giftcard.pk for giftcards in assigned_cards if giftcard.active]Do:assigned_ids [] deactivated_ids [] for giftcard in giftcards: assigned_ids.append(giftcard.pk) if giftcard.active is True: deactivated_ids.append(giftcard.pk)十六、数据库访问共享 Postgres 是扩展瓶颈每个 Pod 都打同一个 Postgres查询要又便宜又少禁止循环内逐行查询N1。按连接字段建 dict 做 O(1) 查找或对全部行做单次聚合 /filter(...).exists()。用 bulkqueryset.update()/.delete()替代逐行 save/delete无界 queryset 用.iterator()而不是list()。只取需要的列.values(...)/.values_list(pk, flatTrue)当你只用这些时不要取完整模型实例。用qs.exists()而不是qs.count() 0。不要在热路径auth、resolvers上增加每请求 DB 查询——复用 dataloader。不要绕过 dataloader 去拿最新值数据本来就是副本延迟的多一次查询换不来真正的强一致。保持select_for_update事务简短持锁期间不要list()无界集合或逐行查询。不要仅仅为了调试/可观测性而持久化行——用日志写操作会复制到每个读副本。十七、数据最小化Data minimization不要把用户 PII如 email反规范化进其他表或事件参数——只存引用 id。任何新的用户引用/PII 字段都要在 anonymizer app 注册并确保 PII 随用户删除而删除on_deletePROTECT deactivate而不是会留下孤儿数据的SET_NULL。十八、文件上传与下载安全把任何下载的文件或 URL 视为不可信Content-Type头可伪造用 magic bytes 确认真实类型强制允许格式白名单拒绝 SVG 及其他可执行/矢量格式强制最大大小并在读取响应体前检查 HTTP 状态码。验证文件扩展名与检测到的文件类型一致。Saleor 中正确的文件校验实现见 saleor/graphql/core/validators/file.py。结语Saleor 的这份开发规范并非零散的风格偏好而是一整套服务于横向扩展、多租户、高并发架构的工程方法论部署模型决定了状态必须外置、操作必须原子、后台工作必须后置测试规范保证了并发与权限逻辑可被精确验证并发模式汇总表给出了可直接套用的原子原语数据迁移规范则解决了大规模 schema 变更的停机风险。对于任何要在 Saleor 上做二次开发或贡献代码的工程师按本文的规范编写代码就是在为这个 headless commerce 平台的生产级稳定性添砖加瓦。【免费下载链接】saleorSaleor Core: the high performance, composable, headless commerce API.项目地址: https://gitcode.com/gh_mirrors/sa/saleor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Saleor 开发规范深度解读：横向扩展架构下的并发、测试与数据迁移实战指南
Saleor 开发规范深度解读横向扩展架构下的并发、测试与数据迁移实战指南【免费下载链接】saleorSaleor Core: the high performance, composable, headless commerce API.项目地址: https://gitcode.com/gh_mirrors/sa/saleor本文基于仓库根目录下的 CLAUDE.md 系统梳理 Saleor Core高性能、可组合的 headless 商务 API的工程开发规范覆盖部署模型心智模型、多租户站点解析、GraphQL 约定、测试规范、并发与线程安全模式、数据迁移最佳实践、性能与数据库访问原则等核心主题。读完本文你将理解为什么 Saleor 要求把所有操作都当作与自身并发执行并掌握F()表达式、select_for_update、on_commit、update_or_create等并发原语在真实项目中的落地用法以及一套可直接复制的 pytest 断言与 GraphQL 授权测试范式。一、部署模型一切规范的第一性原理Saleor 面向**水平扩展horizontal scale-out**部署大量完全相同的 Kubernetes PodWeb 服务 Celery worker并发运行在负载均衡器之后共享一个 Postgres带读副本与共享的 Redis/Valkey 消息代理。任何一个请求或任务可能落在任意 Pod 上同一份代码同时运行在多个进程中。这个部署模型是阅读本文所有规范的起点CLAUDE.md 强调写代码时要时刻牢记这一点进程内/本地状态不能作为事实来源source of truth。永远不要依赖模块级全局变量、内存缓存或上一个请求设置的值会在下一个请求存在这样的假设——下一个调用可能落在不同的 Pod 上。共享状态只存在于 Postgres 与 Valkey/Redis 中。假设每个操作都会与自身并发执行。使用原子数据库操作F()、update_or_create、select_for_update替代先检查后执行check-then-act。绝不阻塞请求/worker 线程。请求处理代码中禁止time.sleep()和无界while True重试循环——一个被阻塞的线程会浪费 Pod 上有限的 worker 槽位。使用有界重试并把等待逻辑交给后台任务。在事务提交之后调度后台工作。使用on_commit/call_event而不是在开启的事务里裸调.delay()否则任务可能在别的 Pod 上先于事务提交运行从而读不到它需要的行。保护共享数据库。它是扩展的瓶颈避免 N1 查询和循环内的逐行写入用 bulk保持select_for_update事务简短只取需要的列不要在热路径上增加每请求查询——复用 dataloader。让任务幂等且可安全重试——broker 可能多次投递同一个任务Pod 可能在运行途中被杀掉或重新调度。二、多租户永远不要依赖settings.SITE_ID一个部署可能用一个进程服务多个站点且SITE_ID可能未设置或与开发环境不同。禁止直接读取它来识别当前站点也禁止假设只有一个站点。用Site.objects.get_current()解析站点GraphQL 层则用get_site_promisedataloader。两者在配置了SITE_ID时按它选站点未配置时回退到请求主机——直接依赖settings.SITE_ID会把所有租户静默折叠成一个。任何按站点隔离的东西都用解析出的站点来作为 key而不是常量。缓存 key、模块级 dict 或查询过滤器若把单个站点写死会让一个租户读到另一个租户的数据。不要假设只有一行。Site.objects.get()在第二个站点出现时就会崩溃要按解析出的站点过滤。从源码看Site.objects.get_current()读取的是打补丁后的进程级全局缓存THREADED_SITE_CACHE见 saleor/site/patch_sites.py该文件将django.contrib.sites的SiteManager.get_current、clear_cache、get_by_natural_key替换为线程安全实现threading.Lock保护按SITE_ID或请求 host 查库并缓存。关键细节该缓存永远不会被Site/SiteSettings的保存操作失效因此同一进程中写入后通过它读到的值可能是陈旧的——这就是下文编写测试一节强调断言方式的原因。测试中需要显式Site.objects.clear_cache()才能破坏缓存。三、GraphQL 约定版本化与权限API 版本化先查看最近的 git tag 确定当前分支基于哪个版本例如 3.22 tag在此基础上新字段的描述需要加上ADDED_IN_{VERSION}子句。GraphQL 权限用PermissionsField描述字段的访问限制而不是在解析器内部手写权限判断。四、架构原则显式调用拒绝隐式信号不要使用 Django signals如post_save、pre_delete、receiver。应当从触发逻辑的代码处显式调用相关逻辑而不是通过信号处理器串联。唯一例外是用于触发数据迁移任务的post_migrate信号见数据迁移一节。五、代码风格使用全局 import 语句所有 import 放在文件顶部不要放在函数、方法或其他局部作用域内。六、测试规范6.1 在 git worktree 中运行测试当 Saleor 被检出在 git worktree 中时优先在该 worktree 的容器内运行操作测试、管理命令等而不是在宿主机上。宿主机的服务可能属于另一个 worktree直接在宿主机运行会打到错误的数据库/缓存或与其他 worktree 的栈冲突。使用.worktree-container/compose.sh包装脚本它指向当前 worktree 的隔离栈在saleor服务内运行命令.worktree-container/compose.sh up -d # 启动当前 worktree 的栈 .worktree-container/compose.sh exec saleor pytest --reuse-db saleor/path/to/test_file.py .worktree-container/compose.sh exec saleor python manage.py migrate在 worktree 中工作时不要直接在宿主机上运行这些操作。6.2 运行测试用pytest运行测试。附加--reuse-db参数以复用测试数据库、加速测试。通过传测试文件路径作为参数来选择要运行的测试。在开发容器内依赖是系统级安装的直接运行pytest即可——无需激活虚拟环境。在宿主机容器外上先进入虚拟环境再执行测试。6.3 编写测试的通用规则使用 given/when/then 结构清晰表达给定条件-执行动作-断言结果。使用pytestfixtures 做 setup 和 teardown测试套件在文件中扁平声明不要包在类里。优先用 fixture 而不是 mock。fixtures 通常在tests/fixtures目录下是pytest.fixture装饰的函数当测试对象没有现成 fixture 时新建一个并可用 factory 向 fixture 传参。断言实际返回值而非不为 None。例如assert email is ab.com而不是assert email is not None。断言 GraphQL 错误时同时断言错误消息。断言 Enum 时在测试文件中 import 枚举用assert error.code MyEnum.SOME_ERROR.name而不是字符串比较。如果已有对值的引用避免断言裸值。例如创建了实体并断言响应中返回该实体时用实体的字段与响应字段比较而不是写死字面量。构建 GraphQL 查询变量时使用枚举值而非普通字符串{field: TransactionSortField.CREATED_AT.name}而不是{field: CREATED_AT}。不要给测试名写多余前缀文件叫test_my_mutation.py时测试应命名test_with_x_y而不是test_my_mutation_with_x_y。测试 GraphQL 预期错误场景时总是断言 errors 列表的期望长度。准备测试数据时把值提取到变量并在断言中复用不要在 setup 和 assertion 之间重复字面量。比较 JSON payload 时用json.loads()转成 dict 再比较不要用json.dumps()比较序列化字符串——key 顺序变化会让字符串比较失效。期望恰好一行时用Model.objects.get()而不是.first()加is not None断言。get()同时完成取数与恰好存在一行的断言可以省掉单独的 not-None 检查。存在性检查优先用qs.exists()而不是qs.count() 0/! 0——exists()比COUNT便宜。示例assert checkout.lines.exists() is False。相似场景用pytest.mark.parametrize参数化而不是复制粘贴几乎相同的测试体。用带_case前缀的字符串参数标识每个用例而不是用ids参数——描述紧挨着数据更易读易改pytest.mark.parametrize( (_case, value), [ (omitted, {}), (empty_list, {lines: []}), ], ) def test_something(_case, value): ...当 mutation 输入是可选的/nullable 时除了省略该字段还要显式测试null值——两者是不同的输入可能被区别处理。永远不要假设缓存是冷的——或热的。缓存是进程级全局的跨测试存活因此缓存由谁预热取决于测试顺序。要断言写入已落地就回读行obj.refresh_from_db(fields(...))而不是通过Site.objects.get_current()这类缓存访问器要统计查询数先预热缓存得到稳态否则显式破坏缓存Site.objects.clear_cache()。一个只因缓存恰好为空才通过的测试会在某个无关调用者预热缓存后立刻失败。6.4 精确断言review 评论的第一大来源CLAUDE.md 专门开辟一节强调精确断言要点如下断言精确的值与数量绝不断言边界或仅仅不存在。不要写len(x) 0、 1、assert not errors、call_count 1——断言精确的计数、枚举或对象才能捕获意外多余的返回结果。期望一个错误时断言恰好一个、完整错误消息以及它的path/字段。在否定/权限拒绝测试中还要断言副作用确实没有发生——余额未变、SomeEvent.objects.exists() is False、对象未被修改。仅有错误断言并不能证明操作被拦截。绝不断言测试自身刚设置并保存的值——那验证的是 ORM 而不是被测代码。要么设置状态要么断言它二者取其一。绝不要把可变输入在传给可能修改它的代码之后又对它自身做断言——这种比较永远不可能失败。应设计为期望值独立于输入用字面量或单独 fixture 构造。copy.deepcopy仅在确有修改风险且没有更干净方案时作为最后手段——它很昂贵涉及 pickle无法扩展到大型套件。优先用真实 fixture/token 而不是 mock 内部机制。用真实的create_access_token*工厂或 pyjwt 生成真正有效/无效的 token而不是 mockget_decoded_token返回固定 payload——否则真实故障不会被发现。修复必须附带回归测试端到端复现实际报告的 bug并针对真正失败的字段/path而不是只测新 helper 的 happy path。把前置条件编码为断言而非注释在调用前写assert flag is True这样默认值一旦变化测试会快速失败。期望恰好一行时用Model.objects.get()它断言唯一性而不是.first() not-None 检查用qs.exists()而不是qs.count() 0。不要硬编码不存在的主键如99999——--reuse-db可能重新分配它。改用负数 id 如-1Postgres 没有无符号整数类型负数永远不可能匹配真实行还省掉一次max(pk)查询。测试中的 URL 使用保留测试域名example.com、*.test绝不使用真实域名。IP 使用网络地址使其无法触达真实服务器例如8.8.8.0或1.1.1.0。参数化近似相同的测试体保持每个测试最小化删掉无关 setup按断言的确切行为命名避免双重否定。6.5 GraphQL 授权测试什么时候应该写授权测试新增受限字段restricted field时新增受限类型restricted type时新增受限 mutation 或 query 时新增公开mutation 或 query 时——这确保我们显式确认新 query/mutation 不需要任何授权。写授权测试时必须全部做到以下几点覆盖所有客户端类型未认证Unauthenticated无权限用户非 staff 用户缺少所需权限的 staff 用户拥有正确权限的 staff 用户使用参数化测试不要创建多个独立测试。绝不仅仅依赖错误消息判断访问被拒绝——因为即使报错仍可能返回了受限数据或执行了操作。因此总是确保受限数据没有出现在 HTTP 响应中若是 GraphQL mutation总是确保数据没有被修改若是 GraphQL mutation总是确保外部调用webhooks、API 调用、邮件等没有被触发。标准模板如下可直接复用pytest.mark.parametrize( (_case, client_fixture, permission_fixture, is_allowed), [ ( Unauthenticated user should be rejected, api_client, None, False, ), ( Authenticated unprivileged user (non-staff) should be rejected, user_api_client, None, False, ), ( Authenticated user w/o the permission should be rejected, staff_api_client, None, False, ), ( Authenticated user w/ the correct permission should be allowed, staff_api_client, permission_manage_settings, True, ), ] ) def test_authorization( request, client_fixture: str, permission_fixture: str | None, is_allowed: bool, ): client request.getfixturevalue(client_fixture) if permission_fixture: perm request.getfixturevalue(permission_fixture) if client.app: client.app.permissions.add(perm) elif client.user: client.user.user_permissions.add(perm) else: raise AssertionError(Couldnt add the permission) # shouldnt occur response client.post_graphql(query, variables) if is_allowed: content get_graphql_content(response) assert content[data] ... else: assert_no_permission(response) content get_graphql_content_from_response(response) assert content[data] ...七、代码结构与分层业务逻辑放在领域层domain layerGraphQL mutation 只做输入校验与编排。token 如何构造/剥离、余额如何变化等属于领域模块而不是 mutation。非 GraphQL 代码绝不能从graphql层 import。如果领域层的任务或模型需要某个 helper把 helper 移到领域模块中由 GraphQL 层从领域层导入——绝不允许反向。在数据产生点做校验fail-fast子实体在与父实体同一个 clean 步骤中清理——不要把同一实体的清理拆到不同阶段。只传函数真正需要的窄值单个 flag 或已取到的 setting而不是让函数去拆解宽泛的info/SiteSettings上下文对象。宽参数会隐藏真实依赖迫使调用方反复穿线上下文。优先用类型化结构dataclass/NamedTuple而不是即兴的dict集合并保持函数返回形状同质不要在同一个列表中混入异构实体。把分支多或冗长的逻辑提取成具名 helper而不是用注释叙述一个约 100 行的函数。先执行短路/全局检查再去取数据。删除重构产生的死代码未使用的校验器、仅为满足 mypy 而加的不可达守卫。要从源头修正类型而不是在错误的上游注解外面包一层cast()。八、命名规范禁止晦涩缩写gc_brand应写成gift_card_brand新字段名与已有的同类字段对齐。使用正向布尔命名——delivery_changed而不是not delivery_unchanged。任何门控废弃deprecated行为的 flag 或字段加legacy_前缀让废弃从名字上一眼可见。名字要反映返回值/含义与正确的领域而不是偶然的调用点不要把格式假设写进名字code 可能含字母时用last_chars而不是last_digits。九、错误处理捕获具体异常绝不捕获裸Exception或DatabaseError。需要时捕获整个 provider 异常族BotoCoreError和ClientErrorGoogleAPIError同时覆盖 4xx和5xx以及最窄的数据库异常OperationalError而不是基类。区分可重试失败网络错误、HTTP 5xx与终态失败4xx、坏数据、ValueError——绝不重试无法自愈的失败终态失败立即返回/中止让无效工作快速排空。读取 HTTP 响应体之前先检查状态码出站网络调用总是传显式 timeout。错误消息必须说明实际原因not found 与 wrong type 不同对列表输入要归因到具体的行/字段配置/店铺级失败用通用非字段错误。传播捕获到的原始消息而不是重写它。校验引用的 GraphQL ID 并返回干净的错误——不要让坏/不存在的 id 暴露出晦涩的 Cannot return null for non-nullable field 崩溃。十、Webhooks 与事件分发分发 webhook 事件时总是用saleor.core.utils.events中的call_event而不是直接调用 manager 方法。Bad:manager.product_variant_discounted_price_updated(price_info, webhookswebhooks)Good:from saleor.core.utils.events import call_event call_event(manager.product_variant_discounted_price_updated, price_info, webhookswebhooks)从实现看saleor/core/utils/events.pycall_event的核心逻辑是检查当前连接是否处于原子事务块connection.in_atomic_block若在事务中则通过transaction.on_commit延迟到提交后执行否则立即执行——这正好呼应调度后台工作在事务提交之后的部署模型要求确保 webhook 不会在数据提交前被其他 Pod 消费。十一、并发与线程安全模式重点章节Saleor 运行在多个并发执行的 Python 服务中。以下是必须遵循的线程安全模式其中多数对应 CLAUDE.md 末尾的Saleor 使用的模式汇总表。11.1 原子递增模式Atomic Increment永远不要用instance.count 1; instance.save()——这不是原子操作会造成丢失更新lost updates。Bad:existing.count 1 existing.save(update_fields[count])Good - 使用 F() 表达式from django.db.models import F Model.objects.filter(pkexisting.pk).update(countF(count) 1)仓库中的真实案例见 saleor/discount/utils/voucher.pyincrease_voucher_code_usage_value使用code.used F(used) 1再save(update_fields[used])完成优惠券码使用次数递增递减则用Case/When守卫下限used__gt0时才F(used) - 1否则置 0。11.2 避免 Check-Then-Act 竞态永远不要在独立操作中先检查记录是否存在再创建/更新它。Bad:existing Model.objects.filter(appapp, keykey).first() if not existing: Model.objects.create(appapp, keykey, ...) # Race condition: duplicate may be created else: existing.update(...)Good - 使用带唯一约束的 update_or_createobj, created Model.objects.update_or_create( appapp, keykey, # Lookup fields defaults{message: message, updated_at: now} # Fields to update )注意update_or_create要真正安全lookup 字段上必须存在唯一约束。11.3 行级锁select_for_update对于需要在同一行上进行多次读/写的复杂操作使用select_for_updatefrom saleor.core.tracing import traced_atomic_transaction with traced_atomic_transaction(): obj Model.objects.select_for_update().get(pkpk) # Perform operations - row is locked until transaction ends obj.save()模式创建 lock_objects.py 模块参考 saleor/payment/lock_objects.pydef my_model_qs_select_for_update() - QuerySet[MyModel]: return MyModel.objects.order_by(pk).select_for_update(of[self])真实实现中transaction_item_qs_select_for_update正是TransactionItem.objects.order_by(pk).select_for_update(of[self])并且组合出get_order_and_transaction_item_locked_for_update、get_checkout_and_transaction_item_locked_for_update这样的高层 helper一次性锁定 ordertransaction item 或 checkouttransaction item保证支付/退款场景的读写一致性。select_for_update 的坑Queryset 是惰性的——未求值的锁 queryset 不会执行 SQL因此不会加锁而且会静默失败。必须在事务内强制求值list(...)、.get()/.first()或把它作为写操作的__in子查询。with transaction.atomic(): # Bad: lazy - no lock _locked model_qs_select_for_update().filter(pk__inpks) # Good: rows locked, only pks fetched _locked list( model_qs_select_for_update().filter(pk__inpks).values_list(pk, flatTrue) ) Model.objects.bulk_update(objs, [field])用assert_locks_rows_before_writefixturesaleor/tests/fixtures.py测试锁——它捕获代码块的查询并断言在UPDATE之前发出了带确定性ORDER BY的FOR UPDATE查询。从源码看该 fixture 从捕获的 SQL 中筛出含FOR UPDATE的查询和UPDATE开头的查询断言恰好两条、且锁查询在写查询之前、FOR UPDATE确实存在。它直接对 SQL 断言是刻意为之因为漏加锁很难用行为测试发现。11.4 数据库事务用traced_atomic_transaction包裹相关操作from saleor.core.tracing import traced_atomic_transaction with traced_atomic_transaction(): # All operations here are atomic obj1.save() obj2.save()从 saleor/core/tracing.py 的实现看它本质是transaction.atomic()外包一层 OpenTelemetry span名为transaction组件属性为orm因此每个事务都会自动产生可观测的追踪跨度。11.5 在提交后调度后台工作永远不要在开启的事务里用裸.delay()调度 Celery 任务——任务可能在别的 Pod 上先于事务提交启动读到它还看不到的数据而失败。用on_commit或call_event它对 webhook 事件就是这么做的让任务只在提交后触发from django.db.transaction import on_commit with traced_atomic_transaction(): obj.save() on_commit(lambda: my_task.delay(obj.pk))多写/读-改-写操作要包在事务里做到全有或全无并在同一个事务里发出状态变更事件。11.6 绝不阻塞请求/worker 线程请求处理或 worker 代码中禁止time.sleep()和无界while True重试循环——阻塞的线程浪费 Pod 上有限的 worker 槽位。改用有界for attempt in range(n)重试把等待卸载给后台任务。11.7 其他并发规则把allow_writer()的作用域限制到恰好需要它的那条语句with allow_writer(): ...而不是整个任务/函数。在每个 Promise.then()回调内重新建立 DB/writer 上下文——它在上一个上下文退出之后才运行类似 async所以在.then()之前打开的 writer/context 在回调内部已经失效。绝不要用独立的empty()/非空检查去 gate 阻塞式Queue.get()——检查与 get 之间最后一个元素可能被取走导致消费者挂起。用非阻塞 get 或哨兵sentinel。把重负载扇出任务大下载、数千对象路由到专门的、有界队列避免单个 bulk mutation 饿死 worker 或让 Pod OOM。11.8 模式汇总表CLAUDE.md 给出的速查表模块示例均为仓库真实路径PatternModule ExampleWhen to UseF()atomic incrementsaleor/discount/utils/voucher.py计数器更新select_for_updatesaleor/payment/lock_objects.py复杂读-改-写update_or_createsaleor/graphql/attribute/utils/type_handlers.pyUpsert 操作traced_atomic_transactionsaleor/core/tracing.py多操作原子性on_committask scheduling任何分发任务的 mutation仅在提交后触发任务十二、数据迁移Data Migrations12.1 将搜索索引标记为 dirty当新增或修改搜索向量索引逻辑时创建一个数据迁移把所有现存行标记为 dirty让它们被重新索引。结构Task 文件—— 放在saleor/app/migrations/tasks/saleorversion.pyMigration 文件—— 放在saleor/app/migrations/number_name.pyTask 规则用app.task(queuesettings.DATA_MIGRATIONS_TASKS_QUEUE_NAME)和allow_writer()装饰分批处理例如每批 1000 行避免长事务通过 app 的lock_objects模块用select_for_update锁定行再更新使用子查询模式先锁行再在单独查询中 filterupdatewith transaction.atomic(): pks ( model_qs_select_for_update() .filter(pk__inbatch_pks) .values_list(pk, flatTrue) ) Model.objects.filter(pk__inpks).update(search_index_dirtyTrue)末尾还有更多行时用.delay()自链self-chain当批次无返回时停止链式调度。Migration 规则用post_migrate信号在全部迁移完成后触发 Celery 任务不要内联触发在RunPython操作内部连接post_migrate用registry.get_app_config(app)作为 sender总是提供migrations.RunPython.noop作为反向操作。参考示例Task 见 saleor/giftcard/migrations/tasks/saleor3_22.pyMigration 见 saleor/giftcard/migrations/0023_mark_gift_cards_search_vector_as_dirty.py。后者在RunPython中定义on_migrations_complete以registry.get_app_config(giftcard)为 sender 连接post_migrateweakFalse迁移完成后.delay()触发批量打 dirty 标记的任务。12.2 并发添加索引 / 唯一约束同步构建索引默认的AddIndex/AddConstraint在构建期间会对整张表持有ACCESS EXCLUSIVE锁阻塞读写。在大表上这意味着显著的停机时间。应改用CONCURRENTLY构建索引不阻塞并发流量。规则CREATE INDEX CONCURRENTLY不能在事务内运行所以创建它的迁移必须设置atomic False。让并发索引独占一个迁移。快速的原子 schema 变更AddField、CheckConstraint等留在常规atomic True迁移中只有慢速、非原子的索引构建单独成迁移。这保证大多数 schema 变更仍是事务性的并在并发构建失败需要重试时限制爆炸半径。要让UniqueConstraint以并发构建的索引为支撑用SeparateDatabaseAndState包裹原始 SQLdatabase_operations用CONCURRENTLY创建索引并通过ALTER TABLE ... ADD CONSTRAINT ... UNIQUE USING INDEX挂上约束state_operations持有对应的AddConstraint(UniqueConstraint(...))让 Django 的模型状态保持同步。总是提供reverse_sql用DROP INDEX CONCURRENTLY IF EXISTS/DROP CONSTRAINT IF EXISTS。参考示例saleor/page/migrations/0030_slug_translation_unique_constraint.pyatomic FalseSeparateDatabaseAndState先CREATE UNIQUE INDEX CONCURRENTLY uniq_lang_slug_pagetransl ON page_pagetranslation (language_code, slug)再ALTER TABLE ... ADD CONSTRAINT ... UNIQUE USING INDEXstate 侧配UniqueConstraint(fields(language_code, slug))。saleor/app/migrations/0040_appextension_identifier_unique_constraint.py从原子的 0039_appextension_identifier_and_more.py 中拆分出来的独立并发迁移为app_appextension (app_id, identifier)构建并发唯一索引。十三、代码风格补充pk与 docstring13.1 Django 正确性用pk而不是id不要使用 Django 模型的id数据库字段引用对象 ID 时用pk。Dont:book Book.objects.get(id1) id book.idDo:book Book.objects.get(pk1) id book.pk13.2 优先使用 docstring 而不是注释描述行为时用 docstring 而不是注释def foo(): Doc string十四、Agent 技能约定docs/agentsIssue tracker本项目已选择退出 issue 跟踪——skills 不得在任何地方创建 issues、PRDs 或 triage 记录把发布到 issue tracker的步骤视为不适用。见 docs/agents/issue-tracker.md。Triage labels不适用——没有 issue 队列因此 triage 状态机与标签词汇表未使用。见 docs/agents/issue-tracker.md。Domain docs单上下文——仓库根只有一个CONTEXT.md加 docs/adr 目录。见 docs/agents/domain.md。十五、性能原则避免不必要地宽泛调用Model.save()和Model.refresh_from_db()。改用只选特定字段的元组例如invoice.refresh_from_db(fields(id,))或invoice.save(update_fields(number,))。避免对同一批对象多次迭代多个 O(N) 操作。Dont:assigned_ids [giftcard.pk for giftcards in assigned_cards] deactivated_ids [giftcard.pk for giftcards in assigned_cards if giftcard.active]Do:assigned_ids [] deactivated_ids [] for giftcard in giftcards: assigned_ids.append(giftcard.pk) if giftcard.active is True: deactivated_ids.append(giftcard.pk)十六、数据库访问共享 Postgres 是扩展瓶颈每个 Pod 都打同一个 Postgres查询要又便宜又少禁止循环内逐行查询N1。按连接字段建 dict 做 O(1) 查找或对全部行做单次聚合 /filter(...).exists()。用 bulkqueryset.update()/.delete()替代逐行 save/delete无界 queryset 用.iterator()而不是list()。只取需要的列.values(...)/.values_list(pk, flatTrue)当你只用这些时不要取完整模型实例。用qs.exists()而不是qs.count() 0。不要在热路径auth、resolvers上增加每请求 DB 查询——复用 dataloader。不要绕过 dataloader 去拿最新值数据本来就是副本延迟的多一次查询换不来真正的强一致。保持select_for_update事务简短持锁期间不要list()无界集合或逐行查询。不要仅仅为了调试/可观测性而持久化行——用日志写操作会复制到每个读副本。十七、数据最小化Data minimization不要把用户 PII如 email反规范化进其他表或事件参数——只存引用 id。任何新的用户引用/PII 字段都要在 anonymizer app 注册并确保 PII 随用户删除而删除on_deletePROTECT deactivate而不是会留下孤儿数据的SET_NULL。十八、文件上传与下载安全把任何下载的文件或 URL 视为不可信Content-Type头可伪造用 magic bytes 确认真实类型强制允许格式白名单拒绝 SVG 及其他可执行/矢量格式强制最大大小并在读取响应体前检查 HTTP 状态码。验证文件扩展名与检测到的文件类型一致。Saleor 中正确的文件校验实现见 saleor/graphql/core/validators/file.py。结语Saleor 的这份开发规范并非零散的风格偏好而是一整套服务于横向扩展、多租户、高并发架构的工程方法论部署模型决定了状态必须外置、操作必须原子、后台工作必须后置测试规范保证了并发与权限逻辑可被精确验证并发模式汇总表给出了可直接套用的原子原语数据迁移规范则解决了大规模 schema 变更的停机风险。对于任何要在 Saleor 上做二次开发或贡献代码的工程师按本文的规范编写代码就是在为这个 headless commerce 平台的生产级稳定性添砖加瓦。【免费下载链接】saleorSaleor Core: the high performance, composable, headless commerce API.项目地址: https://gitcode.com/gh_mirrors/sa/saleor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考