Build Pydantic models dynamically from YAML schema — extends, cross-references, field descriptions, two-pass resolution. Production learnings from resource-catalog CLI.

schema-driven-models.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "pydantic>=2.0",
#   "pyyaml>=6.0",
# ]
# ///
"""
schema-driven-models.py

Build Pydantic models dynamically from YAML schema definitions.
Supports cross-references, inheritance (extends), field descriptions,
and nested generics — all driven by data, not code.

Usage:
    ./schema-driven-models.py                           # Run demo
    ./schema-driven-models.py my-schema.yaml            # Load from file

Learnings from production use (resource-catalog CLI):
  - extends chains fields from a parent entity
  - Two-pass loading resolves forward references
  - default_factory=list avoids shared mutable defaults
  - Entities can be defined as top-level presets or nested
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

import yaml
from pydantic import BaseModel, Field, create_model


# ── Type resolution ──────────────────────────────────────────────────────────

TYPE_MAP: dict[str, type] = {
    "str": str,
    "int": int,
    "float": float,
    "bool": bool,
    "Path": Path,
    "Any": Any,
}

# Registry enables cross-model references (e.g. Address referenced by Person)
_model_registry: dict[str, type[BaseModel]] = {}

# Presets registry for entity inheritance (extends)
_presets: dict[str, dict] = {}


def _resolve_type(spec: str) -> type:
    """Convert a YAML type string to a Python type."""
    # Check model registry first (enables cross-references)
    model_type = _model_registry.get(spec)
    if model_type is not None:
        return model_type

    if isinstance(spec, str):
        if " | None" in spec:
            base = _resolve_type(spec.replace(" | None", ""))
            return base | None  # type: ignore
        if spec.startswith("list["):
            inner = _resolve_type(spec[5:-1])
            return list[inner]  # type: ignore
        if spec.startswith("dict["):
            ks, vs = spec[5:-1].split(", ", 1)
            return dict[_resolve_type(ks), _resolve_type(vs)]  # type: ignore
        if spec in TYPE_MAP:
            return TYPE_MAP[spec]

        # Fallback: check if spec is a known entity name
        try:
            ref = _build_entity(spec, {})
            return ref
        except ValueError:
            pass

        msg = f"Unknown type: {spec}"
        raise ValueError(msg)

    msg = f"Unsupported spec: {spec}"
    raise ValueError(msg)


# ── Entity building with extends support ─────────────────────────────────────


def _resolve_entity_spec(name: str) -> dict:
    """Find an entity definition: presets → schema models."""
    if name in _presets:
        return dict(_presets[name])
    raise ValueError(f"Unknown entity: {name}")


def _build_entity(name: str, force: bool = False) -> type[BaseModel]:
    """Build a Pydantic model from an entity definition, resolving extends chains.

    Key patterns from production use:
      - ``extends`` chains fields from a parent entity
      - Pass ``force=True`` to rebuild over a placeholder model
      - ``default_factory=list`` for list fields (avoids shared mutable defaults)
    """
    cache_key = name
    if not force and cache_key in _model_registry:
        existing = _model_registry[cache_key]
        if existing.model_fields:
            return existing

    spec = _resolve_entity_spec(name)

    # Collect all fields: parent (extends) first, then own
    all_fields: dict[str, dict] = {}
    if "extends" in spec:
        parent = _resolve_entity_spec(spec["extends"])
        for fname, fspec in parent.get("fields", {}).items():
            all_fields[fname] = dict(fspec) if isinstance(fspec, dict) else {"type": fspec}

    for fname, fspec in spec.get("fields", {}).items():
        all_fields[fname] = dict(fspec) if isinstance(fspec, dict) else {"type": fspec}

    # Build create_model kwargs
    kwargs: dict[str, tuple[type, Any]] = {}
    for fname, fspec in all_fields.items():
        py_type = _resolve_type(fspec.get("type", "str"))
        default = fspec.get("default")
        desc = fspec.get("description", "")

        if default is not None and isinstance(default, list):
            kwargs[fname] = (py_type, Field(default_factory=lambda: list(default), description=desc))
        elif default is not None:
            kwargs[fname] = (py_type, Field(default=default, description=desc))
        elif fspec.get("required"):
            kwargs[fname] = (py_type, Field(description=desc))
        else:
            kwargs[fname] = (py_type, Field(default=None, description=desc))

    model = create_model(name, **kwargs)  # type: ignore
    _model_registry[cache_key] = model
    return model


# ── Schema loading (two-pass) ────────────────────────────────────────────────


def load_schema(path: Path) -> dict[str, type[BaseModel]]:
    """Load a YAML schema file with two-pass model building.

    Pass 1 registers placeholder models so forward references resolve.
    Pass 2 rebuilds each model with actual fields.
    """
    raw = yaml.safe_load(path.read_text(encoding="utf-8"))

    # Top-level presets can be referenced via extends
    for name, spec in raw.get("presets", {}).items():
        _presets[name] = spec

    # Register models defined directly in the schema
    models: dict[str, type[BaseModel]] = {}

    # Pass 1: register placeholders for all models
    for name in raw.get("entities", {}):
        _model_registry[name] = create_model(name)  # type: ignore

    # Pass 2: rebuild with actual fields
    for name, spec in raw.get("entities", {}).items():
        _presets[name] = spec  # Also add to presets (for extends)

    for name in raw.get("entities", {}):
        models[name] = _build_entity(name, force=True)

    return models


# ── Demo ─────────────────────────────────────────────────────────────────────

DEMO_SCHEMA = """
presets:
  base-entity:
    description: "Base fields shared by all entities"
    fields:
      id:
        type: str
        required: true
        description: "Unique identifier"
      description:
        type: str
        required: true
        description: "Description"

entities:
  Address:
    fields:
      street:
        type: str
        required: true
        description: "Street name"
      city:
        type: str
        default: "Vienna"
        description: "City"
      zip:
        type: str
        description: "Postal code"
      country:
        type: str
        default: "Austria"
        description: "Country"

  Person:
    extends: base-entity
    fields:
      name:
        type: str
        required: true
        description: "Full name"
      age:
        type: int
        default: 0
        description: "Age in years"
      email:
        type: str | None
        description: "Email address"
      address:
        type: Address
        description: "Home address"
      tags:
        type: list[str]
        default: []
        description: "Labels"
"""


def main() -> None:
    schema_path = Path(sys.argv[1]) if len(sys.argv) > 1 else None

    if schema_path:
        models = load_schema(schema_path)
    else:
        tmp = Path("/tmp/_schema_demo.yaml")
        tmp.write_text(DEMO_SCHEMA, encoding="utf-8")
        models = load_schema(tmp)
        tmp.unlink()

    print("=== Dynamically created models ===\n")
    for name, model in models.items():
        print(f"  {name}: ({len(model.model_fields)} fields)")
        for fname, field in model.model_fields.items():
            ann = field.annotation
            desc = field.description or ""
            default = f" = {field.default!r}" if field.default is not None else ""
            print(f"    {fname}: {ann}  {desc}{default}")
        print()

    if "Person" in models and "Address" in models:
        Person = models["Person"]
        Address = models["Address"]

        addr = Address(street="123 Main St", zip="1010")
        print("  Address instance:", addr.model_dump())

        person = Person(id="P01", description="Alice's entry", name="Alice", age=30, address=addr, tags=["admin", "dev"])
        print("  Person instance:", person.model_dump())

        person2 = Person(id="P02", description="Bob's entry", name="Bob", tags=["dev"])
        print("  Person with defaults:", person2.model_dump())

        # Validation works
        try:
            Person(id="P03", name="Charlie", age="not a number")
        except Exception as e:
            print(f"  Validation error: {e}")

    print("\n  Key patterns demonstrated:")
    print("    - extends: Person inherits id, description from base-entity")
    print("    - Cross-ref: Person.address type is Address model")
    print("    - Optional: email is str | None")
    print("    - default_factory: tags=list avoids shared mutable default")
    print("    - Two-pass: forward references resolve via model registry")


if __name__ == "__main__":
    main()