PAYATHON 2026

如何静态验证 Python、YAML 与 JSON Schema 流水线的阶段类型兼容性？

数据阿明

可以统一阶段契约，将 YAML 中的连接编译成 Python 调用，再进行静态类型检查，在执行流水线前发现阶段连接错误。mypy、Pydantic 和 JSON Schema 校验器可以分别承担其中的工作，但它们各自检查通过，不代表整个流水线的类型兼容性已得到证明。

为什么单独使用类型检查器或 JSON Schema 不够？

三种检查分别针对不同对象：

检查对象需要回答的问题
Python 阶段实现函数参数、返回值及内部操作是否符合类型注解？
YAML 配置阶段名称、参数、依赖关系是否有效？
阶段连接上游可能产生的输出，是否都能被下游接受？

mypy 检查 Python 代码。对于只存在于 YAML 中的连接关系，需要先将其转换成类型检查器能分析的代码或其他明确的契约表示。复杂的动态反射和元编程也会限制静态分析能力。mypy FAQ

jsonschema.validate(instance, schema) 检查某个实例是否符合一份 Schema。用它验证 YAML 解析后的配置，只能说明配置符合规范，不能证明两个阶段的数据契约兼容。jsonschema 校验文档

可实施的模式：把 YAML 连接编译成 Python 调用

先要求每个阶段提供明确的输入、输出类型。下面用简单类型演示：

# stages.py

def load_numbers() -> list[int]:
    return [1, 2, 3]


def sum_numbers(values: list[int]) -> int:
    return sum(values)


def join_words(values: list[str]) -> str:
    return ",".join(values)

假设 YAML 配置了以下连接，配置格式仅用于示例：

stages:
  - id: source
    uses: load_numbers
  - id: sink
    uses: join_words
    input: source

构建阶段读取配置，解析阶段注册表，生成一个供静态检查使用的模块：

# generated_plan.py
from stages import join_words, load_numbers


def check_plan() -> None:
    source_output = load_numbers()
    join_words(source_output)

然后在 CI 中运行：

mypy --strict stages.py generated_plan.py

这段调用存在类型冲突：load_numbers() 返回 list[int]，而 join_words() 接收 list[str]。类型检查无需执行 check_plan()，也无需运行阶段函数。

如果 YAML 的下游改为 sum_numbers，生成器就应生成对应的 sum_numbers(source_output) 调用，使检查结果与实际配置一致。

YAML 到调用代码的转换层需要自行实现，mypy 不会自动完成。生成器与执行器应共用阶段解析和连接规则，避免检查一种连接、实际执行另一种连接。

如何让 JSON Schema 与 Python 类型保持一致？

优先选择一处作为契约的定义来源，减少手工维护两套定义造成的偏差。

如果以 Python 类型为准，Pydantic v2 提供 BaseModel.model_json_schema() 和 TypeAdapter.json_schema() 来生成 JSON Schema。例如：

from pydantic import TypeAdapter

numbers_schema = TypeAdapter(list[int]).json_schema()
words_schema = TypeAdapter(list[str]).json_schema()

这些 API 负责生成 Schema，本身不负责检查 YAML 中的阶段连接。Pydantic JSON Schema 文档

如果阶段边界传递的是序列化后的 JSON，需要区分输出的 serialization Schema 与输入的 validation Schema。Python 对象的类型兼容，不代表经过序列化、反序列化或类型转换后仍然兼容。

如果现有 JSON Schema 必须独立维护，就需要增加 Schema 兼容性检查，不能只比较 Schema 文件是否相同。

Schema 兼容性应该检查什么？

对于直接传递 JSON 数据、没有中间转换的连接，设 L(S) 为满足 Schema S 的全部 JSON 实例，契约兼容要求：

L(upstream_output_schema) ⊆ L(downstream_input_schema)

也就是说，上游契约允许的每一种输出，都必须满足下游的输入要求。这个条件保证的是契约层面的兼容，前提是阶段实现确实遵守声明的输出契约。

例如，上游允许任意 integer，下游只接受满足 minimum: 0 的整数，就不能判为兼容。即使某次样本恰好为正数，也不能证明所有输出都满足要求。

对象还需要处理必填字段、额外字段以及嵌套约束。在 JSON Schema 中，仅在 properties 中声明字段不会让它自动成为必填字段；额外字段的接收规则也需要明确设置。JSON Schema 对象规则

如果自行实现兼容性检查器，应先限定支持的 Schema 方言与关键字。对于尚未支持的组合、引用或条件约束，返回“无法判定”或拒绝构建，避免静默判为兼容。

建议怎样接入构建流程？

  1. 校验 YAML 解析后的配置，检查阶段是否存在、引用是否有效、参数是否完整。
  2. 根据实际连接生成带具体类型的 Python 调用，交给 mypy 检查。
  3. 对跨 JSON 边界的连接，额外验证输出 Schema 到输入 Schema 的兼容性。
  4. 将错误映射回 YAML 的阶段 ID 和对应端口，方便定位。

配置校验和代码生成属于流水线运行前的预检，会执行检查程序，但不需要执行业务阶段。

阶段接口应避免退化为 Any 或 Callable[..., Any]，否则不兼容的连接可能被放过；Any 会绕过许多静态类型约束。mypy 类型说明

对于外部输入、动态加载的实现，以及普通类型注解无法表达的数值范围或业务约束，仍需保留运行时校验。静态检查能提前发现声明层面的错误，但不能保证任意 Python 实现和实际数据都符合声明。

备注：内容仅供参考。