一直被错误设计的 HTTP API 参数校验模式
传统 Web framework 或者 HTTP API framework 对于接口参数校验的模式通常如下：
- 为每个参数绑定一个 Validator 定义
- 调用 Validation 对所有参数进行校验
- 如果验证失败则返回客户端一个错误列表，包含每一个验证失败的参数以及失败原因
但上述模式是一种错误的设计，是一种职责错位。接口调用本身隐含着契约，应该由调用者负责遵守契约，被调用者负责拒绝违反契约的调用。而不是被调用者逐项校验参数、聚合结果并返回一份错误列表。对于违反调用契约的参数，接口应当尽早失败（fail-fast），直接拒绝调用并返回 ContractViolation 错误。换句话说，接口不负责用户表单验证。如有类似需求，应该另外设计一个专用的用户表单验证接口。
以网页应用为例，在调用后端接口前，前端代码应该保证调用参数符合契约定义，而不应该依赖后端返回的错误消息列表来纠正请求参数。这种错误消息列表本质上是面向人类用户阅读的，仅在验证用户输入时才有意义。因此，应该将验证用户输入的功能从业务接口中剥离。用户输入的验证逻辑可以完全集成在前端代码里，也可以单独设计一个针对用户输入内容的验证接口。
当然，这一切的前提是接口契约必须有明确、规范且可供调用方遵循的文档定义。
这里以 FastAPI 为例，按照传统设计的接口代码如下：
pythonfrom fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field
app = FastAPI()
class User(BaseModel):
name: str = Field(min_length=1)
age: int | None = None
email: EmailStr
@app.post("/users")
def create_user(user: User):
return {
"name": user.name,
"age": user.age,
"email": user.email,
}这里接口的契约如下：
name必填，且长度至少为 1age可选，或者是一个整数email必填，且必须是有效的邮箱地址
如果 POST 提交如下数据：
json{
"name": "",
"age": "not a number",
"email": "invalid email address"
}则接口会返回 422 响应，其中包含一个错误列表：
json{
"detail": [
{
"type": "string_too_short",
"loc": [
"body",
"name"
],
"msg": "String should have at least 1 character",
"input": "",
"ctx": {
"min_length": 1
}
},
{
"type": "int_parsing",
"loc": [
"body",
"age"
],
"msg": "Input should be a valid integer, unable to parse string as an integer",
"input": "not a number"
},
{
"type": "value_error",
"loc": [
"body",
"email"
],
"msg": "value is not a valid email address: An email address must have an @-sign.",
"input": "invalid email address",
"ctx": {
"reason": "An email address must have an @-sign."
}
}
]
}但接口其实只需要返回一个 ContractViolation 错误响应即可：
json{
"error": {
"id": "......",
"code": "ContractViolation",
"message": "The `name` parameter violates the API contract."
}
}接口真正的责任是验证参数是否满足业务规则。比如 name 和 email 是否已经被其他用户占用。模拟代码如下：
pythonfrom uuid import uuid4
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, EmailStr, Field
app = FastAPI()
class User(BaseModel):
name: str = Field(min_length=1)
age: int | None = None
email: EmailStr
class BusinessRuleViolation(Exception):
def __init__(self, code: str, message: str):
self.code = code
self.message = message
@app.exception_handler(RequestValidationError)
async def contract_violation_handler(
request: Request,
exc: RequestValidationError,
):
error = exc.errors()[0]
location = error["loc"]
parameter = next(
(str(item) for item in location if item != "body"),
"request",
)
return JSONResponse(
status_code=400,
content={
"error": {
"id": str(uuid4()),
"code": "ContractViolation",
"message": (
f"The `{parameter}` parameter "
"violates the API contract."
),
}
},
)
@app.exception_handler(BusinessRuleViolation)
async def business_rule_violation_handler(
request: Request,
exc: BusinessRuleViolation,
):
return JSONResponse(
status_code=409,
content={
"error": {
"id": str(uuid4()),
"code": exc.code,
"message": exc.message,
}
},
)
def user_name_exists(name: str) -> bool:
# Simulate a database query.
return name == "alice"
def user_email_exists(email: str) -> bool:
# Simulate a database query.
return str(email) == "[email protected]"
@app.post("/users")
def create_user(user: User):
if user_name_exists(user.name):
raise BusinessRuleViolation(
"UserNameAlreadyExists",
"The user name is already in use.",
)
if user_email_exists(user.email):
raise BusinessRuleViolation(
"EmailAlreadyExists",
"The email address is already in use.",
)
# Create the user here.
return {
"name": user.name,
"age": user.age,
"email": user.email,
}不过有时候的确存在接口验证表单输入的需求。比如，接口契约经常随产品需求变动，或者契约规则相对复杂，前后端维护两套验证 Schema 成本较高。对此的建议是，提供一个专门的表单验证接口。比如 /users 接口对应 /users.validate 接口，两者共用同一套 Schema 即可。对于 Python，甚至可以简单地通过装饰器来实现：
pythonfrom functools import wraps
from typing import Callable
from uuid import uuid4
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, EmailStr, Field, ValidationError
app = FastAPI()
# ----------------------------------------------------------------------
# API contract
# ----------------------------------------------------------------------
class User(BaseModel):
name: str = Field(min_length=1)
age: int | None = None
email: EmailStr
# ----------------------------------------------------------------------
# Business rule violation
# ----------------------------------------------------------------------
class BusinessRuleViolation(Exception):
def __init__(self, code: str, message: str):
self.code = code
self.message = message
# ----------------------------------------------------------------------
# API error handlers
# ----------------------------------------------------------------------
@app.exception_handler(RequestValidationError)
async def contract_violation_handler(
request: Request,
exc: RequestValidationError,
):
"""
The API contract was violated.
This is deliberately fail-fast: only the first violation is
reported to the caller.
"""
error = exc.errors()[0]
parameter = next(
(
str(item)
for item in error["loc"]
if item not in ("body", "query", "path")
),
"request",
)
return JSONResponse(
status_code=400,
content={
"error": {
"id": str(uuid4()),
"code": "ContractViolation",
"message": (
f"The `{parameter}` parameter "
"violates the API contract."
),
}
},
)
@app.exception_handler(BusinessRuleViolation)
async def business_rule_violation_handler(
request: Request,
exc: BusinessRuleViolation,
):
return JSONResponse(
status_code=409,
content={
"error": {
"id": str(uuid4()),
"code": exc.code,
"message": exc.message,
}
},
)
# ----------------------------------------------------------------------
# Form validation decorator
# ----------------------------------------------------------------------
def form_validation(
schema: type[BaseModel],
):
"""
Add a corresponding `.validate` endpoint.
For example:
@form_validation(User)
@app.post("/users")
def create_user(...):
...
automatically creates:
POST /users.validate
"""
def decorator(func: Callable):
# The path is taken from the FastAPI route decorator.
# FastAPI stores it in the function's __wrapped__ chain only
# after registration, so we derive it from the function name
# convention here instead of relying on internal FastAPI APIs.
#
# For this example, the endpoint name is create_user -> /users.
endpoint_name = func.__name__
if endpoint_name.startswith("create_"):
resource = endpoint_name[len("create_"):]
path = f"/{resource}.validate"
else:
raise ValueError(
"The endpoint function must use the "
"`create_<resource>` naming convention."
)
@wraps(func)
async def validate(request: Request):
try:
data = await request.json()
schema.model_validate(data)
except ValidationError as exc:
errors = []
for error in exc.errors():
errors.append(
{
"field": ".".join(
str(item)
for item in error["loc"]
),
"code": error["type"],
"message": error["msg"],
}
)
return JSONResponse(
status_code=200,
content={
"valid": False,
"errors": errors,
},
)
return JSONResponse(
status_code=200,
content={
"valid": True,
},
)
app.post(path)(validate)
return func
return decorator
# ----------------------------------------------------------------------
# Business rules
# ----------------------------------------------------------------------
def user_name_exists(name: str) -> bool:
# Simulate a database query.
return name == "alice"
def user_email_exists(email: str) -> bool:
# Simulate a database query.
return email == "[email protected]"
# ----------------------------------------------------------------------
# API endpoint
# ----------------------------------------------------------------------
@form_validation(User)
@app.post("/users")
def create_user(user: User):
"""
The API only receives data that satisfies the API contract.
Business validation is performed here.
"""
if user_name_exists(user.name):
raise BusinessRuleViolation(
"UserNameAlreadyExists",
"The user name is already in use.",
)
if user_email_exists(user.email):
raise BusinessRuleViolation(
"EmailAlreadyExists",
"The email address is already in use.",
)
# Create the user here.
return {
"name": user.name,
"age": user.age,
"email": user.email,
}新增的 /users.validate 接口返回表单验证结果：
json{
"valid": false,
"errors": [
{
"field": "name",
"code": "string_too_short",
"message": "String should have at least 1 character"
},
{
"field": "age",
"code": "int_parsing",
"message": "Input should be a valid integer, unable to parse string as an integer"
},
{
"field": "email",
"code": "value_error",
"message": "value is not a valid email address: An email address must have an @-sign."
}
]
}特别要注意的是，这里的 users.validate 接口，在 "valid": false 的情况下，返回的 HTTP 状态码仍是 200，而不是 4XX。它表示的是验证结果，而不是接口调用失败。当然，这个 *.validate 接口只是可选项，名字也是随便起的，可以视具体业务按需实现。