Pydantic v1→v2 Migration: Experience Upgrading 28 Models in a Test Project

# Pydantic v1
class JWTPayload(BaseModel):
    user_id: int
    email: str
    roles: List[str]
    exp: int
    iat: int
 
# Metrics before migration
Latency (P95): 12ms for token validation
CPU: 18% utilization for validation
Throughput: ~2000 req/sec before CPU throttling

orders = [Order(**row) for row in csv_reader]  # Peak consumption: 1.2GB

# 1000 users with profiles and permissions
users = [UserResponse.from_orm(u) for u in db_users]
serialized = [u.dict() for u in users]  # ~450ms

class Order(BaseModel):
    product_id: int
    quantity: int
    price: Decimal
 
# Regular creation (with validation)
order = Order(product_id="123", quantity="5", price="99.99")
# 1. Type validation: "123" -> int, "5" -> int
# 2. Constraint validation (if any)
# 3. Object creation
# Time: ~10-50 μs per object (depends on model complexity)

# Creation without validation
order = Order.model_construct(product_id=123, quantity=5, price=Decimal("99.99"))
# 1. Just creates object with passed values
# 2. NO type checking
# 3. NO validators
# Time: ~1-2 μs per object (5-25x faster)

# CSV with 50,000 orders from internal system
# Format guaranteed (generated by our own code)
 
# ❌ Slow: validate every row
orders = [Order(**row) for row in csv_rows]  # ~2.5 seconds + 1.2GB RAM
 
# ✅ Fast: validate only file structure
def validate_csv_structure(headers):
    expected = set(Order.model_fields.keys())
    if not expected.issubset(headers):
        raise ValueError(f"Invalid CSV: missing {expected - set(headers)}")
 
validate_csv_structure(csv_headers)  # Once per file
orders = [Order.model_construct(**row) for row in csv_rows]  # ~0.3 seconds + 380MB RAM

# Script to find all BaseModel (rg = ripgrep, fast grep alternative)
rg "class \w+\(BaseModel\)" --type py -c
 
# Result:
API Gateway: 8 models
Order Service: 12 models
User Service: 6 models
Shared libs: 2 models
Total: 28 models

# check_dependencies.py
import ast
import sys
 
class ValidatorVisitor(ast.NodeVisitor):
    """Traverses Python file AST and collects Pydantic v1 usage statistics."""
 
    def __init__(self):
        self.validators = []        # List of models with @validator
        self.config_classes = []    # List of models with nested class Config
 
    def visit_FunctionDef(self, node):
        """Called for every function/method in code."""
        # Check method decorators
        for decorator in node.decorator_list:
            # If found @validator — it's v1 API, v2 needs @field_validator
            if isinstance(decorator, ast.Name) and decorator.id == 'validator':
                self.validators.append(node.name)
 
    def visit_ClassDef(self, node):
        """Called for every class in code."""
        # Check class body for nested class Config
        for item in node.body:
            if isinstance(item, ast.ClassDef) and item.name == 'Config':
                # class Config -> model_config = ConfigDict() in v2
                self.config_classes.append(node.name)
        self.generic_visit(node)  # Continue traversing nested nodes
 
# Run on all project Python files, collect statistics
# Analysis results:
# 18 models with @validator          ← need to change to @field_validator
# 14 models with Config class        ← need to change to model_config = ConfigDict()
# 8 models with orm_mode = True      ← need to change to from_attributes = True
# 3 models with JSON encoders        ← need to rewrite to @field_serializer

# migrate_pydantic.py
import libcst as cst
from libcst import matchers as m
 
class PydanticV2Transformer(cst.CSTTransformer):
    """Automatic Pydantic v1 -> v2 migration."""
 
    def leave_FunctionDef(self, original_node, updated_node):
        # @validator -> @field_validator
        if self._has_validator_decorator(updated_node):
            return self._convert_validator(updated_node)
        return updated_node
 
    def leave_ClassDef(self, original_node, updated_node):
        # Config -> model_config = ConfigDict(...)
        if self._has_config_class(updated_node):
            return self._convert_config(updated_node)
        return updated_node
 
    def _convert_validator(self, node):
        """@validator('field') -> @field_validator('field')."""
        new_decorators = []
        for decorator in node.decorators:
            if m.matches(decorator, m.Decorator(decorator=m.Name("validator"))):
                # Replace decorator name
                new_dec = decorator.with_changes(
                    decorator=cst.Name("field_validator")
                )
                new_decorators.append(new_dec)
            else:
                new_decorators.append(decorator)
 
        # Add @classmethod if missing
        has_classmethod = any(
            m.matches(d, m.Decorator(decorator=m.Name("classmethod")))
            for d in new_decorators
        )
        if not has_classmethod:
            new_decorators.insert(0, cst.Decorator(decorator=cst.Name("classmethod")))
 
        return node.with_changes(decorators=new_decorators)
 
    def _convert_config(self, node):
        """class Config -> model_config = ConfigDict(...)."""
        # Find Config inside model
        config_class = None
        new_body = []
 
        for item in node.body:
            if isinstance(item, cst.ClassDef) and item.name.value == "Config":
                config_class = item
            else:
                new_body.append(item)
 
        if not config_class:
            return node
 
        # Extract parameters from Config
        config_params = self._extract_config_params(config_class)
 
        # Create model_config = ConfigDict(...)
        config_dict = cst.Assign(
            targets=[cst.AssignTarget(target=cst.Name("model_config"))],
            value=cst.Call(
                func=cst.Name("ConfigDict"),
                args=[
                    cst.Arg(
                        keyword=cst.Name(k),
                        value=cst.Name(str(v)) if isinstance(v, bool) else cst.SimpleString(f'"{v}"')
                    )
                    for k, v in config_params.items()
                ]
            )
        )
 
        # Insert at beginning of class body
        new_body.insert(0, config_dict)
 
        return node.with_changes(body=new_body)
 
# Apply to files
for file_path in all_model_files:
    with open(file_path) as f:
        source = f.read()
 
    tree = cst.parse_module(source)
    transformer = PydanticV2Transformer()
    new_tree = tree.visit(transformer)
 
    with open(file_path, 'w') as f:
        f.write(new_tree.code)

# v1 - worked (but it's an ANTI-PATTERN!)
class Order(BaseModel):
    product_id: int
    quantity: int
 
    @validator('product_id')
    def check_product_exists(cls, v):
        # DB query inside validator
        product = db.query(Product).get(v)
        if not product:
            raise ValueError(f'Product {v} not found')
        return v
 
# v2 - needs rewriting
class Order(BaseModel):
    product_id: int
    quantity: int
 
    # Validator must be @classmethod in v2
    @field_validator('product_id')
    @classmethod
    def check_product_exists(cls, v):
        # ⚠️ Problem: db unavailable in classmethod
        # Solution: remove I/O from validator
        if v <= 0:
            raise ValueError('Product ID must be positive')
        return v
 
    # Moved existence check to endpoint
    @classmethod
    def create_with_validation(cls, data: dict, db: Session):
        order = cls(**data)
        # Check product after schema validation
        product = db.query(Product).get(order.product_id)
        if not product:
            raise HTTPException(404, f'Product {order.product_id} not found')
        return order

# v1
class Event(BaseModel):
    timestamp: datetime
    data: dict
 
    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }
 
# v2 - use field_serializer
class Event(BaseModel):
    timestamp: datetime
    data: dict
 
    @field_serializer('timestamp')
    def serialize_datetime(self, value: datetime) -> str:
        return value.isoformat()

# v1 - worked
for field_name, field in MyModel.__fields__.items():
    print(f"{field_name}: {field.type_}")
 
# v2 - needs rewriting
for field_name, field_info in MyModel.model_fields.items():
    print(f"{field_name}: {field_info.annotation}")

# Ran entire test suite
pytest tests/ -v --cov
 
# Results:
Coverage: 87% (was 87% - unchanged)
Tests passed: 142/145
Tests failed: 3 ← problems!

# Config model
class FeatureFlags(BaseModel):
    new_checkout: bool
    beta_features: bool
    debug_mode: bool
 
# Redis stored strings "true"/"false"
# v1: converted "false" -> False ✅
# v2 strict mode: "false" -> True (any non-empty string) ❌
 
# Result: all feature flags turned on in test stand

class FeatureFlags(BaseModel):
    model_config = ConfigDict(strict=True)  # Only exact types
 
    new_checkout: bool
    beta_features: bool
    debug_mode: bool
 
# Added explicit conversion in Redis loading code
def load_feature_flags(redis_client):
    data = redis_client.hgetall("feature_flags")
    # Explicitly convert strings to bool
    parsed = {
        k: v.lower() == "true" if isinstance(v, str) else v
        for k, v in data.items()
    }
    return FeatureFlags(**parsed)

# Memory profiling
import tracemalloc
 
tracemalloc.start()
 
# Before file processing
snapshot1 = tracemalloc.take_snapshot()
 
# Processing 50k orders
orders = [Order(**row) for row in csv_rows]
process_orders(orders)
 
# After processing
snapshot2 = tracemalloc.take_snapshot()
 
top_stats = snapshot2.compare_to(snapshot1, 'lineno')
for stat in top_stats[:3]:
    print(stat)
 
# Output:
# /app/models/order.py:45: size=842 MiB (+842 MiB), count=50234 (+50234)
# /venv/pydantic/main.py:156: size=234 MiB (+234 MiB), count=150702 (+150702)

# Before (with validation)
orders = [Order(**row) for row in csv_rows]  # 1.2GB memory
 
# After (without validation for trusted data)
orders = [Order.model_construct(**row) for row in csv_rows]  # 380MB memory
 
# Moved validation to CSV loading stage (fail fast)
def validate_csv_header(headers):
    expected = set(Order.model_fields.keys())
    if not expected.issubset(headers):
        raise ValueError(f"Missing columns: {expected - set(headers)}")

# kubernetes/api-gateway-canary.yaml
apiVersion: v1
kind: Service
metadata:
  name: api-gateway-canary
spec:
  selector:
    app: api-gateway
    version: v2-pydantic
  ports:
    - port: 8000
 
---
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: api-gateway-traffic-split
spec:
  hosts:
    - api-gateway
  http:
    - match:
        - headers:
            x-canary:
              exact: "true"
      route:
        - destination:
            host: api-gateway-canary
          weight: 100
    - route:
        - destination:
            host: api-gateway-stable
          weight: 95
        - destination:
            host: api-gateway-canary
          weight: 5 # Started with 5%

# prometheus_metrics.py
from prometheus_client import Histogram, Counter
 
# Metrics for v1 vs v2 comparison
validation_duration = Histogram(
    'pydantic_validation_duration_seconds',
    'Time spent in Pydantic validation',
    ['version', 'model_name']
)
 
validation_errors = Counter(
    'pydantic_validation_errors_total',
    'Total validation errors',
    ['version', 'model_name', 'error_type']
)
 
# In code
def validate_request(data: dict) -> RequestModel:
    version = "v2" if PYDANTIC_V2_ENABLED else "v1"
 
    with validation_duration.labels(version=version, model_name="RequestModel").time():
        try:
            return RequestModel(**data)
        except ValidationError as e:
            for error in e.errors():
                validation_errors.labels(
                    version=version,
                    model_name="RequestModel",
                    error_type=error['type']
                ).inc()
            raise

# Latency comparison v1 vs v2
histogram_quantile(0.95,
  rate(pydantic_validation_duration_seconds_bucket[5m])
) by (version)
 
# Error rate v1 vs v2
rate(pydantic_validation_errors_total[5m]) by (version, error_type)

# Was: global strict
class OrderItem(BaseModel):
    model_config = ConfigDict(strict=True)
    id: int
    qty: int
 
# Became: hybrid approach
class OrderItem(BaseModel):
    model_config = ConfigDict(strict=False)  # Flexible by default
 
    id: int  # Allow "123" -> 123
    qty: int
    price: Decimal = Field(strict=True)  # But money only exact types!

from my_app.config import settings
 
if settings.PYDANTIC_V2_ENABLED:
    from pydantic.v2 import BaseModel, ValidationError
else:
    from pydantic import BaseModel, ValidationError  # v1

# benchmark_versions.py
import timeit
from typing import List
import pydantic.v1 as pydantic_v1
from pydantic import BaseModel as BaseModelV2
 
# Same models on v1 and v2
class UserV1(pydantic_v1.BaseModel):
    id: int
    username: str
    email: str
 
class UserV2(BaseModelV2):
    id: int
    username: str
    email: str
 
# Test data
data = [
    {"id": i, "username": f"user{i}", "email": f"user{i}@example.com"}
    for i in range(10000)
]
 
# Benchmark v1
time_v1 = timeit.timeit(
    lambda: [UserV1(**row) for row in data],
    number=10
) / 10
 
# Benchmark v2
time_v2 = timeit.timeit(
    lambda: [UserV2(**row) for row in data],
    number=10
) / 10
 
print(f"v1: {time_v1:.3f}s")
print(f"v2: {time_v2:.3f}s")
print(f"Speedup: {time_v1/time_v2:.1f}x")

# model_dependencies.py
import ast
import networkx as nx
import matplotlib.pyplot as plt
 
def extract_model_dependencies(file_path):
    """Builds dependency graph between models."""
 
    with open(file_path) as f:
        tree = ast.parse(f.read())
 
    graph = nx.DiGraph()
 
    for node in ast.walk(tree):
        if isinstance(node, ast.ClassDef):
            # Is this a Pydantic model?
            if any(base.id == 'BaseModel' for base in node.bases if isinstance(base, ast.Name)):
                model_name = node.name
                graph.add_node(model_name)
 
                # Look for dependencies in field annotations
                for item in node.body:
                    if isinstance(item, ast.AnnAssign):
                        annotation = item.annotation
                        # Simple case: other_field: OtherModel
                        if isinstance(annotation, ast.Name):
                            graph.add_edge(model_name, annotation.id)
                        # List[OtherModel]
                        elif isinstance(annotation, ast.Subscript):
                            if isinstance(annotation.slice, ast.Name):
                                graph.add_edge(model_name, annotation.slice.id)
 
    return graph
 
# Usage
graph = extract_model_dependencies("models/")
 
# Topological sort for migration order
try:
    migration_order = list(nx.topological_sort(graph))
    print("Migrate in order:", migration_order)
except nx.NetworkXError:
    print("Circular dependencies detected!")
    cycles = list(nx.simple_cycles(graph))
    print("Cycles:", cycles)

# test_pydantic_regression.py
import pytest
from decimal import Decimal
 
# Tests for specific bugs found
def test_feature_flag_string_bool():
    """Feature flags from Redis should parse correctly."""
 
    class Flags(BaseModel):
        enabled: bool
 
    # v1 parsed correctly, v2 needs strict mode
    assert Flags(enabled="true").enabled is True
    assert Flags(enabled="false").enabled is False  # Important!
    assert Flags(enabled=True).enabled is True
 
def test_decimal_precision():
    """Money should preserve precision."""
    class Payment(BaseModel):
        amount: Decimal
 
    # Check that 0.1 + 0.2 = 0.3 exactly
    p1 = Payment(amount="0.1")
    p2 = Payment(amount="0.2")
    assert (p1.amount + p2.amount) == Decimal("0.3")
 
def test_model_construct_memory():
    """model_construct shouldn't create extra objects."""
    import tracemalloc
 
    class Item(BaseModel):
        id: int
        name: str
 
    data = [{"id": i, "name": f"Item {i}"} for i in range(10000)]
 
    tracemalloc.start()
    snapshot1 = tracemalloc.take_snapshot()
 
    items = [Item.model_construct(**row) for row in data]
 
    snapshot2 = tracemalloc.take_snapshot()
    top_stats = snapshot2.compare_to(snapshot1, 'lineno')
 
    # Should be less than 10MB for 10k objects
    total_size = sum(stat.size for stat in top_stats)
    assert total_size < 10 * 1024 * 1024, f"Memory usage too high: {total_size / 1024 / 1024:.1f}MB"