Python类型驱动开发实战：用Pydantic v2构建零容错数据处理管线

一、先看清楚问题长什么样

这个服务做的事情不复杂：从五六个第三方API拉数据，清洗合并之后写进数据仓库。听起来是个标准的ETL流程，但坑就坑在”第三方API”这四个字上。不同供应商的接口规范各搞一套，有的用驼峰命名，有的用下划线；同一个字段，A家返回字符串，B家返回整数，C家干脆有时候给有时候不给。原来的代码为了兼容这些情况，写了一层又一层的if-else和try-except，到最后连原作者自己都看不懂了。

举个真实例子，拿”用户订单”这个数据结构来说，三个上游返回的JSON分别是这样的：

# 供应商A
{
    "orderId": "A-20240115-8892",
    "userId": 10423,
    "amount": "199.00",
    "status": 1,
    "created_at": "2024-01-15T08:30:00Z"
}

# 供应商B
{
    "order_id": 92718,
    "user_id": "usr_10423",
    "amount": 199.0,
    "status": "paid",
    "create_time": "2024-01-15 16:30:00"
}

# 供应商C —— 注意这个没写amount字段
{
    "orderId": "C8812",
    "customerRef": 10423,
    "state": "completed",
    "timestamp": 1705307400
}

字段名不一样、类型五花八门、有些字段说没就没。老代码的处理方式是在每个供应商的解析函数里手动做转换和默认值填充，然后拼成一个统一的字典往下传。问题是字典这东西，在Python里就是个黑盒——你没法一眼看出它里面到底有什么键，每个键对应什么类型。等到下游代码访问某个键的时候才发现不对劲，异常已经抛出来了。

二、换个思路：让类型帮你干活

Python的类型提示这几年发展很快，但光靠原生的typing模块做运行时校验还是差点意思。isinstance写多了代码会变得很啰嗦，而且处理不了”这个字段可能是字符串也可能是整数”这种联合类型的情况，更不用说自动转换和嵌套模型的校验了。

这里就要请出Pydantic v2。v2版本用Rust重写了核心校验引擎，性能比v1快了一个数量级，而且对联合类型、泛型、自定义校验器的支持都更完善。它的核心思路很简单：你定义好数据的形状（shape），数据进来的时候自动校验、自动转换、自动补齐默认值。不合规的数据直接拦截，根本不会流到下游。

先来定义一个统一的订单模型：

from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import Optional

from pydantic import BaseModel, Field, field_validator


class OrderStatus(str, Enum):
    PENDING = "pending"
    PAID = "paid"
    SHIPPED = "shipped"
    COMPLETED = "completed"
    CANCELLED = "cancelled"


class UnifiedOrder(BaseModel):
    order_id: str = Field(..., min_length=1, description="订单唯一标识")
    user_id: int = Field(..., gt=0, description="用户ID")
    amount: Decimal = Field(..., gt=0, description="订单金额")
    status: OrderStatus = Field(default=OrderStatus.PENDING, description="订单状态")
    created_at: datetime = Field(..., description="创建时间")

    @field_validator('order_id')
    @classmethod
    def order_id_must_be_reasonable(cls, v: str) -> str:
        # 去掉首尾空格，统一转大写前缀
        v = v.strip()
        if len(v) > 64:
            raise ValueError(f'订单ID过长: {len(v)}个字符')
        return v

    @field_validator('amount', mode='before')
    @classmethod
    def coerce_amount(cls, v):
        # 上游可能传字符串、整数、浮点数，统一转成Decimal
        if isinstance(v, str):
            v = v.replace(',', '').strip()
        return Decimal(str(v))

这个模型定义了几个关键点：amount字段用了mode='before'的校验器，意味着在Pydantic做类型转换之前就会执行，这样无论是"199.00"还是199.0还是199，都能先被转成统一的字符串再交给Decimal处理。status字段用了枚举，限定了只能是那五种状态，传进来的数字或者别的字符串都会被拦截。

定义好模型之后，下一步是为每个供应商写一个适配器，把它们的原始数据映射到UnifiedOrder上：

from typing import Any


class SupplierAdapter:
    """供应商适配器基类"""

    @staticmethod
    def parse(raw: dict[str, Any]) -> UnifiedOrder:
        raise NotImplementedError


class SupplierAAdapter(SupplierAdapter):
    STATUS_MAP = {
        1: OrderStatus.PENDING,
        2: OrderStatus.PAID,
        3: OrderStatus.SHIPPED,
        4: OrderStatus.COMPLETED,
        5: OrderStatus.CANCELLED,
    }

    @staticmethod
    def parse(raw: dict[str, Any]) -> UnifiedOrder:
        return UnifiedOrder(
            order_id=raw.get("orderId", ""),
            user_id=raw.get("userId", 0),
            amount=raw.get("amount", "0"),
            status=SupplierAAdapter.STATUS_MAP.get(raw.get("status"), OrderStatus.PENDING),
            created_at=raw.get("created_at", "1970-01-01T00:00:00Z"),
        )


class SupplierBAdapter(SupplierAdapter):
    STATUS_MAP = {
        "pending": OrderStatus.PENDING,
        "paid": OrderStatus.PAID,
        "shipped": OrderStatus.SHIPPED,
        "completed": OrderStatus.COMPLETED,
        "cancelled": OrderStatus.CANCELLED,
    }

    @staticmethod
    def parse(raw: dict[str, Any]) -> UnifiedOrder:
        raw_user_id = raw.get("user_id", "usr_0")
        if isinstance(raw_user_id, str) and raw_user_id.startswith("usr_"):
            raw_user_id = raw_user_id[4:]
        return UnifiedOrder(
            order_id=str(raw.get("order_id", "")),
            user_id=int(raw_user_id),
            amount=raw.get("amount", 0),
            status=SupplierBAdapter.STATUS_MAP.get(raw.get("status", ""), OrderStatus.PENDING),
            created_at=raw.get("create_time", "1970-01-01T00:00:00"),
        )


class SupplierCAdapter(SupplierAdapter):
    STATE_MAP = {
        "pending": OrderStatus.PENDING,
        "completed": OrderStatus.COMPLETED,
        "cancelled": OrderStatus.CANCELLED,
    }

    @staticmethod
    def parse(raw: dict[str, Any]) -> UnifiedOrder:
        timestamp = raw.get("timestamp", 0)
        if isinstance(timestamp, (int, float)) and timestamp > 0:
            created = datetime.fromtimestamp(timestamp)
        else:
            created = datetime(1970, 1, 1)
        return UnifiedOrder(
            order_id=raw.get("orderId", ""),
            user_id=raw.get("customerRef", 0),
            amount=raw.get("amount", "0.00"),
            status=SupplierCAdapter.STATE_MAP.get(raw.get("state", ""), OrderStatus.PENDING),
            created_at=created,
        )

注意适配器里的parse方法只负责把原始字段映射过来，不做严格校验。真正把关的是UnifiedOrder的构造过程——如果某个必填字段缺失，Pydantic会直接抛出ValidationError，精确到哪个字段、什么原因。

三、加上异步和批量处理

实际场景里，上游接口的响应时间从200毫秒到8秒不等，串行拉取显然不现实。这里用asyncio配合httpx做并发请求，同时把验证逻辑也异步化——虽然Pydantic的校验本身是同步的，但我们可以用asyncio.to_thread把CPU密集的校验丢到线程池里，避免阻塞事件循环。

import asyncio
from dataclasses import dataclass
from typing import Callable

import httpx
from pydantic import ValidationError


@dataclass
class FetchResult:
    supplier_name: str
    orders: list[UnifiedOrder]
    errors: list[dict]

    @property
    def success_count(self) -> int:
        return len(self.orders)

    @property
    def error_count(self) -> int:
        return len(self.errors)


async def fetch_from_supplier(
    client: httpx.AsyncClient,
    name: str,
    url: str,
    adapter: Callable[[dict], UnifiedOrder],
    semaphore: asyncio.Semaphore,
) -> FetchResult:
    orders: list[UnifiedOrder] = []
    errors: list[dict] = []

    async with semaphore:
        try:
            response = await client.get(url, timeout=15.0)
            response.raise_for_status()
            raw_data = response.json()
        except httpx.HTTPError as exc:
            return FetchResult(
                supplier_name=name,
                orders=[],
                errors=[{"supplier": name, "error": f"HTTP请求失败: {exc}"}],
            )

    # 如果返回的是列表，逐条处理
    items = raw_data if isinstance(raw_data, list) else [raw_data]

    for idx, item in enumerate(items):
        if not isinstance(item, dict):
            errors.append({"supplier": name, "index": idx, "error": f"非字典类型: {type(item)}"})
            continue
        try:
            # 把适配和校验一起丢到线程池，避免阻塞事件循环
            order = await asyncio.to_thread(adapter, item)
            orders.append(order)
        except ValidationError as ve:
            errors.append({
                "supplier": name,
                "index": idx,
                "error": ve.errors(),
                "raw": item,
            })
        except Exception as exc:
            errors.append({
                "supplier": name,
                "index": idx,
                "error": str(exc),
                "raw": item,
            })

    return FetchResult(supplier_name=name, orders=orders, errors=errors)

Semaphore用来控制并发数，避免把上游打崩。每个供应商的请求失败或者数据校验失败都不会影响其他供应商的收集，错误信息被妥善记录而不是直接抛出。asyncio.to_thread在这里的另一个好处是：当数据量大的时候，Pydantic的校验开销可以分散到多个线程上，不会让主事件循环卡住。

主调度函数把几个供应商的协程一起跑：

async def collect_all_orders() -> dict[str, FetchResult]:
    suppliers = [
        ("SupplierA", "https://api.a.example.com/orders", SupplierAAdapter.parse),
        ("SupplierB", "https://api.b.example.com/orders", SupplierBAdapter.parse),
        ("SupplierC", "https://api.c.example.com/orders", SupplierCAdapter.parse),
    ]

    semaphore = asyncio.Semaphore(3)  # 最多同时3个请求

    async with httpx.AsyncClient() as client:
        tasks = [
            fetch_from_supplier(client, name, url, adapter, semaphore)
            for name, url, adapter in suppliers
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    output: dict[str, FetchResult] = {}
    for (name, _, _), result in zip(suppliers, results):
        if isinstance(result, Exception):
            output[name] = FetchResult(
                supplier_name=name,
                orders=[],
                errors=[{"supplier": name, "error": f"协程异常: {result}"}],
            )
        else:
            output[name] = result
    return output

跑完之后，output里每个供应商的结果一目了然：成功解析了多少条，哪些数据校验失败了、失败原因是什么。下游只需要消费result.orders这个列表，里面的每一条数据都是经过完整校验的UnifiedOrder实例——类型明确、字段齐全、格式统一。

四、错误处理不是打日志就完了

校验失败的数据不应该直接丢弃。很多团队的做法是打个ERROR日志然后当没看见，等到数据对不上了再回头翻日志，效率很低。更好的做法是把校验失败的数据单独存起来，能做自动修复的就修复，修不了的通知到人。

这里我设计了一个轻量的错误收集与修复机制：

import json
from pathlib import Path
from typing import Any


class ValidationErrorRecorder:
    """把校验失败的数据写入本地JSONL文件，方便后续排查和修复"""

    def __init__(self, base_dir: str = "./error_records"):
        self.base_dir = Path(base_dir)
        self.base_dir.mkdir(parents=True, exist_ok=True)

    def record(self, supplier: str, errors: list[dict]) -> None:
        if not errors:
            return
        filepath = self.base_dir / f"{supplier}_errors.jsonl"
        with open(filepath, "a", encoding="utf-8") as f:
            for error_entry in errors:
                f.write(json.dumps(error_entry, ensure_ascii=False, default=str) + "n")

    def load_unresolved(self, supplier: str) -> list[dict]:
        filepath = self.base_dir / f"{supplier}_errors.jsonl"
        if not filepath.exists():
            return []
        records = []
        with open(filepath, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line:
                    records.append(json.loads(line))
        return records

    def clear_after_fix(self, supplier: str) -> None:
        filepath = self.base_dir / f"{supplier}_errors.jsonl"
        if filepath.exists():
            filepath.unlink()


class AutoFixer:
    """对一些已知的、可修复的问题做自动修正"""

    @staticmethod
    def try_fix(raw: dict[str, Any]) -> dict[str, Any] | None:
        fixed = dict(raw)
        made_change = False

        # 修复1：amount字段可能是带千位分隔符的字符串，如 "1,234.56"
        if "amount" in fixed and isinstance(fixed["amount"], str):
            if "," in fixed["amount"]:
                fixed["amount"] = fixed["amount"].replace(",", "")
                made_change = True

        # 修复2：时间戳可能是毫秒级的（13位），截断到秒级
        if "timestamp" in fixed and isinstance(fixed["timestamp"], (int, float)):
            if fixed["timestamp"] > 1e12:
                fixed["timestamp"] = fixed["timestamp"] // 1000
                made_change = True

        # 修复3：userId可能是浮点数（比如从Excel读出来的）
        if "userId" in fixed and isinstance(fixed["userId"], float):
            if fixed["userId"] == int(fixed["userId"]):
                fixed["userId"] = int(fixed["userId"])
                made_change = True

        return fixed if made_change else None

在fetch_from_supplier里，对于校验失败的数据可以先尝试AutoFixer.try_fix，修复成功就重新走一遍校验，修复失败再记录到ValidationErrorRecorder里。这样一来，真正需要人工介入的数据量会大幅减少。

这个错误记录器的设计还有一个好处：JSONL格式一行一条记录，可以直接用grep或者jq做快速分析，不需要额外部署什么工具。运维同事拿到文件就能定位问题。

五、把校验前置到开发阶段

Pydantic模型定义好之后，它不只是运行时工具。你可以直接在单元测试里用同一个模型来构造测试数据，保证测试用例的数据结构和生产环境完全一致。更重要的是，Pydantic v2支持从模型生成JSON Schema，这个Schema可以直接喂给API文档工具或者前端团队的Mock服务：

# 生成JSON Schema
schema = UnifiedOrder.model_json_schema()
print(json.dumps(schema, indent=2, ensure_ascii=False))

输出的Schema描述了每个字段的类型、是否必填、约束条件，甚至包括field_validator里定义的校验逻辑的说明（如果你在Field的description参数里写了的话）。前后端对接的时候，这份Schema就是一份活文档，比手写的接口文档靠谱得多。

另外，如果你用VS Code或者PyCharm，Pydantic模型的类型推断体验也很舒服。定义一个UnifiedOrder变量之后，编辑器会自动补全.order_id、.amount等属性，不需要翻回去看模型定义。类型即文档，说的就是这个感觉。

六、性能这一点，别听别人说，自己测

关于Pydantic的性能，v2相比v1的提升官方给出的数据是5到50倍，具体取决于数据结构和校验复杂度。我在这个项目里也粗略测了一下：单条数据校验大约12微秒，10万条数据批量校验在1.2秒左右（M1 Pro芯片）。对比原来手写的校验逻辑，性能基本持平，但代码量减少了将近70%，而且不会漏掉边界情况。

如果你的场景对性能极其敏感（比如每秒要校验百万级的数据），可以考虑用Pydantic的model_validate方法配合预编译的校验器，或者直接用msgspec这种更底层的库。不过对于绝大多数业务场景，Pydantic v2的性能已经完全够用了。真正拖慢速度的通常是网络IO和数据库写入，而不是数据校验本身。

七、回过头来看，到底解决了什么

这个数据聚合服务重构之后跑了两个月，期间上游接口变过三次——有一次是SupplierA把status字段的值从数字改成了字符串，有一次是SupplierC新增了一个amount字段但把类型设成了null，还有一次SupplierB直接改了域名。每一次变更，系统都没有静默吞掉错误数据，而是在校验层精确地拦截并记录了问题。负责对接的同事根据错误记录去跟供应商沟通，修复周期从原来的”等用户反馈才发现”缩短到了小时级别。

总结几个值得带走的东西：

• 把数据校验放在系统边界上。数据一进来就做严格校验，不合规的当场拦住，不要让脏数据流到业务逻辑里。Pydantic的BaseModel就是一道很好的边界墙。
• 适配器模式天然适合多源数据整合。每个上游对应一个适配器，映射逻辑内聚在适配器里，不会互相污染。新增一个供应商只需要新增一个适配器类，别的代码不用动。
• 错误记录和自动修复是生产系统的基本素养。光抛异常不够，要把异常信息结构化地存下来，能自动修的就自动修，修不了的及时通知到人。JSONL是一种轻量好用的持久化格式。
• 类型提示和模型定义是活的文档。花时间把模型写清楚，后面维护的时候会感谢自己。生成的JSON Schema还能直接对接上下游工具链。

如果你也在维护类似的数据处理系统，不妨挑一个模块试试这个思路。从一个小的数据模型开始，用Pydantic定义好输入输出的形状，然后逐步把校验逻辑从业务代码里抽出来。这个过程本身就会暴露出很多之前隐藏的问题——但这是好事，早发现早处理，总比线上炸了再修要强。