首页 > Python基础教程 >
-
Pydantic Schema生成指南:自定义JSON Schema
第一章:Schema生成基础
1.1 默认Schema生成机制
PYTHON
from pydantic import BaseModel
class User(BaseModel):
id: int
name: str = Field(..., max_length=50)
print(User.schema_json(indent=2))
输出特征:
JSON
{
"title": "User",
"type": "object",
"properties": {
"id": {
"title": "Id",
"type": "integer"
},
"name": {
"title": "Name",
"type": "string",
"maxLength": 50
}
},
"required": [
"id",
"name"
]
}
1.2 Schema生成流程
graph TD
A[字段定义] --> B[元数据收集]
B --> C[类型映射]
C --> D[约束转换]
D --> E[Schema组装]
第二章:核心定制技术
2.1 字段级元数据注入
PYTHON
from pydantic import BaseModel, Field
class Product(BaseModel):
sku: str = Field(
...,
json_schema_extra={
"x-frontend": {"widget": "search-input"},
"x-docs": {"example": "ABC-123"}
}
)
print(Product.schema()["properties"]["sku"])
输出:
JSON
{
"title": "Sku",
"type": "string",
"x-frontend": {
"widget": "search-input"
},
"x-docs": {
"example": "ABC-123"
}
}
2.2 类型映射重载
PYTHON
from pydantic import BaseModel
from pydantic.json_schema import GenerateJsonSchema
class CustomSchemaGenerator(GenerateJsonSchema):
def generate(self, schema):
if schema["type"] == "string":
schema["format"] = "custom-string"
return schema
class DataModel(BaseModel):
content: str
print(DataModel.schema(schema_generator=CustomSchemaGenerator))
第三章:动态Schema生成
3.1 运行时Schema构建
PYTHON
from pydantic import create_model
from pydantic.fields import FieldInfo
def dynamic_model(field_defs: dict):
fields = {}
for name, config in field_defs.items():
fields[name] = (
config["type"],
FieldInfo(**config["field_params"])
)
return create_model('DynamicModel', **fields)
model = dynamic_model({
"timestamp": {
"type": int,
"field_params": {"ge": 0, "json_schema_extra": {"unit": "ms"}}
}
})
3.2 环境感知Schema
PYTHON
from pydantic import BaseModel, ConfigDict
class EnvAwareSchema(BaseModel):
model_config = ConfigDict(json_schema_mode="dynamic")
@classmethod
def __get_pydantic_json_schema__(cls, core_schema, handler):
schema = handler(core_schema)
if os.getenv("ENV") == "prod":
schema["required"].append("audit_info")
return schema
第四章:企业级应用模式
4.1 OpenAPI增强方案
PYTHON
from pydantic import BaseModel
class OpenAPICompatible(BaseModel):
model_config = dict(
json_schema_extra={
"components": {
"schemas": {
"ErrorResponse": {
"type": "object",
"properties": {
"code": {"type": "integer"},
"message": {"type": "string"}
}
}
}
}
}
)
4.2 版本化Schema管理
PYTHON
from pydantic import BaseModel, field_validator
class VersionedModel(BaseModel):
model_config = ConfigDict(extra="allow")
@classmethod
def __get_pydantic_json_schema__(cls, core_schema, handler):
schema = handler(core_schema)
schema["x-api-version"] = "2.3"
return schema
class V1Model(VersionedModel):
@classmethod
def __get_pydantic_json_schema__(cls, *args):
schema = super().__get_pydantic_json_schema__(*args)
schema["x-api-version"] = "1.2"
return schema
第五章:错误处理与优化
5.1 Schema验证错误
PYTHON
try:
class InvalidSchemaModel(BaseModel):
data: dict = Field(format="invalid-format")
except ValueError as e:
print(f"Schema配置错误: {e}")
5.2 性能优化策略
PYTHON
from functools import lru_cache
class CachedSchemaModel(BaseModel):
@classmethod
@lru_cache(maxsize=128)
def schema(cls, **kwargs):
return super().schema(**kwargs)
课后Quiz
Q1:如何添加自定义Schema属性?
A) 使用json_schema_extra
B) 修改全局配置
C) 继承GenerateJsonSchema
Q2:处理版本兼容的正确方式?
动态注入版本号
创建子类覆盖Schema
维护多个模型
Q3:优化Schema生成性能应使用?
LRU缓存
增加校验步骤
禁用所有缓存
错误解决方案速查表
错误信息 原因分析 解决方案
ValueError: 无效的format类型 不支持的Schema格式 检查字段类型与格式的兼容性
KeyError: 缺失必需字段 动态Schema未正确注入 验证__get_pydantic_json_schema__实现
SchemaGenerationError 自定义生成器逻辑错误 检查类型映射逻辑
MemoryError 大规模模型未缓存 启用模型Schema缓存
架构箴言:Schema设计应遵循”契约优先”原则,建议使用Git版本控制管理Schema变更,通过CI/CD流水线实现Schema的自动化测试与文档生成,建立Schema变更通知机制保障多团队协作。
来源:https://blog.cmdragon.cn/posts/3bd5ffd5fdcb/