yeuimu
c545c1b87d
死锁根因:sds-refresh-pending 与 label-ocr 整点并发写同一分区 Orders_2026_08, 互等对方未提交事务(ShareLock)。24h 内同类死锁 7 起。 - sync_orders_and_goods 所有写事务先取同一把 PG advisory 锁(事务级),单锁无环,死锁结构性消除 - 同步按批(1000单)独立提交,持锁窗口从整个 pending 集缩到单批 - 连接预设 lock_timeout=15s,DeadlockDetected/LockNotAvailable 整事务重试 3 次(退避1/3/7s) - label-ocr 的 update_ocr_batch/update_ocr_fields 走同一把锁+重试 - sds/riin refresh_pending 的 upsert_records 加任务级重试(retries=3)
136 lines
8.1 KiB
Markdown
136 lines
8.1 KiB
Markdown
# 系统设计架构
|
|
|
|
本项目把两个外部业务系统(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 的"售后"。
|