Python

一个修改商品价格的功能，接口检查价格是不是整数，应用服务检查一次，领域对象检查一次，保存时又检查一次。代码分成了四层，同一个问题也被问了四遍。 这些检查看起来都在提高可靠性，但如果删掉其中一条，我们能说清楚究竟失去了什么保护吗？ 之前的《让领域模型和仓储模型各自负责》讨论了模型与持久化之间的边界。这篇继续讨论一个更具体的问题：边界已经划分之后，哪些检查应该保留，哪些可以交给前面的代码保证？ 如果你已经接触过 Python 的 dataclass 和简单分层，但觉得业务代码越来越像校验框架，可以从下面这个改价功能开始。 一个价格，为什么检查四遍 假设商品后台收到这样的 JSON： { "sku": "P1001", "price_cents": 9900 } 价格使用整数分。对于这个例子，业务还约定：价格不能为负数，停售商品不能改价。 容易写出的代码是，在入口、服务、模型和 Repository 中都放上类似判断： if type(price_cents) is not int: raise ValueError("价格必须是整数分") 问题在于，这四次检查通常处理的是同一个已经解析完成的值。只要内部代码没有重新引入不可信数据，重复检查提供的保护就很有限。 真正需要区分的是两个问题： 问题 例子 负责的位置 输入是否符合接口约定？ 缺少价格、价格是字符串 外部输入入口 操作是否符合业务规则？ 负价格、对停售商品改价 领域模型 应用服务负责组织操作：加载商品、调用改价方法、保存商品。Repository 负责读写和映射。它们不需要把前两类判断各自重写一份。 先建立一个可以信任的入口 Python 的类型注解不会自动验证运行时数据。写下 price_cents: int，并不能阻止调用方传入字符串。 因此，“内部可以信任类型”需要前提：外部数据进入内部流程之前，确实经过了解析和检查。 下面是这个入口。示例采用严格协议，只接受两个字段；不自动把字符串价格转换成整数。 from dataclasses import dataclass, replace @dataclass(frozen=True) class ChangePriceRequest: sku: str price_cents: int def decode_change_price(data: object) -> ChangePriceRequest: if not isinstance(data, dict): raise ValueError("请求必须是 JSON 对象") if set(data) != {"sku", "price_cents"}: raise ValueError("请求必须且只能包含 sku 和 price_cents") sku = data["sku"] if not isinstance(sku, str) or not sku.strip(): raise ValueError("sku 必须是非空字符串") if type(data["price_cents"]) is not int: raise ValueError("价格必须是整数分") return ChangePriceRequest(sku.strip(), data["price_cents"]) 这里使用 type(value) is int，是因为 Python 的 bool 是 int 的子类。这个接口不接受 true 作为价格。 ...

一个策略选择工具，本来只需要让 Agent 查看模板、选择版本、取得写作规则。写着写着，代码却变成了另一副样子：遍历字段名、判断字典形状、检查数组元素、猜测字段类型，再把已经检查过的数据交给下一层继续检查。 每一条判断单独看都有解释，放在一起却让人看不清工具到底做什么。 这发生在一次写作策略工具教程的审查中。删除一个通用转换助手后，代码甚至更长了：检查没有减少，只是搬进了另一个函数。问题并不是少用了一个框架，也不是变量命名不够规范，而是还没决定哪些错误需要处理，就先把能想到的检查写了出来。 本文适合已经了解 Python dataclass、正在让 Agent 编写小型工具或分层应用的读者。以下例子从那次审查中缩小而来，用来说明设计取舍，不是完整策略工具的实现。 前两篇分别讨论模型与仓储的职责和入口检查与业务约束的分工。这篇往前追问一步：在决定校验放在哪里之前，我们为什么需要这条校验？ 一个数组转换，怎么长成了通用校验器 假设模板包含 structure 字段。JSON 中它是数组，领域对象中希望用 tuple 表示： structure = tuple(data["structure"]) 这行代码完成转换，但不保证输入正确： assert tuple("post") == ("p", "o", "s", "t") 如果 Agent 把数组写成字符串，转换不会报错，错误会以另一种形式进入模型。因此，在接收 Agent 请求的地方，检查它是不是字符串数组是有具体理由的。 但这不意味着每个处理 structure 的方法，都应该重新执行下面三项检查： 它是不是数组？ 数组里的元素是不是字符串？ 数组是不是非空？ 前两项属于输入协议；第三项取决于业务：这个模板是否允许没有正文结构？如果模型已经保证结构不能为空，入口无需重复保证一次。 同样的字段，也可能来自程序自己写出的文件。是否检查，要先确定文件的使用方式。 数据来源 能依赖的前提 处理方式 Agent 提交的编辑 JSON 尚未确认类型和字段 入口解析，拒绝不符合约定的结构 已解析的内部请求对象 调用者遵守既定契约 直接使用，不重复判断类型 同一程序写出、没有独立编辑入口的文件 写入与读取遵循同一格式 可以直接映射，加载失败时暴露错误 允许人工编辑或导入的文件 内容可能不符合格式 把文件读取当作输入边界，定义需要的检查 “放在仓库里”不等于“可信”。如果维护者经常手改 JSON，它就是一个真实的编辑入口。反过来，受控文件也不是永远不会损坏；选择直接映射，意味着不承诺为每一种损坏提供定制诊断，更不保证发现所有能被转换的错误类型。 一个能直接运行的例子 缩小需求：模板名称不能为空，结构至少有一项，每一项不能是空白文字。外部请求只接受 name 和 structure，不自动转换数字或字符串数组。 保存下面代码为 template_boundary.py，使用 Python 3.10 或更新版本运行： from dataclasses import dataclass @dataclass(frozen=True) class Template: name: str structure: tuple[str, ...] def __post_init__(self): if not self.name.strip(): raise ValueError("模板名称不能为空") if not self.structure or any(not step.strip() for step in self.structure): raise ValueError("模板必须包含非空的写作步骤") def decode_template(data: object) -> Template: if not isinstance(data, dict) or set(data) != {"name", "structure"}: raise ValueError("请求必须包含且只能包含 name 和 structure") if not isinstance(data["name"], str): raise ValueError("name 必须是字符串") steps = data["structure"] if not isinstance(steps, list) or not all(isinstance(step, str) for step in steps): raise ValueError("structure 必须是字符串数组") return Template(name=data["name"], structure=tuple(steps)) def writing_outline(template: Template) -> str: return "\n".join(f"{i}. {step}" for i, step in enumerate(template.structure, 1)) if __name__ == "__main__": template = decode_template({ "name": "问题与解释", "structure": ["提出具体问题", "解释原因与适用条件"], }) print(writing_outline(template)) cases = [ ({"name": "问题与解释", "structure": "post"}, "structure 必须是字符串数组"), ({"name": "问题与解释", "structure": [42]}, "structure 必须是字符串数组"), ({"name": "问题与解释", "structure": []}, "模板必须包含非空的写作步骤"), ({"name": "问题与解释", "structure": [" "]}, "模板必须包含非空的写作步骤"), ({"name": " ", "structure": ["解释"]}, "模板名称不能为空"), ] for payload, expected in cases: try: decode_template(payload) except ValueError as error: assert str(error) == expected else: raise AssertionError(f"错误输入被接受：{payload!r}") print("正常输入和 5 个失败场景均符合约定") 运行 python3 template_boundary.py，输出： ...

在请求模型里看到 @field_validator("score")，很容易以为它只是给方法加了一个标记：这个方法由谁调用？返回的值又去了哪里？ 它的作用是告诉 Pydantic：校验 score 字段时，自动执行下面的方法，并把返回值作为这个字段的最终值。 本文适合已经会写 BaseModel、但还不熟悉自定义校验的读者，示例使用 Pydantic v2。 类型正确，还不够 假设接口接收一次考试成绩： { "score": 120 } int 可以说明成绩应当是整数，但 120 也是整数。对于满分 100 分的考试，还需要检查成绩是否在 0～100 之间。 这些规则可以放进字段校验器，在请求进入业务处理前统一完成。 一个可以直接运行的例子 把下面代码保存为 example.py，通过 uv run --with 'pydantic>=2,<3' python example.py 执行： from pydantic import BaseModel, ValidationError, field_validator class ExamResult(BaseModel): score: int @field_validator("score") @classmethod def validate_score(cls, value: int) -> int: if not 0 <= value <= 100: raise ValueError("成绩必须在 0～100 之间") return value result = ExamResult(score=85) assert result.score == 85 print(result.score) try: ExamResult(score=120) except ValidationError as exc: error = exc.errors()[0] assert error["loc"] == ("score",) print(error["msg"]) else: raise AssertionError("应该拒绝超过满分的成绩") 输出： ...

这篇文章只记录 Agent 工具调用的最小闭环：声明工具 schema、接收 function_call、执行本地函数，并通过 function_call_output 继续下一次模型决策。

下面这段代码不会打印任何内容，也不会报错： for x in range(10, 7): print(x) 原因很简单：range(10, 7) 使用默认的正步长，起点已经超过终点，因此范围为空，循环体一次也不执行。 本文适合刚开始学习 Python 循环的读者，阅读约需 3 分钟。 先看懂 range() 的三个参数 range() 常见的写法有三种： list(range(4)) # [0, 1, 2, 3] list(range(2, 7)) # [2, 3, 4, 5, 6] list(range(2, 7, 2)) # [2, 4, 6] 完整形式是 range(start, stop, step)： start：起点，省略时默认为 0。 stop：终点，不包含在范围内。 step：步长，省略时默认为 1。 range() 返回的是一个范围对象，并不是列表。这里使用 list()，只是为了直接观察其中的值。 对于正步长，每个值都必须满足 x < stop。因此，range(2, 7) 从 2 开始，每次增加 1，到 6 为止，不包含 7。 起点碰到或超过终点，会发生什么？ 默认步长是 1，所以如果一开始就有 start >= stop，连第一个值都不符合条件： list(range(7, 7)) # [] list(range(10, 7)) # [] 空范围是正常结果，不是异常。 for 循环没有值可以遍历，就会直接跳过循环体，继续执行后面的代码： ...

这是 Redis 任务队列系列的第一篇：从一个具体的计算任务开始，用 Python 实现生产者和消费者，先掌握 Redis List 的入队与阻塞出队，再看为什么它还不能直接作为可靠的生产任务队列。

接口返回后继续计算，是否就算任务队列？要看任务如何等待、被领取以及在进程中断后恢复。本文从一个五任务场景推导生产者与消费者模型，再讨论并发、积压、数据库租约，以及 Celery 和 Redis 分别负责什么。

在阅读 Repository 或 Service 代码时，经常会同时看到 @staticmethod 和 @abstractmethod。它们都写在方法上方，很容易被误认为是同一类装饰器。 实际上，两者解决的是不同问题： @staticmethod 解决方法如何被调用 @abstractmethod 约束子类必须提供什么行为 理解这个区别后，方法应该放在类中、模块中，还是抽象基类中，会清楚很多。 @staticmethod：不依赖实例状态的方法 普通实例方法默认接收 self： class KeywordMatcher: def contains(self, text: str, keyword: str) -> bool: return keyword.casefold() in text.casefold() 如果方法体不读取或修改 self，可以写成静态方法： class KeywordMatcher: @staticmethod def contains(text: str, keyword: str) -> bool: return keyword.casefold() in text.casefold() 调用时不需要创建实例： KeywordMatcher.contains("变压器采购项目", "变压器") 静态方法没有自动传入的 self，因此不能访问实例属性： class KeywordMatcher: def __init__(self, case_sensitive: bool): self.case_sensitive = case_sensitive @staticmethod def contains(text: str, keyword: str) -> bool: # 这里不能访问 self.case_sensitive return keyword in text 如果函数只是通用计算，并不属于某个类的概念，直接定义为模块级函数通常更自然： def contains(text: str, keyword: str) -> bool: return keyword.casefold() in text.casefold() 所以“不需要 self”是使用 @staticmethod 的重要条件，但还要问一句：这个行为是否真的属于该类？如果答案是否定的，就不必为了使用装饰器而把函数塞进类里。 ...

JSON 不只有键值对，还可以表示数组、字符串、数字和布尔值。本文用关键词订阅示例解释 Python tuple、list、dict 与 SQLAlchemy JSON 列之间的映射关系。

Python 类型标注中的 tuple[T, ...]：任意长度和固定长度 副标题： tuple[DocumentBlock, ...] 中的 ... 不是一个要放进元组的值，而是 类型标注中的“任意数量”记号。理解它，才能看懂解析结果为什么使用元组，以及它和列表的边界。 适读人群： 正在使用 Python 类型标注、dataclass 或文档解析模型的开发者 阅读时间： 6 分钟 从 ParsedDocument 的 blocks 字段说起 在文档解析服务中，可以把解析结果表示成一个不可变的数据对象： from dataclasses import dataclass @dataclass(frozen=True) class DocumentBlock: block_id: str raw_text: str @dataclass(frozen=True) class ParsedDocument: blocks: tuple[DocumentBlock, ...] 这里最容易让人疑惑的是： blocks: tuple[DocumentBlock, ...] 这个标注到底表示什么？为什么不是： blocks: tuple[DocumentBlock] 答案是：前者表示“任意长度、每一项都是 DocumentBlock”，后者表示“固定只有一项， 而且这一项是 DocumentBlock”。 tuple[T, ...] 表示任意长度 把它拆开看： tuple[T, ...] ^ ^ | └── 元素数量不限 └────── 每一项的类型是 T 因此： ...