pydantic-data-models

name	pydantic-data-models
description	Expert guidance for using Pydantic V2 for data validation, serialization, and type-safe Python. Use when defining data models, validating API inputs/outputs, parsing JSON, handling LLM structured outputs, or integrating with Temporal workflows. Covers BaseModel, dataclasses, validators, and serialization.

Pydantic Data Models Skill

Overview

Pydantic provides data validation and settings management using Python type annotations. It's essential for type-safe applications, API input/output validation, and structured data handling with LLMs.

When to Use This Skill

• Defining data models for workflow inputs/outputs
• Validating API responses and LLM outputs
• Parsing and serializing JSON data
• Creating type-safe dataclasses
• Building configuration management
• Integrating with Temporal, Ray, or FastAPI

Basic Models

Defining a Model

from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class ContractEntity(BaseModel):
    """Entity extracted from a contract document."""
    
    contract_parties: list[str]
    effective_date: Optional[str] = None
    expiration_date: Optional[str] = None
    payment_terms: dict[str, str]
    key_risks: list[str]
    confidence_score: float
    extraction_model: str

# Create instance
entity = ContractEntity(
    contract_parties=["Company A", "Company B"],
    effective_date="2025-01-15",
    expiration_date="2026-01-15",
    payment_terms={"amount": "$50,000", "frequency": "annual"},
    key_risks=["Non-compete clause"],
    confidence_score=0.92,
    extraction_model="gpt-5.2"
)

# Access fields
print(entity.contract_parties)  # ['Company A', 'Company B']
print(entity.confidence_score)  # 0.92

Model Methods

# Serialize to dictionary
data = entity.model_dump()
# {'contract_parties': ['Company A', 'Company B'], 'effective_date': '2025-01-15', ...}

# Serialize to JSON string
json_str = entity.model_dump_json(indent=2)

# Get JSON schema
schema = ContractEntity.model_json_schema()

# Create copy with updates
updated = entity.model_copy(update={"confidence_score": 0.95})

Validation

From Dictionary

# Validate dictionary data
data = {
    "contract_parties": ["Acme Corp", "Widget Inc"],
    "payment_terms": {"amount": "100K"},
    "key_risks": [],
    "confidence_score": "0.85",  # Will be coerced to float
    "extraction_model": "gpt-5.1"
}

entity = ContractEntity.model_validate(data)
print(entity.confidence_score)  # 0.85 (as float)

From JSON

json_data = '{"contract_parties": ["A", "B"], "payment_terms": {}, "key_risks": [], "confidence_score": 0.9, "extraction_model": "gpt-5.2"}'

entity = ContractEntity.model_validate_json(json_data)

Validation Errors

from pydantic import ValidationError

try:
    entity = ContractEntity(
        contract_parties="not a list",  # Wrong type
        payment_terms={},
        key_risks=[],
        confidence_score="invalid",  # Can't parse as float
        extraction_model="gpt-5.2"
    )
except ValidationError as e:
    print(e.errors())
    # [{'type': 'list_type', 'loc': ('contract_parties',), ...}, ...]

Field Configuration

Using Field()

from pydantic import BaseModel, Field
from typing import Annotated

class ProcessingResult(BaseModel):
    """Result from document processing."""
    
    document_path: str = Field(description="Path to the processed document")
    chunks: list[str] = Field(default_factory=list)
    processing_time_ms: int = Field(ge=0, description="Processing time in milliseconds")
    retry_attempts: int = Field(default=0, ge=0, le=10)
    confidence: float = Field(ge=0.0, le=1.0)
    
    # Using Annotated pattern (recommended)
    error_log: Annotated[list[str], Field(default_factory=list)]

# Validation enforces constraints
result = ProcessingResult(
    document_path="/docs/contract.pdf",
    processing_time_ms=1500,
    confidence=0.95
)

Field Aliases

class APIResponse(BaseModel):
    """Model with field aliases for API compatibility."""
    
    internal_id: str = Field(alias="id")
    created_at: datetime = Field(alias="createdAt")
    
    model_config = {"populate_by_name": True}

# Can use either name
resp = APIResponse(id="123", createdAt="2025-01-15T10:00:00")
resp = APIResponse(internal_id="123", created_at="2025-01-15T10:00:00")

Nested Models

class DocumentChunk(BaseModel):
    """A chunk of a document."""
    content: str
    start_page: int
    end_page: int
    embedding: list[float] | None = None

class ExtractedEntity(BaseModel):
    """An entity extracted from a chunk."""
    entity_type: str
    value: str
    confidence: float

class ProcessedDocument(BaseModel):
    """Complete processed document."""
    path: str
    chunks: list[DocumentChunk]
    entities: list[ExtractedEntity]
    metadata: dict[str, str] = Field(default_factory=dict)

# Nested validation works automatically
doc = ProcessedDocument(
    path="/docs/contract.pdf",
    chunks=[
        {"content": "...", "start_page": 1, "end_page": 2},
        {"content": "...", "start_page": 3, "end_page": 4},
    ],
    entities=[
        {"entity_type": "date", "value": "2025-01-15", "confidence": 0.95},
    ]
)

Custom Validators

Field Validators

from pydantic import BaseModel, field_validator

class ContractInput(BaseModel):
    document_path: str
    extraction_model: str
    
    @field_validator("document_path")
    @classmethod
    def validate_path(cls, v: str) -> str:
        if not v.endswith((".pdf", ".docx", ".txt")):
            raise ValueError("Must be a PDF, DOCX, or TXT file")
        return v
    
    @field_validator("extraction_model")
    @classmethod
    def validate_model(cls, v: str) -> str:
        allowed = {"gpt-5.2", "gpt-5.1", "gpt-4o"}
        if v not in allowed:
            raise ValueError(f"Model must be one of {allowed}")
        return v

Model Validators

from pydantic import BaseModel, model_validator

class DateRange(BaseModel):
    start_date: str
    end_date: str
    
    @model_validator(mode="after")
    def validate_date_range(self) -> "DateRange":
        if self.end_date < self.start_date:
            raise ValueError("end_date must be after start_date")
        return self

Temporal Integration

Workflow Input/Output Models

from dataclasses import dataclass
from pydantic import BaseModel

# For Temporal, use dataclasses (recommended) or Pydantic models

@dataclass
class PipelineInput:
    """Input for the document processing workflow."""
    document_paths: list[str]
    extraction_model: str = "gpt-5.2"
    fallback_model: str = "gpt-5.1"
    max_retries: int = 3

@dataclass
class PipelineOutput:
    """Output from the document processing workflow."""
    entities: list[dict]
    total_chunks: int
    processing_time_ms: int
    models_used: list[str]
    errors: list[str]

# Or use Pydantic models (also works with Temporal)
class ProcessingConfig(BaseModel):
    """Configuration for processing."""
    model_config = {"frozen": True}  # Immutable
    
    chunk_size: int = 1000
    overlap: int = 100
    embedding_model: str = "text-embedding-3-small"  # From AZURE_OPENAI_TEXTEMBEDDING_DEPLOYMENT_NAME

Activity Data Models

from pydantic import BaseModel
from typing import Optional

class ExtractionResult(BaseModel):
    """Result from entity extraction activity."""
    entities: dict
    model_used: str
    tokens_used: int
    latency_ms: int
    fallback_used: bool = False
    error: Optional[str] = None
    
    def is_successful(self) -> bool:
        return self.error is None

LLM Structured Output

Parsing LLM JSON Responses

import json
from pydantic import BaseModel, ValidationError

class LLMExtractionOutput(BaseModel):
    """Expected structure from LLM extraction."""
    parties: list[str]
    dates: dict[str, str]
    amounts: list[str]
    risks: list[str]

async def parse_llm_response(raw_response: str) -> LLMExtractionOutput:
    """Parse and validate LLM JSON response."""
    try:
        # Parse JSON
        data = json.loads(raw_response)
        
        # Validate with Pydantic
        return LLMExtractionOutput.model_validate(data)
        
    except json.JSONDecodeError as e:
        raise ValueError(f"Invalid JSON from LLM: {e}")
    except ValidationError as e:
        raise ValueError(f"LLM output doesn't match schema: {e}")

Schema for LLM Prompts

def get_extraction_prompt(text: str) -> str:
    """Generate prompt with schema for LLM."""
    
    # Get Pydantic model's JSON schema
    schema = LLMExtractionOutput.model_json_schema()
    
    return f"""Extract entities from the following text and return as JSON.

Your response MUST match this JSON schema:
{json.dumps(schema, indent=2)}

Text:
{text}

JSON Output:"""

Serialization Options

class DetailedResult(BaseModel):
    id: str
    data: dict
    internal_score: float = Field(exclude=True)  # Exclude from serialization
    created_at: datetime

result = DetailedResult(
    id="123",
    data={"key": "value"},
    internal_score=0.95,
    created_at=datetime.now()
)

# Default serialization
print(result.model_dump())
# {'id': '123', 'data': {'key': 'value'}, 'created_at': datetime(...)}

# Exclude fields
print(result.model_dump(exclude={"data"}))
# {'id': '123', 'created_at': datetime(...)}

# Include only specific fields
print(result.model_dump(include={"id", "created_at"}))

# JSON mode (dates become strings)
print(result.model_dump(mode="json"))
# {'id': '123', 'data': {'key': 'value'}, 'created_at': '2025-01-15T...'}

Configuration

Model Config

from pydantic import BaseModel, ConfigDict

class StrictModel(BaseModel):
    """Model with strict configuration."""
    
    model_config = ConfigDict(
        strict=True,           # No type coercion
        frozen=True,           # Immutable instances
        extra="forbid",        # No extra fields allowed
        str_strip_whitespace=True,  # Strip strings
        validate_default=True, # Validate default values
    )
    
    name: str
    value: int

class FlexibleModel(BaseModel):
    """Model with flexible configuration."""
    
    model_config = ConfigDict(
        extra="allow",         # Allow extra fields
        from_attributes=True,  # Enable ORM mode
        populate_by_name=True, # Allow field names or aliases
    )
    
    id: int
    name: str

Generic Models

from typing import Generic, TypeVar
from pydantic import BaseModel

T = TypeVar("T")

class APIResponse(BaseModel, Generic[T]):
    """Generic API response wrapper."""
    success: bool
    data: T
    error: str | None = None

class User(BaseModel):
    id: int
    name: str

# Use with specific types
user_response: APIResponse[User] = APIResponse(
    success=True,
    data={"id": 1, "name": "Alice"}
)

list_response: APIResponse[list[User]] = APIResponse(
    success=True,
    data=[{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
)

Best Practices

1. Use dataclasses for Temporal - Simpler serialization
2. Prefer Annotated pattern - Annotated[int, Field(ge=0)]
3. Define clear schemas for LLM outputs - Helps with parsing
4. Use frozen models for immutability - Thread-safe
5. Validate early - At API/workflow boundaries
6. Include descriptions - Self-documenting schemas

Installation

pip install pydantic>=2.10.0

References

• Pydantic Documentation
• BaseModel API
• Field Configuration
• Validators
• JSON Schema