“

Series: S8 Technology Deep Dives · Article 17 | Level: Advanced | Reading Time: 20 min

Pydantic v2 Deep Dive: Data Validation and Model Layer Design in coomia-dip

#TL;DR

Pydantic v2 is the data validation and serialization core for coomia-dip Python services (Reasoning & Decision Layer + Agent Runtime Layer + SDK & Developer Experience Layer), with the Rust-rewritten pydantic-core delivering 5-50x performance improvements
This article analyzes Pydantic v2's CoreSchema compilation model, Validator/Serializer dual-track system, Discriminated Union advanced usage, and coomia-dip's dynamic Ontology model validation approach
Covers Model Config best practices, custom types, JSON Schema generation, FastAPI/gRPC integration, and key v1 to v2 migration changes

#1. Pydantic v2 Architecture Revolution

#1.1 v1 vs v2 Architecture

Code

Pydantic v1:
  Python layer: Field → Validator → Model
  All Python → Performance limited

Pydantic v2:
  Python layer: Field → Model definition
       ↓ compile
  Rust layer (pydantic-core):
    CoreSchema → Compiled Validator → High-speed validation
               → Compiled Serializer → High-speed serialization

#1.2 Performance Gains

Operation	v1	v2	Improvement
Model instantiation (simple)	4.2 us	0.3 us	14x
Model instantiation (complex)	85 us	3.5 us	24x
JSON parsing	12 us	0.8 us	15x
JSON serialization	8.5 us	0.5 us	17x
Validation failure	15 us	1.2 us	12x

#2. CoreSchema Compilation Model

#2.1 Compilation Flow

Python

from pydantic import BaseModel

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

# Pydantic v2 internal compilation:
# 1. Analyze type annotations → Generate CoreSchema (JSON description)
# 2. CoreSchema → Rust Validator (compile time)
# 3. Runtime calls Validator for validation (Rust speed)

print(OntologyInstance.__pydantic_core_schema__)

#2.2 Strict vs Lax Mode

Python

from pydantic import BaseModel, ConfigDict

class StrictModel(BaseModel):
    model_config = ConfigDict(strict=True)
    count: int
    name: str

StrictModel(count=42, name="test")     # ✅
StrictModel(count="42", name="test")   # ❌ ValidationError

class LaxModel(BaseModel):
    count: int
    name: str

LaxModel(count="42", name="test")      # ✅ "42" → 42 auto-coercion

#3. Model Definition Best Practices

#3.1 coomia-dip Core Models

Python

from pydantic import BaseModel, Field, ConfigDict, field_validator, model_validator

class OntologyInstanceBase(BaseModel):
    model_config = ConfigDict(
        from_attributes=True,
        populate_by_name=True,
        str_strip_whitespace=True,
        validate_default=True,
        use_enum_values=True,
    )

    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:
        if not v[0].isupper():
            raise ValueError(f"ObjectType must be PascalCase: {v}")
        return v

    @model_validator(mode="after")
    def validate_timestamps(self) -> "OntologyInstanceBase":
        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

#3.2 Nested Models

Python

class PropertyDefinition(BaseModel):
    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}
        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):
    name: str
    display_name: str = ""
    properties: list[PropertyDefinition] = Field(default_factory=list)
    primary_key: list[str] = Field(default_factory=lambda: ["id"])

#4. Validators and Serializers

#4.1 Field Validator

Python

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

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

#4.2 Serializer

Python

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

    @field_serializer("created_at")
    def serialize_datetime(self, v: datetime, _info) -> str:
        return v.isoformat()

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

instance.model_dump(exclude_none=True)
instance.model_dump(include={"id", "properties"})
instance.model_dump(by_alias=True)
instance.model_dump_json(indent=2)

#5. Discriminated Unions

Python

from typing import Literal, Union, Annotated
from pydantic import Field

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

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)

#6. Dynamic Model Generation

Python

from pydantic import create_model

def create_instance_model(schema: ObjectTypeSchema) -> type[BaseModel]:
    field_definitions = {}
    type_mapping = {
        "STRING": (str, ...), "INTEGER": (int, ...), "FLOAT": (float, ...),
        "BOOLEAN": (bool, ...), "DATETIME": (datetime, ...),
    }

    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_definitions[prop.name] = (python_type, Field(default=default))

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

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]

#7. Ontology Property Validation

Python

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

def build_property_validator(name: 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.pattern and isinstance(v, str):
            if not re.match(constraint.pattern, v):
                raise ValueError(f"{name}: does not match 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.enum_values and v not in constraint.enum_values:
            raise ValueError(f"{name}: must be one of {constraint.enum_values}")
        return v
    return validator

#8. JSON Schema Generation

Python

schema = OntologyInstanceBase.model_json_schema()
# Automatic OpenAPI-compatible JSON Schema

from pydantic import TypeAdapter
adapter = TypeAdapter(list[OntologyInstanceBase])
json_schema = adapter.json_schema()

#9. FastAPI Integration

Python

@app.post("/api/v1/query", response_model=QueryResponse)
async def execute_query(request: QueryRequest = Body(...)):
    # FastAPI automatically uses Pydantic v2 for validation and serialization
    result = await query_service.execute(request)
    return QueryResponse(instances=result.instances, total_count=result.total)

# Use ORJSONResponse for 3-10x faster JSON serialization
from fastapi.responses import ORJSONResponse
app = FastAPI(default_response_class=ORJSONResponse)

#10. Performance Benchmarks and Optimization

Python

# Fastest path: model_validate_json (Rust parses JSON directly)
instance = MyModel.model_validate_json(json_bytes)  # 2.2 us

# Batch validation with TypeAdapter
adapter = TypeAdapter(list[OntologyInstanceBase])
instances = adapter.validate_python(raw_list)

# Frozen models (immutable → hashable)
class ImmutableModel(BaseModel):
    model_config = ConfigDict(frozen=True)
    id: str
    name: str

#11. Key Takeaways

Topic	Key Conclusion
Performance	v2 is 5-50x faster than v1 (Rust core)
CoreSchema	Compile-time generation, Rust runtime validation
Strict mode	Recommended for production, prevents implicit coercion
Dynamic models	create_model() generates models from Schema dynamically
Discriminated Union	Auto-routes to correct subtype by tag field
Serializer	model_dump / model_dump_json replace dict/json
JSON Schema	Auto-generated, directly usable for OpenAPI/Swagger
Optimization	model_validate_json is the fastest path

“

Next up: S8-18 dives into Google OR-Tools, exploring how coomia-dip uses constraint solvers for intelligent decision optimization.