voluptuous，一个非常实用的 Python 库！

线上接口最烦的一类报错，不是 500。

是那种字段传错了，但日志里只剩一句：

order submit failed: invalid payload

然后你顺着链路翻半天，发现前端把 amount 从数字传成了字符串，pay_type 多传了个空格，items 里面还有个商品数量是 0。

这种问题我一般不想在业务代码里一层层 if。

太脏。

Python 里有个库，叫 voluptuous，干的就是这类脏活：校验字典、配置、接口参数、JSON 数据。它不像 Pydantic 那么重，也不像手写判断那么散。小脚本、后台任务、内部接口，用起来很顺手。

安装就一行：

pip install voluptuous

先看一个接口参数。

from voluptuous import Schema, Required, Optional, All, Length, Range, In, Coerce, MultipleInvalid
order_schema = Schema({
    Required("user_id"): All(str, Length(min=6, max=32)),
    Required("amount"): All(Coerce(float), Range(min=0.01)),
    Required("pay_type"): In(["wechat", "alipay", "balance"]),
    Optional("remark", default=""): All(str, Length(max=120)),
})

这里我最喜欢的是 Coerce。

实际接口里你很少能保证上游传来的类型完全干净。尤其是管理后台、Excel 导入、老系统对接，金额字段今天是 12.8，明天可能就是 "12.8"。

不用 Coerce 的话，代码大概率会写成这样：

amount = data.get("amount")
try:
    amount = float(amount)
except Exception:
raise ValueError("amount error")
if amount <= 0:
raise ValueError("amount error")

这种代码写多了，业务函数就开始发臭。校验、转换、业务判断混在一起，后面排查问题很难受。

用 voluptuous 可以把入口先收干净：

defsubmit_order(raw):
try:
        data = order_schema(raw)
except MultipleInvalid as e:
        print(f"bad order payload: {e}")
return {"ok": False, "msg": str(e)}
# 走到这里，amount 已经是 float，remark 也有默认值
    print("create order:", data)
return {"ok": True}

试一下：

submit_order({
"user_id": "u10086",
"amount": "19.90",
"pay_type": "wechat"
})

拿到的数据会变成：

{
"user_id": "u10086",
"amount": 19.9,
"pay_type": "wechat",
"remark": ""
}

这地方很适合放在接口入口、消息消费入口、批量导入入口。

我见过不少 Python 项目，参数校验全靠业务函数里面随手判断。刚开始没问题，字段一多就开始乱。尤其是 MQ 消费，消息一旦有脏数据，消费失败重试，重试失败又进死信，最后查半天，发现只是一个字段少了。

这种地方就应该先把消息拦下来。

from voluptuous import Schema, Required, Optional, All, Coerce, Range, In
stock_msg_schema = Schema({
    Required("sku"): All(str, Length(min=3, max=40)),
    Required("warehouse"): In(["bj", "sh", "gz"]),
    Required("change"): All(Coerce(int), Range(min=-9999, max=9999)),
    Optional("trace_id", default="-"): str,
})

消费时不要上来就改库存：

defhandle_stock_message(msg):
try:
        event = stock_msg_schema(msg)
except MultipleInvalid as e:
        print(f"drop dirty stock msg, err={e}, raw={msg}")
return
if event["change"] == 0:
        print(f"ignore empty stock change, trace={event['trace_id']}")
return
    print(
f"apply stock, sku={event['sku']}, "
f"wh={event['warehouse']}, change={event['change']}, "
f"trace={event['trace_id']}"
    )

我这里会直接把脏消息丢掉，还是重试，要看业务。

库存、资金、订单状态这类消息，不能随便丢，最好落异常表。日志、埋点、同步缓存这种消息，字段错了还无限重试，就有点自己折磨自己了。

voluptuous 还有个很实用的点，能校验嵌套结构。

比如一个订单里有商品列表：

item_schema = Schema({
    Required("sku"): All(str, Length(min=3, max=40)),
    Required("qty"): All(Coerce(int), Range(min=1, max=999)),
    Optional("gift", default=False): bool,
})
checkout_schema = Schema({
    Required("buyer"): All(str, Length(min=6, max=32)),
    Required("items"): All([item_schema], Length(min=1)),
    Optional("coupon"): str,
})

这个写法比自己循环判断清爽很多。

defcheckout(raw):
try:
        data = checkout_schema(raw)
except MultipleInvalid as e:
        print(f"checkout payload invalid: {e}")
return
    total_qty = sum(x["qty"] for x in data["items"])
    print(f"buyer={data['buyer']}, total_qty={total_qty}")

注意这里的 [item_schema]，表示列表里的每个元素都要符合 item_schema。

这东西看着简单，但批量导入特别有用。

比如运营给你一个 JSON 文件，里面几百条商品配置。你要是不先校验，导到一半才发现某条数据少字段，那回滚、补偿、重跑都麻烦。

我一般会先写一个小校验脚本：

import json
from pathlib import Path
from voluptuous import Schema, Required, Optional, All, Coerce, Range, Length, MultipleInvalid
product_schema = Schema({
    Required("sku"): All(str, Length(min=3, max=40)),
    Required("name"): All(str, Length(min=1, max=80)),
    Required("price"): All(Coerce(float), Range(min=0)),
    Optional("online", default=False): bool,
})
defcheck_file(path):
    rows = json.loads(Path(path).read_text(encoding="utf-8"))
    bad = []
for idx, row in enumerate(rows, 1):
try:
            product_schema(row)
except MultipleInvalid as e:
            bad.append((idx, str(e), row.get("sku")))
if bad:
for idx, err, sku in bad[:20]:
            print(f"line={idx}, sku={sku}, err={err}")
raise SystemExit(f"found dirty rows: {len(bad)}")
    print(f"all clean, rows={len(rows)}")

这个脚本不复杂，但很值。

因为它把问题拦在导入前，而不是让问题进数据库以后再补锅。

还有配置文件也适合用它。

Python 项目里经常会有这种配置：

raw_config = {
"redis": {
"host": "127.0.0.1",
"port": "6379",
"db": "0"
    },
"worker": {
"threads": "8",
"mode": "fast"
    }
}

端口、线程数这种字段，不校验也能跑。直到某天配置写成 "8c"，程序启动到一半才炸。

config_schema = Schema({
    Required("redis"): {
        Required("host"): str,
        Required("port"): All(Coerce(int), Range(min=1, max=65535)),
        Optional("db", default=0): All(Coerce(int), Range(min=0, max=15)),
    },
    Required("worker"): {
        Optional("threads", default=4): All(Coerce(int), Range(min=1, max=64)),
        Optional("mode", default="safe"): In(["safe", "fast"]),
    }
})

启动时先跑一下：

defload_config(raw):
try:
return config_schema(raw)
except MultipleInvalid as e:
raise RuntimeError(f"bad app config: {e}") from e

配置错了，启动阶段就死。

我更喜欢这种死法。比程序跑起来以后，某个后台线程半夜悄悄挂掉强多了。

不过 voluptuous 也不是万能的。

它适合轻量校验，适合字典结构，适合脚本和服务入口。你要是做大型 API，字段很多，还要自动生成文档、IDE 类型提示、复杂模型转换，那 Pydantic 更合适。

但很多内部工具根本不需要那么重。

一个定时脚本，一个数据清洗任务，一个 MQ 消费者，一个后台接口，把入口数据校验干净，就能少掉一堆低级脏问题。

我对这类库的判断很简单：它不应该让业务代码变复杂。

voluptuous 这个库好的地方就在这。校验规则单独放，业务函数拿到的就是相对干净的数据。出错时也能把错误卡在入口，不让脏数据继续往后流。

写 Python 小工具时，我一般会把它放在 schema.py 里。

业务代码不关心字段怎么来的，只关心字段已经对了。

这就够了。