Pydantic v1:
  Python 层：Field → Validator → Model
  全部 Python 实现 → 性能受限

Pydantic v2:
  Python 层：Field → Model 定义
       ↓ 编译
  Rust 层 (pydantic-core):
    CoreSchema → Compiled Validator → 高速验证
                → Compiled Serializer → 高速序列化

from pydantic import BaseModel

class OntologyInstance(BaseModel):
    id: str
    object_type: str
    properties: dict[str, Any]
    version: int = 0

# Pydantic v2 内部编译流程：
# 1. 分析类型注解 → 生成 CoreSchema（JSON 描述）
# 2. CoreSchema → Rust Validator（编译期完成）
# 3. 运行时调用 Validator 执行验证（Rust 速度）

# 查看 CoreSchema
print(OntologyInstance.__pydantic_core_schema__)
# {
#   'type': 'model',
#   'cls': <class 'OntologyInstance'>,
#   'schema': {
#     'type': 'model-fields',
#     'fields': {
#       'id': {'type': 'model-field', 'schema': {'type': 'str'}},
#       'object_type': {'type': 'model-field', 'schema': {'type': 'str'}},
#       'properties': {'type': 'model-field', 'schema': {'type': 'dict', ...}},
#       'version': {'type': 'model-field', 'schema': {'type': 'default', ...}},
#     }
#   }
# }

from pydantic import BaseModel, ConfigDict

class StrictModel(BaseModel):
    model_config = ConfigDict(strict=True)

    count: int
    name: str

# Strict 模式：不接受类型强制转换
StrictModel(count=42, name="test")     # ✅
StrictModel(count="42", name="test")   # ❌ ValidationError: int expected

class LaxModel(BaseModel):
    # 默认 Lax 模式
    count: int
    name: str

LaxModel(count="42", name="test")      # ✅ "42" → 42 自动转换
LaxModel(count="abc", name="test")     # ❌ 无法转换

from datetime import datetime
from typing import Any, Literal
from pydantic import BaseModel, Field, ConfigDict, field_validator, model_validator

class OntologyInstanceBase(BaseModel):
    """Ontology 实例基类"""
    model_config = ConfigDict(
        from_attributes=True,          # 支持 ORM 模式
        populate_by_name=True,         # 支持字段别名
        str_strip_whitespace=True,     # 自动去除字符串空白
        validate_default=True,         # 验证默认值
        use_enum_values=True,          # Enum 使用值而非名称
        json_schema_extra={
            "examples": [
                {
                    "id": "inst-001",
                    "object_type": "Employee",
                    "properties": {"name": "Alice", "age": 30},
                }
            ]
        },
    )

    id: str = Field(..., min_length=1, max_length=255, pattern=r"^[a-zA-Z0-9_-]+$")
    object_type: str = Field(..., alias="objectType", min_length=1, max_length=100)
    world_id: str = Field(..., alias="worldId")
    properties: dict[str, Any] = Field(default_factory=dict)
    version: int = Field(default=0, ge=0)
    created_at: datetime | None = Field(default=None, alias="createdAt")
    updated_at: datetime | None = Field(default=None, alias="updatedAt")

    @field_validator("object_type")
    @classmethod
    def validate_object_type(cls, v: str) -> str:
        """ObjectType 必须是 PascalCase"""
        if not v[0].isupper():
            raise ValueError(f"ObjectType must be PascalCase: {v}")
        return v

    @model_validator(mode="after")
    def validate_timestamps(self) -> "OntologyInstanceBase":
        """updated_at 不能早于 created_at"""
        if self.created_at and self.updated_at:
            if self.updated_at < self.created_at:
                raise ValueError("updated_at cannot be before created_at")
        return self

class PropertyDefinition(BaseModel):
    """Ontology 属性定义"""
    name: str = Field(..., min_length=1)
    data_type: Literal["STRING", "INTEGER", "FLOAT", "BOOLEAN", "DATETIME", "ARRAY", "MAP"]
    required: bool = False
    default_value: Any = None
    description: str = ""
    constraints: dict[str, Any] = Field(default_factory=dict)

    @field_validator("default_value")
    @classmethod
    def validate_default_matches_type(cls, v, info):
        if v is None:
            return v
        data_type = info.data.get("data_type")
        type_map = {
            "STRING": str, "INTEGER": int, "FLOAT": (int, float),
            "BOOLEAN": bool, "DATETIME": str,
        }
        expected = type_map.get(data_type)
        if expected and not isinstance(v, expected):
            raise ValueError(f"Default value type mismatch: expected {data_type}")
        return v

class ObjectTypeSchema(BaseModel):
    """ObjectType 完整 Schema"""
    name: str
    display_name: str = ""
    properties: list[PropertyDefinition] = Field(default_factory=list)
    primary_key: list[str] = Field(default_factory=lambda: ["id"])
    indexes: list[list[str]] = Field(default_factory=list)

    @model_validator(mode="after")
    def validate_primary_key_exists(self) -> "ObjectTypeSchema":
        prop_names = {p.name for p in self.properties}
        for pk in self.primary_key:
            if pk != "id" and pk not in prop_names:
                raise ValueError(f"Primary key '{pk}' not in properties")
        return self

from pydantic import field_validator, field_serializer
from typing import Annotated
from pydantic import AfterValidator, BeforeValidator

# 方式 1：装饰器
class ActionRequest(BaseModel):
    action_type: str
    parameters: dict[str, Any]

    @field_validator("action_type")
    @classmethod
    def validate_action_type(cls, v: str) -> str:
        valid_types = {"CREATE", "UPDATE", "DELETE", "CUSTOM"}
        if v.upper() not in valid_types:
            raise ValueError(f"Invalid action type: {v}")
        return v.upper()

# 方式 2：Annotated 类型
def validate_positive(v: int) -> int:
    if v <= 0:
        raise ValueError("Must be positive")
    return v

PositiveInt = Annotated[int, AfterValidator(validate_positive)]

class PaginationParams(BaseModel):
    limit: PositiveInt = Field(default=100, le=1000)
    offset: Annotated[int, Field(ge=0)] = 0

class OntologyInstance(BaseModel):
    id: str
    properties: dict[str, Any]
    created_at: datetime

    @field_serializer("created_at")
    def serialize_datetime(self, v: datetime, _info) -> str:
        """序列化为 ISO 格式字符串"""
        return v.isoformat()

    @field_serializer("properties")
    def serialize_properties(self, v: dict, _info) -> dict:
        """移除 None 值"""
        return {k: v for k, v in v.items() if v is not None}

# model_dump 控制
instance = OntologyInstance(id="1", properties={"a": 1, "b": None}, created_at=datetime.now())

# 排除 None 值
instance.model_dump(exclude_none=True)

# 只包含特定字段
instance.model_dump(include={"id", "properties"})

# 使用别名
instance.model_dump(by_alias=True)

# JSON 序列化
instance.model_dump_json(indent=2)

from typing import Literal, Union, Annotated
from pydantic import Discriminator, Tag

class CreateAction(BaseModel):
    type: Literal["CREATE"] = "CREATE"
    object_type: str
    properties: dict[str, Any]

class UpdateAction(BaseModel):
    type: Literal["UPDATE"] = "UPDATE"
    object_id: str
    changes: dict[str, Any]

class DeleteAction(BaseModel):
    type: Literal["DELETE"] = "DELETE"
    object_id: str
    soft_delete: bool = True

# Discriminated Union（按 type 字段区分）
ActionPayload = Annotated[
    Union[CreateAction, UpdateAction, DeleteAction],
    Field(discriminator="type"),
]

class ActionRequest(BaseModel):
    world_id: str
    action: ActionPayload

# 自动匹配正确的类型
req = ActionRequest.model_validate({
    "world_id": "w1",
    "action": {"type": "CREATE", "object_type": "Employee", "properties": {"name": "Alice"}},
})
assert isinstance(req.action, CreateAction)

def get_property_type(v: Any) -> str:
    if isinstance(v, dict):
        return v.get("data_type", "unknown")
    return "unknown"

StringProperty = Annotated[StringPropertyModel, Tag("STRING")]
IntegerProperty = Annotated[IntegerPropertyModel, Tag("INTEGER")]
ArrayProperty = Annotated[ArrayPropertyModel, Tag("ARRAY")]

DynamicProperty = Annotated[
    Union[StringProperty, IntegerProperty, ArrayProperty],
    Discriminator(get_property_type),
]

from pydantic import create_model

def create_instance_model(schema: ObjectTypeSchema) -> type[BaseModel]:
    """根据 ObjectType Schema 动态生成 Pydantic 模型"""
    field_definitions = {}

    type_mapping = {
        "STRING": (str, ...),
        "INTEGER": (int, ...),
        "FLOAT": (float, ...),
        "BOOLEAN": (bool, ...),
        "DATETIME": (datetime, ...),
        "ARRAY": (list[Any], ...),
        "MAP": (dict[str, Any], ...),
    }

    for prop in schema.properties:
        python_type, default = type_mapping.get(
            prop.data_type, (Any, ...)
        )

        if not prop.required:
            python_type = python_type | None
            default = prop.default_value

        field_info = Field(
            default=default,
            description=prop.description,
            **prop.constraints,
        )
        field_definitions[prop.name] = (python_type, field_info)

    return create_model(
        f"{schema.name}Instance",
        __base__=OntologyInstanceBase,
        **field_definitions,
    )

# 使用
employee_schema = ObjectTypeSchema(
    name="Employee",
    properties=[
        PropertyDefinition(name="name", data_type="STRING", required=True),
        PropertyDefinition(name="age", data_type="INTEGER", constraints={"ge": 0, "le": 150}),
        PropertyDefinition(name="department", data_type="STRING"),
    ],
)

EmployeeModel = create_instance_model(employee_schema)
emp = EmployeeModel(
    id="emp-001", object_type="Employee", world_id="w1",
    name="Alice", age=30, department="Engineering",
)

from functools import lru_cache

class DynamicModelRegistry:
    """动态模型注册表（缓存已编译的模型）"""

    def __init__(self):
        self._models: dict[str, type[BaseModel]] = {}

    def get_or_create(self, schema: ObjectTypeSchema) -> type[BaseModel]:
        cache_key = f"{schema.name}_{hash(frozenset(
            (p.name, p.data_type) for p in schema.properties
        ))}"

        if cache_key not in self._models:
            self._models[cache_key] = create_instance_model(schema)

        return self._models[cache_key]

    def invalidate(self, object_type: str) -> None:
        keys_to_remove = [k for k in self._models if k.startswith(f"{object_type}_")]
        for key in keys_to_remove:
            del self._models[key]

model_registry = DynamicModelRegistry()

class PropertyConstraint(BaseModel):
    """属性约束定义"""
    min_length: int | None = None
    max_length: int | None = None
    pattern: str | None = None
    ge: float | None = None
    le: float | None = None
    enum_values: list[Any] | None = None
    unique: bool = False

def build_property_validator(
    name: str, data_type: str, constraint: PropertyConstraint
) -> Callable:
    """构建属性验证函数"""
    def validator(v: Any) -> Any:
        if constraint.min_length and isinstance(v, str) and len(v) < constraint.min_length:
            raise ValueError(f"{name}: min length is {constraint.min_length}")
        if constraint.max_length and isinstance(v, str) and len(v) > constraint.max_length:
            raise ValueError(f"{name}: max length is {constraint.max_length}")
        if constraint.pattern and isinstance(v, str):
            import re
            if not re.match(constraint.pattern, v):
                raise ValueError(f"{name}: does not match pattern {constraint.pattern}")
        if constraint.ge is not None and isinstance(v, (int, float)) and v < constraint.ge:
            raise ValueError(f"{name}: must be >= {constraint.ge}")
        if constraint.le is not None and isinstance(v, (int, float)) and v > constraint.le:
            raise ValueError(f"{name}: must be <= {constraint.le}")
        if constraint.enum_values and v not in constraint.enum_values:
            raise ValueError(f"{name}: must be one of {constraint.enum_values}")
        return v
    return validator

class CrossPropertyValidator:
    """跨属性验证规则"""

    @staticmethod
    def validate_date_range(instance: dict, start_field: str, end_field: str) -> None:
        start = instance.get(start_field)
        end = instance.get(end_field)
        if start and end and start > end:
            raise ValueError(f"{start_field} must be before {end_field}")

    @staticmethod
    def validate_conditional_required(
        instance: dict, condition_field: str, condition_value: Any,
        required_field: str,
    ) -> None:
        if instance.get(condition_field) == condition_value:
            if not instance.get(required_field):
                raise ValueError(
                    f"{required_field} is required when {condition_field}={condition_value}"
                )

# Pydantic v2 自动生成 JSON Schema
schema = OntologyInstanceBase.model_json_schema()

# 输出：
# {
#   "title": "OntologyInstanceBase",
#   "type": "object",
#   "properties": {
#     "id": {"type": "string", "minLength": 1, "maxLength": 255, "pattern": "^[a-zA-Z0-9_-]+$"},
#     "objectType": {"type": "string", "minLength": 1, "maxLength": 100},
#     ...
#   },
#   "required": ["id", "objectType", "worldId"],
#   "examples": [...]
# }

from pydantic import TypeAdapter
from typing import TypedDict

class InstanceDict(TypedDict):
    id: str
    object_type: str
    properties: dict[str, Any]

adapter = TypeAdapter(InstanceDict)
validated = adapter.validate_python({"id": "1", "object_type": "Emp", "properties": {}})
json_schema = adapter.json_schema()

from fastapi import FastAPI, Query, Body

app = FastAPI()

class QueryRequest(BaseModel):
    oql: str = Field(..., description="OQL 查询语句")
    world_id: str
    limit: int = Field(default=100, ge=1, le=10000)
    offset: int = Field(default=0, ge=0)

class QueryResponse(BaseModel):
    instances: list[OntologyInstanceBase]
    total_count: int
    has_more: bool
    execution_time_ms: float

@app.post("/api/v1/query", response_model=QueryResponse)
async def execute_query(request: QueryRequest = Body(...)):
    # FastAPI 自动使用 Pydantic v2 验证和序列化
    result = await query_service.execute(request)
    return QueryResponse(
        instances=result.instances,
        total_count=result.total,
        has_more=result.offset + result.limit < result.total,
        execution_time_ms=result.duration_ms,
    )

from fastapi.responses import ORJSONResponse

app = FastAPI(default_response_class=ORJSONResponse)

# ORJSONResponse 使用 orjson 库，比标准 json 快 3-10 倍
# Pydantic v2 + orjson = 极致序列化性能

# 测试：100,000 次模型实例化
import timeit

class SimpleModel(BaseModel):
    id: str
    name: str
    value: int

data = {"id": "1", "name": "test", "value": 42}

# Pydantic v2
time_v2 = timeit.timeit(lambda: SimpleModel(**data), number=100_000)
# → 0.35 秒（3.5 us/次）

# 使用 model_validate（更快，跳过 Python __init__）
time_v2_validate = timeit.timeit(
    lambda: SimpleModel.model_validate(data), number=100_000
)
# → 0.28 秒（2.8 us/次）

# JSON 解析
json_str = '{"id": "1", "name": "test", "value": 42}'
time_v2_json = timeit.timeit(
    lambda: SimpleModel.model_validate_json(json_str), number=100_000
)
# → 0.22 秒（2.2 us/次 — 直接 Rust 解析 JSON）

# 1. 使用 model_validate_json 直接解析 JSON（最快路径）
instance = MyModel.model_validate_json(json_bytes)

# 2. 禁用不需要的验证
class FastModel(BaseModel):
    model_config = ConfigDict(
        validate_assignment=False,  # 不验证赋值
        revalidate_instances="never",  # 不重新验证嵌套模型
    )

# 3. 使用 TypeAdapter 批量验证
adapter = TypeAdapter(list[OntologyInstanceBase])
instances = adapter.validate_python(raw_list)  # 批量验证

# 4. frozen 模型（不可变 → 可哈希）
class ImmutableModel(BaseModel):
    model_config = ConfigDict(frozen=True)
    id: str
    name: str