inkreach-prefect/docs/dev/systematic-design.md

系统设计架构

本项目把两个外部业务系统(RIIN 分销、SDS POD)的 REST API 封装为可复用的 Python 客户端, 再用 Prefect 编排成 ETL 工作流(REST API → PostgreSQL)。

设计理念:高内聚、低耦合、关注点分离

核心原则贯穿整个项目:

1. 高内聚 —— 每个模块只承担一个职责。api/riin/ 只管 RIIN,api/sds/ 只管 SDS, flows/ 只管编排。一个模块内部的代码都围绕同一件事。
2. 低耦合 —— 客户端层(api/)零 Prefect 依赖,可独立使用、独立测试; 编排层(flows/)只通过客户端的公开方法调用它。两层之间是单向依赖。
3. 不过度抽象 —— 仅在行为稳定且重复时才抽取共享代码。当前 2 个系统,用组合式 helper 而非强制基类;等出现 3+ 个行为高度一致的源,再考虑抽象 Protocol 接口(YAGNI)。

分层结构

inkreach-prefect/
├── api/                        # 客户端层:纯 httpx + pydantic,不 import prefect
│   ├── _common/                #   共享 helper(组合,非强制基类)
│   │   ├── http.py             #     TokenHttpClient:token 注入 + 401 自动重登
│   │   ├── cache.py            #     TTLCache:主数据就近缓存(随 client 生命周期)
│   │   ├── concurrency.py      #     run_concurrent:线程池 + 并发上限
│   │   └── batch.py            #     parse_id_text / split_lines / chunk
│   ├── riin/                   #   RIIN 分销平台
│   │   ├── client.py           #     RiinDisClient:认证/订单/店铺/辅助主数据
│   │   └── models.py           #     pydantic schema + 枚举
│   └── sds/                    #   SDS POD 平台
│       ├── client.py           #     SdsMngClient:认证/订单/产品/物流
│       └── models.py
├── flows/                      # 编排层:依赖 api/,使用 prefect
│   ├── riin_orders_etl.py      #   @task 封装 → @flow: 抽取→转换→落 CSV(后接 PG)
│   ├── sds_orders_etl.py
│   └── shared/load.py          #   共享 task:凭据加载 / CSV / PG 写入
└── pyproject.toml              # build-system=hatchling,packages=["api"]

层次边界(关键)

层	依赖	职责
api/ 客户端层	httpx、pydantic(禁止 import prefect)	REST 封装、认证、分页、schema
flows/ 编排层	prefect、api/、pandas	任务编排、重试、缓存、并发、调度、密钥

客户端层绝不反向依赖 Prefect。这样客户端能在 ETL 之外被复用(脚本、其他服务、单测)。 缓存/重试/并发这些编排能力留在 flows 层,客户端只保留自身需要的就近缓存(TTLCache) 和内部并发(run_concurrent,用于同一次调用内的多页并发)。

客户端层设计要点

TokenHttpClient 基类(_common/http.py)

两个系统共享"token 注入 + 401 自动重登"这一稳定行为,故提取为基类(继承)。 差异通过构造参数由子类配置:

系统	base_url	token 头	固定头	登录
RIIN	tshirt.riin.com	Authorization	X-Time-Zone / X-Requested-With	username + MD5(password)
SDS	pod-api.sdspod.com	access-token	无	username + password + 公司名

子类只需实现 _login(),其余 token 生命周期(缓存/失效/重试)由基类统一处理。

模式还原(忠实移植自 TS SDK)

• 分页三态:单页 / limit(取首页 N 条)/ all(全量并发分页)
• 批量切分:RIIN 订单 OID/快递单号 >500 自动分批;SDS 订单 keywords >200 自动分批
• 就近缓存:主数据(店铺/工厂/客户/统计)用 TTLCache 缓存 ~5 分钟
• 并发:多页/多批用 run_concurrent(线程池,默认 10)

Schema

每个系统用 pydantic v2 建模参数与返回(extra='allow' 不丢字段)。

• RIIN 全 camelCase → alias_generator=to_camel。
• SDS 订单/产品 camelCase;物流域混用 snake_case,对 snake_case 字段显式 Field(alias=...)。 注意:响应字段名避免用 Python 内置名(如 list 会与 list[Order] 注解冲突,改用 records + alias="list")。

编排层设计要点(对照 Prefect 文档)

TS SDK 概念	Prefect 对应	位置
401 自动重试	@task(retries=3, retry_delay_seconds=[2,5,15])	flows/*.py 的 extract task
主数据缓存	task 级 cache_key_fn(按需启用)	flows
并发限制器	Prefect 并发限制 / .submit()+.map()	flows
凭据	Prefect Secret Block(加密)	flows 内部读取

任务粒度遵循 Prefect"小而可观测"原则:extract / transform / load 各为独立 task。 凭据在 extract task 内部读取 Secret,不作为参数传递,避免泄露到日志/UI。

密钥管理

• PostgreSQL:SqlAlchemyConnector Block,名 pg-inkreach-dmp。
• RIIN / SDS:JSON 存入 Secret Block,名 riin-credentials / sds-credentials, 值为 {"username":"...","password":"..."}。
• 本地开发备用:.env(已 gitignore)。

扩展性:如何接入第三个系统

加法式,不碰现有代码(开闭原则):

1. api/<新系统>/ 新增 client.py(继承 TokenHttpClient)+ models.py。
2. flows/<新系统>_orders_etl.py 新增一个 flow,复用 flows/shared/load.py。
3. 注册该系统的凭据 Secret Block。

若新系统行为与现有高度一致,可白嫖 api/_common/ 的 helper;差异大就自包含实现。 暂不强制统一 Connector 接口 —— 系统异质(REST 不同认证 vs DB vs 文件),强行抽象会变成漏抽象。 等行为趋同后再抽 Protocol。

同步任务层(flows/riin、flows/sds)

对齐 inkreach-bi-auth-mng 的 sync tasks,每系统 6 个,效果一致:

任务	RIIN	SDS	说明
时间窗口同步	order_sync	order_sync	按时间范围(+筛选)拉取
增量同步	incremental_sync	incremental_sync	从 DB MAX(orderedAt) 增量,无参
近期同步	recent_sync	recent_sync	近 N 天
刷新未完结	refresh_pending	refresh_pending	DB 未完结订单号批量刷新
指定单号/100前缀	by_orders	refresh_100_prefix	RIIN 按单号;SDS 替换 100 前缀暂存
数据核验	verify	verify	逐天比对远程/本地计数并修复差异

共享编排层(flows/shared)

• mappers.py:RIIN/SDS 原始订单 → 统一 OrderUpsertInput / GoodsUpsertInput,含状态/平台/快递来源/国家归一映射。
• db.py:SyncDB.sync_orders_and_goods() —— Orders/Goods 批量 ON CONFLICT upsert,写入前用库中已有 createdAt 覆盖(保证幂等),Goods 写前删除该订单不再出现的行;另有 get_latest_ordered_at / get_pending_orders / get_order_count_by_status / get_local_order_nos / get_local_goods_count / get_orders_with_100_prefix 等查询。
  • 写并发控制(2026-08 死锁修复):所有写 Orders/Goods 的事务(同步 flow 与 label-ocr 的 OCR 回写)先取同一把 PG advisory 锁(DB_WRITE_LOCK_KEY,事务级,commit/rollback 即释放)串行化——单锁无环,写事务之间结构性无死锁;sync_orders_and_goods 按批(ORDER_CHUNK=1000)独立提交,持锁窗口从"整个 pending 集"缩到"单批";连接预设 lock_timeout=15s;对 DeadlockDetected / LockNotAvailable 整事务自动重试 3 次(退避 1/3/7s),另有 flow/task 级 retries 兜底。
• clients.py:客户端工厂(读 Secret Block)、fetch_riin/sds_all_pages(顺序分页,页间 500ms,仅对 5xx/429/超时重试)、RIIN 商品并发拉取(并发 20)、SDS 物流渠道字典(24h 缓存)。

关键差异点

• RIIN 订单与商品分离(需 getGoods 逐单拉取,并发 20);SDS 订单内联 items(直接映射,无需额外请求)。
• SDS 同步默认跳过待付款(refresh_pending 除外);RIIN 无此过滤。
• 两个系统时间维度不同:RIIN 用 timeType=3(商家推送)+ startTime/endTime;SDS 用 payDate 或 createdDate。
• verify 用各自的统计接口(RIIN getCount、SDS getStatus)按状态比对,排除 SDS 的"售后"。