FastAPI Pydantic动态调整Schema
title: FastAPI Pydantic动态调整Schema
from pydantic import BaseModel
class DynamicModel(BaseModel):
@classmethod
def __get_pydantic_json_schema__(cls, core_schema, handler):
schema = handler(core_schema)
if os.getenv("ENV") == "prod":
schema["required"].append("audit_trail")
return schema
class ProdModel(DynamicModel):
audit_trail: Optional[str]
from pydantic import Field
def dynamic_field(config: dict):
return Field(
json_schema_extra={
"x-ui-config": config
}
)
class UIModel(BaseModel):
username: str = dynamic_field({"widget": "search"})
from pydantic import validator
class ConditionalModel(BaseModel):
user_type: str
company: Optional[str] = None
@classmethod
def __get_pydantic_json_schema__(cls, core_schema, handler):
schema = handler(core_schema)
if "company" in schema["properties"]:
schema["required"].append("company")
return schema
@validator("user_type")
def check_user_type(cls, v):
if v == "enterprise":
cls.__fields__["company"].required = True
return v
class TenantAwareSchema(BaseModel):
class Config:
extra = "allow"
@classmethod
def __get_pydantic_json_schema__(cls, core_schema, handler):
schema = handler(core_schema)
tenant = get_current_tenant()
if tenant != "admin":
del schema["properties"]["sensitive_field"]
return schema
from feature_flag import FeatureFlag
class FeatureModel(BaseModel):
@classmethod
def __get_pydantic_json_schema__(cls, core_schema, handler):
schema = handler(core_schema)
if FeatureFlag.is_enabled("new_checkout"):
schema["properties"]["payment"]["x-component"] = "v2-payment"
return schema
from functools import lru_cache
class CachedSchema(BaseModel):
@classmethod
@lru_cache(maxsize=128)
def schema(cls, **kwargs):
return super().schema(**kwargs)
class HighTrafficModel(CachedSchema):
data: dict
class DeltaSchema(BaseModel):
base_schema: dict
delta: dict
@classmethod
def __get_pydantic_json_schema__(cls, core_schema, handler):
schema = handler(core_schema)
return apply_json_patch(schema, cls.delta)
try:
class ConflictingSchema(BaseModel):
@classmethod
def __get_pydantic_json_schema__(cls, *args):
return {"type": "object", "properties": {"id": {"type": "string"}}}
except SchemaConflictError as e:
print(f"Schema冲突: {e}")
class VersionedSchema(BaseModel):
_schema_versions = []
@classmethod
def rollback_schema(cls, version: int):
cls.__get_pydantic_json_schema__ = cls._schema_versions[version]
| 错误码 | 现象 | 解决方案 |
|---|---|---|
| 422 | 动态字段验证失败 | 检查字段类型兼容性 |
| 500 | Schema生成超时 | 启用缓存和增量更新 |
| 409 | 字段定义冲突 | 使用版本隔离策略 |
| 401 | 未授权字段访问 | 加强租户权限校验 |
评论
发表评论