KEEL · 龙骨 · A CURRICULUM FOR THE AI ERA
05 · Pydantic v2、StrEnum 与注册表工厂 — keel 龙骨
三个「让数据自己管好自己」的写法：Pydantic v2 的类型化边界（验证+序列化）、StrEnum 让枚举值直接可比可序列化、注册表字典把「新增一种类型」变成「加一行」。
三个「让数据自己管好自己」的写法：Pydantic v2 的类型化边界（验证+序列化）、StrEnum 让枚举值直接可比可序列化、注册表字典把「新增一种类型」变成「加一行」。
一、Pydantic v2：跨越进程边界的类型安全
1.1 入口验证：model_validate 重建对象
# 任务载荷在消息队列里走了一遭，只剩 JSON dict
request = JobRequest.model_validate(request_data or {})
dict 的字段拼错名、类型不对，都要等到运行到那一行才炸；model_validate 一步把 dict 变回 Pydantic 模型——字段名错 → ValidationError 当场指认；类型不对 → 自动转换或报错。之后所有代码都在类型保护下工作。
1.2 出口序列化：model_dump(mode="json")
request_data = prepared_request.model_dump(mode="json") # JSON 模式序列化
mode="json" 是细节考点：默认 model_dump() 会把 datetime/UUID/枚举保留成 Python 对象（dump 出来的 dict 还不是纯 JSON）；mode="json" 把它们全部转成 JSON 兼容类型。凡是「要跨进程/要写队列/要进 HTTP 响应」的 dump，一律 mode="json"——漏写的症状很典型：本地跑得好好的，一进队列就 Object of type datetime is not JSON serializable。
1.3 还有一层职责：Pydantic 模型当「数据契约文档」
class JobRequest(BaseModel):
job_type: str = "export"
payload: dict = {}
priority: int = 5
is_test: bool = True
读代码时先看模型定义，就知道这次运行的全部输入。模型即文档、即验证、即序列化配置——这是 Pydantic 在大型项目里的真正价值。
二、StrEnum：让枚举「是字符串」而不是「像字符串」
from enum import StrEnum
class JobStatus(StrEnum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
为什么不用普通 Enum / 常量 / 字符串字面量：
# 字符串字面量的问题
if state["status"] == "runing": # 拼错了，字符串不报错，静默永远为 False
...
# 普通 Enum 的问题
class Status(Enum):
RUNNING = "running"
state["status"] == Status.RUNNING # 存进队列再读出来是 str "running"，比较为 False！
json.dumps(Status.RUNNING) # 不能直接序列化
StrEnum（Python 3.11+）两边都解决：成员就是 str 子类——
JobStatus.RUNNING == "running" # True！从队列读回的字符串直接比
redis.hset(key, "status", status) # 直接可写（str 语义）
json.dumps({"status": JobStatus.RUNNING}) # 可序列化为 "running"
可迁移规则：枚举值要进 JSON/数据库/和前端比对 → StrEnum（JSON 侧也常用 str, Enum 多继承的老写法）；纯进程内逻辑枚举 → 普通 Enum 够用。
三、注册表 + 工厂：新增一种类型 = 改一处
3.1 Worker 角色注册表
WORKER_DEFS: dict[str, WorkerDef] = {
"export": WorkerDef(name="export", queue_name="q_export", functions=[...]),
"import": WorkerDef(name="import", queue_name="q_import", functions=[...]),
"unified": WorkerDef(name="unified", queue_name="q_unified", functions=[...]),
}
新增第四种 Worker = 在字典里加一个条目，拉起逻辑一行不改——未知名还有启动前校验兜底（取用时抛 ValueError）。
3.2 任务 Provider 工厂
class JobProviderFactory:
PROVIDER_MAP = {"export": ExportProvider, "import": ImportProvider} # 名字 → 类
@classmethod
def create(cls, provider_name: str, **kwargs):
provider_cls = cls.PROVIDER_MAP.get(provider_name)
if provider_cls is None:
raise ValueError(f"unknown provider: {provider_name}") # 未知类型：启动期报错
return provider_cls(**kwargs)
注册表模式三件套：字典注册（名字→类/定义）→ 统一入口取用（get/create）→ 未知名尽早报错。对比 if provider == "export": ... elif ... 的 if 链：注册表把「新增类型的改动范围」压缩到字典一行，且所有类型处理路径结构一致。
if 链什么时候也可以：类型只有两三种、逻辑彼此差异巨大、不会新增——if 链更直白。注册表的本质收益是「开放扩展、结构统一」，类型会增长的场景才值得。
四、可迁移的套路
- 跨进程数据 → Pydantic 模型：入口
model_validate，出口model_dump(mode="json")，两个动作成对出现； - 进 JSON/DB 的枚举 → StrEnum，杜绝「字符串拼错静默失效」和「枚举与字符串比较恒 False」；
- 类型会增长的分支逻辑 → 注册表字典 + 工厂 + 未知名报错；「加一行支持新类型」是检验标准；
- 三者合起来是一个主题：让「数据正确性」和「类型扩展性」由语言/框架机制保证，而不是靠人肉小心。
↓ 下一步：06 章 · 动态分发与反射：把方法存成字符串，运行时再映射回真实 callable——把 Python「按名字在运行时找回并调用方法」的反射能力讲透。