Python

Python Callable 类型标注：怎么描述一个 handler 函数？ 副标题： Callable[[dict[str, Any]], Any] 不是一个新类，而是在类型层面说明： 这个参数是一个“能被调用的对象”，它接收什么参数，返回什么结果。 标签： Python / 类型标注 / Callable / Type Alias / Pipeline 适读人群： 正在写回调、handler、pipeline step、插件机制的 Python 开发者 阅读时间： 6 min 背景：为什么 handler 类型看起来这么长？ 在 pipeline runner 里，经常会有这种 handler 表： handlers = { "prepare": prepare, "enrich": enrich, } 每个 handler 都接收同一个运行时上下文： def prepare(context: dict[str, Any]) -> dict[str, bool]: context["prepared"] = True return {"prepared": True} 如果直接给 handler 表写类型，会变成： from collections.abc import Callable, Mapping from typing import Any handlers: Mapping[str, Callable[[dict[str, Any]], Any]] 这一串不难，但读起来吵。它把“workflow name 到 workflow handler 的映射”写成了一段类型语法。 ...

Python 从 dict/list 原型到稳定接口：什么时候该整理类型？ 副标题： 验证想法时先用普通变量和 dict/list 跑通行为；模型稳定后，再把它们整理成函数参数、dataclass、helper 和类型标注。 标签： Python / 原型验证 / 类型标注 / 接口设计 / Pipeline 适读人群： 正在从脚本式代码过渡到可维护模块的 Python 开发者 阅读时间： 7 min 背景：为什么一开始不用把类型设计得很完整？ 写一个新功能时，经常会卡在这个问题上： 我应该一开始就写 dataclass、class、Mapping、Callable 吗？ 还是先用普通 dict/list 把行为跑通？ 答案不是固定的，但有一个很实用的判断： 模型还不确定时，先用 dict/list 验证行为； 模型稳定后，再把稳定的东西整理成接口。 这不是偷懒，而是在降低错误抽象的成本。 比如你想验证一个 graph pipeline runner，最早可以只写： graph_plans = { "parallel": { "prepare": (), "enrich": ("prepare",), "extract": ("prepare",), } } 这个变量一开始不是“最终架构”，只是为了回答一个问题： dependency map 能不能表达 prepare 之后 enrich 和 extract 同时 ready？ 如果这个问题都没验证清楚，提前设计完整类层次反而会遮住真正的行为压力。 第一阶段：用普通变量验证模型 最早的脚本里，很多东西看起来像“常量”： ...

Python 类型标注里为什么用 Mapping、Sequence 这类抽象接口？ 副标题： 当一个函数只读取配置，不修改调用方传入的字典时，用 Mapping 比直接写 dict 更能表达接口意图；同理，Iterable、Sequence 和 Callable 也在表达“我需要什么能力”。 标签： Python / 类型标注 / Mapping / Sequence / Callable / 接口设计 适读人群： 正在写 Python 配置对象、runner、pipeline、SDK 参数的开发者 阅读时间： 7 min 背景：为什么不是所有字典参数都写成 dict？ 写 Python 配置型代码时，经常会有这种参数： graph_plans = { "parallel": { "prepare": (), "enrich": ("prepare",), "extract": ("prepare",), } } 如果一个 runner 只是读取它： runner = GraphPipelineRunner(graph_plans=graph_plans, handlers=handlers) 那么类型标注可以写成： from collections.abc import Mapping def __init__( self, *, graph_plans: Mapping[str, Mapping[str, tuple[str, ...]]], handlers: Mapping[str, WorkflowHandler], ) -> None: ... 刚开始看会觉得复杂：为什么不直接写 dict？ ...

函数命名不是取名字：它是在描述系统边界 副标题： 好的函数名不只是“好听”。它应该让读者看出：这个函数在做读取、 解析、校验、筛选、创建，还是修改状态。 适读人群： 正在写 Python service、runner、pipeline、SDK 或业务模块的开发者 阅读时间： 8 min 背景：为什么 get_dependencies 不是总比 resolve_graph 直观？ 看一个小型 graph pipeline runner： def trigger(self, mode, context): dependencies = self._resolve_graph(mode) self._validate_graph(dependencies) record = self._create_graph_record(mode, dependencies, context) while True: ready_nodes = self._ready_nodes(record) if not ready_nodes: break for node_name in ready_nodes: ... 刚开始可能会问： 为什么不叫 get_dependencies？ 为什么不叫 get_handler？ 为什么 ready_nodes 没有动词？ 为什么 skip_dependents 一看就像会改状态？ 这些不是风格洁癖。函数命名背后其实是在表达系统边界： mode 是外部语言，graph 是内部执行结构。 handler name 是配置语言，handler 是可调用对象。 ready node 是运行时状态，不是普通列表。 skip dependents 是副作用，不是查询。 这篇文章用一个 pipeline runner 作为例子，讲清楚函数命名应该如何暴露设计意图。 ...

标题 请求日志一定要带 RequestId 吗？Python 成熟实践与落地指南 副标题 / 摘要 几乎所有“请求相关”的日志都应该带 requestId，但要通过自动注入而不是手工拼接。 本文给出 Python 成熟做法、工程场景与与 tracing 的关系，帮你真正落地。 目标读者 初学者：第一次处理线上问题，不懂为什么日志要串 requestId。 中级开发者：需要一套可复制的 Python 日志注入方案。 团队负责人：想建立统一的日志与追踪规范。 背景 / 动机 当系统出现错误时，最常见的现场是： “某个时间点报错了，但不知道是哪次请求导致的。” 如果所有“请求相关日志”都有 requestId，你就能一条链串起来： 从入口 → DB → RPC → 异常，一次请求的关键路径一眼可见。 在微服务/多进程环境里，requestId 更是日志协作的最低门槛。 核心概念 requestId：一次请求的唯一编号，用于日志串联与快速定位。 trace_id / span_id：分布式追踪中的链路标识（trace）与步骤标识（span）。 上下文传播：跨线程 / 协程 / 服务传递 requestId 或 trace。 自动注入：通过 middleware + logging filter，在日志里自动带 requestId。 思维推导（从朴素到工程可用） 朴素做法：每条日志手动写 request_id，很快遗漏、重复、维护成本高。 痛点暴露：一次请求会跨多个函数/协程/库层，手写方式不可控。 关键观察：requestId 本质是“请求上下文”，应由框架统一注入。 方法选择：在入口生成 requestId → 传入上下文 → logging 自动注入。 正确性理由：上下文随请求自然传播，日志格式统一且不侵入业务代码。 A — Algorithm（题目与算法） 题目还原 “是不是每一条日志都应该带 requestId？” ...

任务编排为什么要放后端：让流程可控、可变、可回放 副标题 / 摘要 在多步骤、可中断、可回放的业务流程中，把“流程顺序与状态机”放在后端，是系统长期可演进的关键。本文从真实工程痛点出发，解释为什么前端不应硬编码流程顺序，并给出一套可落地的后端 Pipeline 编排思路与最小实现。 目标读者 正在设计多步骤流程 / 向导式产品的后端工程师 需要支撑 Web / App / Admin 多端一致流程的技术负责人 在 AI / LLM 产品中处理“模型自动 + 人工确认”混合流程的团队 背景 / 动机：问题通常是怎么爆出来的？ 很多系统一开始都很简单： 前端：第 1 步 → 第 2 步 → 第 3 步 后端：校验 + 存数据 但随着业务演进，以下需求几乎一定会出现： 步骤 变多：从 3 步变成 10+ 步 步骤 可选：根据条件跳过 / 插入新步骤 步骤 可中断：需要用户确认、补充信息、人工审核 步骤 可重试 / 可回放：失败后从中间继续，而不是全部重来 步骤 多端一致：Web / App / 内部工具共享同一流程 如果此时流程顺序仍然写在前端： 每次流程变更 = 多端发版 出问题时无法准确回答：现在卡在哪一步？ 想加监控、审计、回放，发现无从下手 根因只有一个： ...

副标题 / 摘要 ABC 用来“写抽象接口并阻止未实现的类被实例化”；ABCMeta 用来“在创建类时施加规则（自动注入、校验、注册）”。本文用最短可运行例子帮你在两者之间做选择。 目标读者 Python 初学者：了解抽象类怎么用、为啥会报 TypeError 中级开发者：在“接口约束”和“元类自动化”之间做取舍 需要做插件/框架能力的人：统一约束子类结构、自动补齐类级属性 背景 / 动机 你可能遇到过这些痛点： 想规定“子类必须实现某些方法”，但团队里总有人忘写 想让一批子类都有统一的类属性（比如 plugin_name），不想每个子类手写一遍 看到别人写 metaclass=ABCMeta，不确定是不是“更高级/更正确” 结论先说：大多数业务代码只需要 ABC；只有当你真的需要“类创建期的自动化规则”时，才考虑直接使用 ABCMeta（或在它上面做扩展）。 核心概念 1）抽象方法（@abstractmethod） 被标记为抽象的方法/属性，表示“必须由子类提供实现”。只要类里还有抽象成员未实现，它就不能被实例化。 2）抽象基类（ABC, Abstract Base Class） 用于定义一组接口约束：能继承、能被 isinstance/issubclass 判断，并能阻止不完整实现的类被实例化。 3）元类（metaclass） 普通类的“类”是 type；元类决定“类是怎么被创建出来的”。你可以在元类里： 在类创建时自动添加/修改类属性 校验子类是否符合规则（命名、属性、方法签名等） 统一注册子类到某个 registry ABCMeta 就是 abc 模块提供的元类：它把“抽象基类能力”实现为一套类创建/实例化规则。 实践指南 / 步骤 步骤 1：只需要“接口约束”——用 ABC 如果你只关心“子类必须实现哪些方法”，直接继承 ABC 是最简洁的写法。 步骤 2：需要“类创建期自动化规则”——用（或继承）ABCMeta 当你希望“子类不用手写，也能按规则自动拥有某些类属性/被校验/被注册”，再考虑元类。 可运行示例 示例 A：用 ABC 做接口约束（推荐默认选项） from abc import ABC, abstractmethod class Repo(ABC): @abstractmethod def save(self, obj) -> None: ... class MemoryRepo(Repo): def save(self, obj) -> None: print("saved:", obj) # Repo() # 取消注释会抛 TypeError：抽象类不能实例化 MemoryRepo().save({"id": 1}) 你得到的是：强约束（没实现抽象方法就不能实例化），且写法清晰。 ...

从参数直传到 Pipeline：一次可复现、可观测的数据处理管线改造实践 副标题： 为什么当处理链变得越来越长时，“配置驱动 + 上下文 + 外部存储”的 Pipeline 模式会比“参数直传”更适合工作？ 标签： Python / Pipeline / ETL / 数据工程 / 架构设计 适读人群： 后端开发、数据工程师、做文档处理/Embedding/索引构建的同学 阅读时间： 10–15 min 摘要： 本文记录我从“领域模型传来传去”的后端式写法，迁移到“配置驱动 Pipeline”模式的过程，总结落地要点、踩过的坑，以及为什么这种模式更适合复杂的数据加工链。 🧭 写这篇文章的动机 做后端时，我长期习惯一种简单粗暴的风格： 需要什么参数就一直往下传，函数链一路 call 下去。 很多业务都是这么写的： 输入是个模型/DTO 处理完传下一个函数 大对象在链路里飘来飘去 但当我开始做 文档处理、Embedding、实体提取、索引构建 这种“多步骤、可重跑、需观测”的任务时，这种写法很快失效： 需要重跑某一步时必须重建整条调用链 中间产物无法落地检查 改一个策略需要改一堆函数签名 并发 / 异常恢复都难处理 后来接触到构建数据处理 Workflow / ETL Pipeline 的方式，发现它的核心思路完全不一样： 配置驱动策略 → 上下文承载运行态 → 外部存储承载数据流。 这套体系让多步处理链突然变得可插拔、能恢复、能观测、能重放。 于是就有了这篇文章，把我的心智迁移过程与实践要点记录下来。 ⚡ TL;DR（你只看这一屏也能理解本文核心） 配置驱动： 路径、模型、超参都写 config，而不是塞进函数参数里。 上下文 context： 统一管理 I/O 句柄、缓存、回调、统计、运行时标志。 外部存储： 步骤间不传大对象，读写约定表名：documents → text_units → entities → index。 可插拔 Pipeline： “步骤名 → 函数指针”的顺序列表，可一键切换 Standard/Fast 等方案。 幂等与恢复： 中间表持久化，可覆盖或版本化，崩溃后能断点续跑。 观测与回调： start/end/进度统一上报，产出 stats.json，定位问题更快。 异步友好： 步骤 await 执行，内部可分片并发或调用 LLM。 取舍： 成本是心智负担增加；收益是可观察、可重跑、可替换、低耦合。 👥 目标读者 适合以下同学阅读： ...

标题 Pydantic vs dataclass vs TypedDict：谁负责什么，怎么组合？ 副标题 / 摘要 承接《别让 Pydantic 占领你的整个项目》，这一篇用对比视角把 Pydantic、dataclass、TypedDict 的定位、取舍和组合方式讲清楚：谁用于 API 校验、谁承载业务状态、谁只做类型提示。 目标读者 FastAPI / Pydantic 用户，想搞清楚“数据类”该放在哪一层 有 0–5 年经验、在做服务端建模的 Python 工程师 已读过前一篇分层文章，想进一步对比具体工具 背景 / 动机：为什么要区分三者？ 在上一篇里，我们强调“Pydantic 应该停留在 API/外围”。很多同学随后会问： “那 Python 原生 dataclass 呢？和 Pydantic 有什么差？” “TypedDict 是不是又一个‘数据类’，要不要取代 dataclass？” “什么时候该用 Pydantic dataclasses，什么时候用标准库？” 不区分清楚，常见后果有： 用 TypedDict 写业务逻辑，测试时才发现它根本不做运行时校验； 用 Pydantic BaseModel 传来传去，导致 Domain 强绑定外部依赖； dataclass 和 Pydantic 混用，序列化和校验边界越来越模糊。 核心概念：一句话定位 Pydantic BaseModel：运行时校验 + 类型转换 + JSON 友好；属于“对外/边界”。 dataclass（标准库）：轻量数据载体，可承载业务方法；不做自动校验，属于“领域/内部”。 TypedDict：仅提供静态类型提示，运行时就是普通 dict；属于“静态约束/外部协议”。 主要差异： ...

这篇文章从一个简单的工单系统出发，展示如何在 Python 项目中以业务对象为中心设计接口、仓储与服务，而不是让 ORM、框架和表结构牵着鼻子走。