Pydantic Field Validator — The Silent None Bug

from pydantic import BaseModel, EmailStr, ValidationError

# Install pydantic with: pip install pydantic[email]
# The [email] extra adds email validation support via the 'email-validator' package

class UserRegistration(BaseModel):
    username: str          # Must be a string — Pydantic will coerce int/float to str
    email: EmailStr        # Must be a valid email address format, not just any string
    age: int               # Pydantic will coerce '25' -> 25, but fail on 'twenty-five'
    is_active: bool = True # Optional field with a default value

# --- Happy path: valid data coming in from a registration form ---
new_user = UserRegistration(
    username="ada_lovelace",
    email="ada@babbage.co.uk",
    age="28",          # Passed as a string (as it would be from a web form) — Pydantic coerces it
    is_active=True
)

print("=== Successful Registration ===")
print(f"Username : {new_user.username}")
print(f"Email    : {new_user.email}")
print(f"Age      : {new_user.age} (type: {type(new_user.age).__name__})")  # Confirm it's now an int
print(f"Active   : {new_user.is_active}")
print(f"As dict  : {new_user.model_dump()}")  # model_dump() serializes the model to a plain dict

print()

# --- Unhappy path: bad data — what happens in production when input isn't sanitized ---
print("=== Failed Registration (bad data) ===")
try:
    bad_user = UserRegistration(
        username="ghost_user",
        email="not-an-email",   # Fails EmailStr validation
        age="twenty-five",      # Cannot be coerced to int
    )
except ValidationError as validation_error:
    # Pydantic gives us a structured error — we can log it, return it as JSON, or display it
    print(f"Caught {validation_error.error_count()} validation error(s):")
    for error in validation_error.errors():
        print(f"  Field : {' -> '.join(str(loc) for loc in error['loc'])}")
        print(f"  Issue : {error['msg']}")
        print(f"  Input : {error['input']}")
        print()

from datetime import date
from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional

class ProductListing(BaseModel):
    # Field() constraints are checked BEFORE custom validators run
    name: str = Field(
        min_length=2,
        max_length=100,
        description="Product display name"
    )
    price_usd: float = Field(
        gt=0,           # gt = greater than (strictly). Use ge= for greater-than-or-equal
        le=99999.99,    # le = less than or equal
        description="Price in US dollars"
    )
    discount_pct: Optional[float] = Field(
        default=0.0,
        ge=0.0,   # Cannot be negative
        lt=100.0  # Cannot be 100% or more (would be free or negative price)
    )
    available_from: date
    available_until: Optional[date] = None
    sku: str = Field(
        pattern=r'^[A-Z]{2,4}-\d{4,8}$',  # e.g. ELEC-00142 or TV-9982100
        description="Stock Keeping Unit code"
    )

    # @field_validator runs AFTER the field is already its correct Python type
    @field_validator('name')
    @classmethod
    def name_must_not_be_generic(cls, product_name: str) -> str:
        # Strip whitespace first — a common real-world data cleaning step
        cleaned_name = product_name.strip()
        generic_names = {'product', 'item', 'thing', 'unnamed', 'test'}
        if cleaned_name.lower() in generic_names:
            raise ValueError(f"Product name '{cleaned_name}' is too generic. Use a specific name.")
        return cleaned_name  # Always return the (possibly cleaned) value

    # @model_validator with mode='after' has access to the fully-built model instance
    @model_validator(mode='after')
    def availability_window_must_be_valid(self) -> 'ProductListing':
        if self.available_until is not None:
            if self.available_until <= self.available_from:
                raise ValueError(
                    f"available_until ({self.available_until}) must be "
                    f"after available_from ({self.available_from})"
                )
        return self  # Always return self from a model_validator


# --- Test 1: Valid product ---
laptop = ProductListing(
    name="  ThinkPad X1 Carbon  ",  # Leading/trailing spaces — our validator cleans this
    price_usd=1299.99,
    discount_pct=15.0,
    available_from=date(2024, 1, 1),
    available_until=date(2024, 12, 31),
    sku="COMP-00891"
)
print(f"Product name (cleaned): '{laptop.name}'")
print(f"Price after discount: ${laptop.price_usd * (1 - laptop.discount_pct / 100):.2f}")
print()

# --- Test 2: Invalid product — multiple errors at once ---
from pydantic import ValidationError
try:
    bad_product = ProductListing(
        name="product",             # Fails our custom validator
        price_usd=-50.0,            # Fails gt=0 constraint
        available_from=date(2024, 6, 1),
        available_until=date(2024, 1, 1),  # Fails model validator (before < after)
        sku="invalid-sku-format"    # Fails regex pattern
    )
except ValidationError as e:
    print(f"Found {e.error_count()} errors:")
    for err in e.errors():
        field = ' -> '.join(str(loc) for loc in err['loc'])
        print(f"  [{field}] {err['msg']}")

from pydantic import BaseModel, Field, field_validator
from typing import List
from datetime import datetime
from enum import Enum

class OrderStatus(str, Enum):
    # Inheriting from str makes JSON serialization work seamlessly
    PENDING   = "pending"
    CONFIRMED = "confirmed"
    SHIPPED   = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"

class ShippingAddress(BaseModel):
    street_line_1: str = Field(min_length=5)
    street_line_2: str = ""
    city: str
    postcode: str
    country_code: str = Field(pattern=r'^[A-Z]{2}$')  # ISO 3166-1 alpha-2, e.g. 'US', 'GB'

class OrderItem(BaseModel):
    product_sku: str
    product_name: str
    quantity: int = Field(ge=1)            # Must order at least 1
    unit_price_usd: float = Field(gt=0)

    @property
    def line_total(self) -> float:
        # Computed property — not a stored field, so Pydantic doesn't validate it
        # but it's always consistent with the validated data
        return round(self.quantity * self.unit_price_usd, 2)

class CustomerOrder(BaseModel):
    order_id: str
    customer_email: str
    status: OrderStatus = OrderStatus.PENDING  # Enum default
    shipping_address: ShippingAddress           # Nested model — validated recursively
    items: List[OrderItem] = Field(min_length=1) # Order must have at least one item
    created_at: datetime = Field(default_factory=datetime.utcnow)  # Auto-set on creation

    @field_validator('order_id')
    @classmethod
    def order_id_must_be_uppercase(cls, raw_id: str) -> str:
        return raw_id.upper()  # Normalise — so 'ord-1234' and 'ORD-1234' are treated the same

    @property
    def order_total(self) -> float:
        return round(sum(item.line_total for item in self.items), 2)


# Simulating a JSON payload arriving from an API client (e.g. a mobile app checkout)
# In real life this would be: CustomerOrder(**request.json()) or FastAPI does it automatically
raw_payload = {
    "order_id": "ord-20240815-9921",
    "customer_email": "grace@hopper.dev",
    "shipping_address": {
        "street_line_1": "123 Compiler Street",
        "city": "Arlington",
        "postcode": "22201",
        "country_code": "US"
    },
    "items": [
        {"product_sku": "BOOK-0042", "product_name": "The Art of Computer Programming", "quantity": 2, "unit_price_usd": 79.99},
        {"product_sku": "BOOK-0187", "product_name": "Clean Code", "quantity": 1, "unit_price_usd": 34.50}
    ]
}

order = CustomerOrder(**raw_payload)

print(f"Order ID    : {order.order_id}")             # Note: auto-uppercased
print(f"Customer    : {order.customer_email}")
print(f"Status      : {order.status.value}")
print(f"Ship to     : {order.shipping_address.city}, {order.shipping_address.country_code}")
print(f"Items       : {len(order.items)}")
for item in order.items:
    print(f"  - {item.product_name} x{item.quantity} @ ${item.unit_price_usd} = ${item.line_total}")
print(f"Order Total : ${order.order_total}")
print()

# Serialise the whole nested structure to a dict (ready to store in a database or return as JSON)
order_dict = order.model_dump()
print("Serialised keys at top level:", list(order_dict.keys()))
print("Nested address keys:", list(order_dict['shipping_address'].keys()))

# Install: pip install pydantic-settings
# Create a .env file in the same directory with the content shown below before running

from pydantic import Field, AnyHttpUrl, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import List

# --- Contents of your .env file (do NOT commit this to git) ---
# DATABASE_URL=postgresql://user:pass@localhost:5432/myapp
# API_SECRET_KEY=super-secret-key-change-this-in-prod
# DEBUG_MODE=false
# ALLOWED_ORIGINS=http://localhost:3000,https://myapp.com
# MAX_CONNECTIONS=10

class AppSettings(BaseSettings):
    # SettingsConfigDict tells Pydantic WHERE to look for values
    model_config = SettingsConfigDict(
        env_file='.env',            # Read from .env file
        env_file_encoding='utf-8',
        case_sensitive=False,       # DATABASE_URL and database_url are the same
        extra='ignore'              # Don't crash if .env has extra keys we don't declare
    )

    # Required fields — no default means the app REFUSES to start if these are missing
    database_url: str = Field(description="Full PostgreSQL connection string")
    api_secret_key: str = Field(min_length=16, description="JWT signing secret")

    # Optional fields with sensible defaults
    debug_mode: bool = False
    app_port: int = Field(default=8000, ge=1024, le=65535)
    max_connections: int = Field(default=5, ge=1, le=100)

    # A comma-separated list in an env var — Pydantic can handle this pattern
    allowed_origins: List[str] = Field(default=["http://localhost:3000"])

    @field_validator('api_secret_key')
    @classmethod
    def secret_key_must_not_be_default(cls, key_value: str) -> str:
        insecure_defaults = {'secret', 'password', 'changeme', 'default', 'test'}
        if key_value.lower() in insecure_defaults:
            raise ValueError(
                "API_SECRET_KEY is set to an insecure default value. "
                "Set a strong random secret before deploying."
            )
        return key_value


# This is the pattern: load settings ONCE at module level.
# Import `settings` from this module everywhere else — never call AppSettings() twice.
try:
    settings = AppSettings()
    print("=== Application Config Loaded Successfully ===")
    print(f"Database     : {settings.database_url[:30]}...")  # Don't log full credentials
    print(f"Port         : {settings.app_port}")
    print(f"Debug Mode   : {settings.debug_mode}")
    print(f"Max Conns    : {settings.max_connections}")
    print(f"CORS Origins : {settings.allowed_origins}")
except Exception as config_error:
    # In a real app you'd use logging.critical() here and sys.exit(1)
    print(f"FATAL: Could not load application config — {config_error}")
    print("Application startup aborted.")

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr, Field
from typing import List, Optional

app = FastAPI(title="User API")

class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=50)
    email: EmailStr
    age: int = Field(ge=18, le=120)  # Must be adult age

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    age: int
    is_active: bool = True

# In-memory store for demo
fake_db: List[UserResponse] = []

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    # FastAPI has already validated user entirely via Pydantic
    # No need to parse request.json() or check field types
    new_id = len(fake_db) + 1
    new_user = UserResponse(id=new_id, **user.model_dump())
    fake_db.append(new_user)
    return new_user

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    if user_id < 1 or user_id > len(fake_db):
        raise HTTPException(status_code=404, detail="User not found")
    return fake_db[user_id - 1]

# To run: uvicorn fastapi_pydantic_integration:app --reload
# Try POST /users with {"username":"alice", "email":"alice@example.com", "age":30}
# Then GET /users/1
# Swagger docs at /docs