Pydantic 集成

Robyn 支持 Pydantic v2 作为可选依赖,用于自动请求体验证和丰富的 OpenAPI 文档生成。验证是按处理函数选择启用的 — 只有当您使用 Pydantic BaseModel 注解参数时才会激活。未使用 Pydantic 注解的处理函数完全不受影响:不进行解析、不进行验证、无额外开销。当未安装 Pydantic 时,Robyn 不会导入它。

安装

使用可选扩展安装带 Pydantic 支持的 Robyn:

安装

pip install "robyn[pydantic]"

robyn[all] 包含 Pydantic、Jinja2 模板引擎以及未来的所有可选功能。

基本用法

定义一个 Pydantic BaseModel,并将其用作处理函数参数的类型注解。Robyn 将自动解析传入的 JSON 请求体,根据模型进行验证,并将验证后的实例注入到处理函数中。

基本 Pydantic 验证

from pydantic import BaseModel
from robyn import Robyn

app = Robyn(__file__)


class UserCreate(BaseModel):
    name: str
    email: str
    age: int
    active: bool = True


@app.post("/users")
def create_user(user: UserCreate):
    """创建新用户"""
    return {
        "name": user.name,
        "email": user.email,
        "age": user.age,
        "active": user.active,
    }


if __name__ == "__main__":
    app.start()

验证错误

当请求体验证失败时,Robyn 会自动返回 422 Unprocessable Entity 响应,并附带结构化的错误详情。您无需编写任何错误处理代码。

例如,发送 {"name": "Alice", "email": "alice@example.com", "age": "not_a_number"} 将产生:

{
  "error": "Validation Error",
  "detail": [
    {
      "type": "int_parsing",
      "loc": ["age"],
      "msg": "Input should be a valid integer, unable to parse string as an integer",
      "input": "not_a_number"
    }
  ]
}

缺少必填字段也会被捕获:

{
  "error": "Validation Error",
  "detail": [
    {
      "type": "missing",
      "loc": ["email"],
      "msg": "Field required",
      "input": {"name": "Alice", "age": 30}
    }
  ]
}

嵌套模型

Pydantic 模型可以引用其他模型。Robyn 会自动处理嵌套验证。

嵌套模型

from pydantic import BaseModel
from robyn import Robyn

app = Robyn(__file__)


class Address(BaseModel):
    street: str
    city: str
    zip_code: str


class UserWithAddress(BaseModel):
    name: str
    email: str
    address: Address


@app.post("/users")
def create_user(data: UserWithAddress):
    """创建带有地址的用户"""
    return {"name": data.name, "city": data.address.city}

如果嵌套的 address 对象缺失或格式错误,Robyn 将返回 422 响应,并包含完整的错误路径(例如 ["address", "city"])。

与 Request 对象配合使用

您可以在同一个处理函数中同时使用 Pydantic 参数和标准的 Request 对象。这样您可以在获取验证后的请求体的同时,访问请求头、查询参数和其他请求元数据。

Pydantic + Request

from pydantic import BaseModel
from robyn import Robyn, Request

app = Robyn(__file__)


class UserCreate(BaseModel):
    name: str
    email: str
    age: int
    active: bool = True


@app.post("/users")
def create_user(request: Request, user: UserCreate):
    """创建用户 — 同时访问原始请求和验证后的模型"""
    return {
        "method": request.method,
        "name": user.name,
        "email": user.email,
    }

直接返回 Pydantic 模型

您可以直接从处理函数返回 Pydantic 模型实例(或模型列表)。Robyn 会自动将其序列化为 JSON 并设置正确的 Content-Type 头 — 无需手动调用 .model_dump()

返回模型

@app.post("/users")
def create_user(user: UserCreate) -> UserCreate:
    """验证并直接返回用户"""
    return user

两种形式都会生成 application/json 响应。单个模型路径使用 Pydantic 基于 Rust 的 model_dump_json() 以获得最大吞吐量。

验证触发机制

Pydantic 验证是基于注解驱动的,而非基于 HTTP 方法。路由器在注册时检查每个处理函数的签名;任何使用 BaseModel 子类注解的参数都会在路由被调用时自动触发 request.body 的验证。这适用于所有 HTTP 方法 — POSTPUTPATCHDELETE 或任何携带请求体的方法。

任意 HTTP 方法

@app.put("/users/:id")
def update_user(user: UserCreate):
    return {"updated": True, "name": user.name}

OpenAPI 集成

当您使用 Pydantic 模型时,Robyn 会自动在 /openapi.json 的 OpenAPI 规范中生成丰富的 JSON Schema。包括:

  • 属性类型stringintegerboolean
  • 必填字段 — 没有默认值的字段会列在 required
  • 默认值 — 在 Schema 中展示
  • 嵌套模型 — 通过 $ref 引用,并放置在 components/schemas

OpenAPI 与 Pydantic

from pydantic import BaseModel
from robyn import Robyn, Request

app = Robyn(__file__)


class Address(BaseModel):
    street: str
    city: str
    zip_code: str


class UserWithAddress(BaseModel):
    name: str
    email: str
    address: Address


@app.post("/users", openapi_tags=["Users"])
def create_user(request: Request, data: UserWithAddress) -> dict:
    """创建带有嵌套地址的用户"""
    return {"name": data.name, "city": data.address.city}

生成的 /openapi.json 将包含:

{
  "paths": {
    "/users": {
      "post": {
        "requestBody": {
          "content": {
            "application/json": {
              "schema": {
                "type": "object",
                "properties": {
                  "name": {"type": "string", "title": "Name"},
                  "email": {"type": "string", "title": "Email"},
                  "address": {"$ref": "#/components/schemas/Address"}
                },
                "required": ["name", "email", "address"],
                "title": "UserWithAddress"
              }
            }
          }
        }
      }
    }
  },
  "components": {
    "schemas": {
      "Address": {
        "type": "object",
        "properties": {
          "street": {"type": "string", "title": "Street"},
          "city": {"type": "string", "title": "City"},
          "zip_code": {"type": "string", "title": "Zip Code"}
        },
        "required": ["street", "city", "zip_code"],
        "title": "Address"
      }
    }
  }
}

Pydantic 与 Body 对比

Robyn 支持两种类型化请求体的方式。根据您的需求选择合适的方式:

特性Body 子类Pydantic BaseModel
安装内置pip install "robyn[pydantic]"
验证无自动验证完整验证,附带详细错误信息
错误响应手动处理自动返回 422 结构化错误
返回序列化手动 dict()自动序列化模型为 JSON
OpenAPI Schema基本类型推断完整 JSON Schema(类型、必填、默认值、$ref
嵌套模型支持(基本)支持(OpenAPI 中使用 $ref
性能开销仅在安装并使用 Pydantic 时

两种方式都支持 OpenAPI 文档。如果您需要验证功能,请使用 Pydantic。如果只需要 OpenAPI Schema 提示而不需要验证,Body 即可满足需求。

重要说明

  • 按处理函数选择启用 — 验证仅在使用 Pydantic BaseModel 注解参数的处理函数上运行。所有其他处理函数(使用 BodyRequest、路径参数等)行为完全不变,无任何额外开销。
  • 每个处理函数只能有一个 Pydantic 请求体参数 — 每个处理函数最多只能有一个使用 Pydantic 模型注解的参数。整个请求体会被解析到这个模型中。如果需要多个模型输入,请将它们组合成一个包含嵌套字段的父模型。
  • 仅验证请求 — Robyn 根据 Pydantic 模型验证传入的请求体,但不验证传出的响应。当您返回模型实例时,它会直接序列化而不进行重新验证。这是出于性能考虑的设计决策 — 如果您构造了模型,它已经是有效的。

下一步

蝙蝠侠想知道 Robyn 的处理函数是否可以分发到多个进程中执行。

Robyn 向他展示了方法!

多进程执行