Mastering Python Type Hints and Pydantic Models for Robust Data Validation
Published on October 8, 2025
Introduction: The Journey from Dynamic to Typed Python
When I first started my journey with Python over a decade ago, one of the things that drew me to the language was its dynamic nature. You could create variables without declaring types, write functions that accepted any parameter, and let Python figure out the details at runtime. It felt liberating compared to the rigid type systems of languages like Java or C++.
However, as my applications grew in complexity and my teams expanded, I began to encounter the challenges that come with this flexibility. Debugging became more difficult, code maintenance was time-consuming, and onboarding new team members required extensive documentation to understand what types of data functions expected.
The introduction of type hints in Python 3.5 and the rise of powerful validation libraries like Pydantic have fundamentally changed how I approach Python development. Today, I want to share my experience and insights about leveraging Python type hints and Pydantic models to build more robust, maintainable applications.
Understanding Python Type Hints: More Than Just Documentation
What Are Type Hints?
Python type hints, introduced in PEP 484, are a way to indicate the expected types of variables, function parameters, and return values. Despite the name "hints," they serve as much more than suggestions – they're a powerful tool for static analysis, IDE support, and runtime validation when combined with the right libraries.
# Without type hints - unclear what's expecteddefprocess_user_data(name,age,email):returnf"User: {name}, Age: {age}, Email: {email}"# With type hints - crystal clear expectationsdefprocess_user_data(name:str,age:int,email:str)->str:returnf"User: {name}, Age: {age}, Email: {email}"
The Evolution of Type Hints
Over the years, Python's type system has evolved significantly:
Python 3.5: Basic type hints with typing module
Python 3.6: Variable annotations and improved typing support
Python 3.8: typing.Literal, typing.Final, and typing.Protocol
Python 3.9: Built-in generics (no need to import from typing)
Python 3.10: Union types with | operator
Python 3.12: Generic type syntax improvements
Advanced Type Hints in Practice
Let me show you some advanced patterns I use regularly in my projects:
Enter Pydantic: Runtime Validation Meets Type Hints
While type hints provide excellent static analysis and IDE support, they don't perform runtime validation by default. This is where Pydantic shines – it bridges the gap between type hints and runtime data validation.
Why Pydantic?
Based on my experience building production APIs and data processing pipelines, here's why Pydantic has become indispensable:
Automatic validation: Converts and validates data based on type hints
Performance: Built on top of pydantic-core (written in Rust)
JSON Schema generation: Automatic API documentation
IDE support: Full type checking and autocompletion
Basic Pydantic Models
Real-World Application: Building a FastAPI Service
Let me share a practical example from a recent project – a user management API built with FastAPI and Pydantic:
Domain Models
FastAPI Endpoints
Advanced Pydantic Patterns and Techniques
Custom Validators and Serializers
Working with Nested Models and Complex Data
Data Flow with Mermaid Sequence Diagrams
To better understand how type hints and Pydantic work together in a typical application flow, let's visualize the data validation process:
Data Transformation Pipeline
Performance Considerations and Best Practices
Optimization Strategies
Based on my experience with high-traffic applications, here are key performance considerations:
Memory Management
Integration with Modern Python Frameworks
FastAPI Integration
SQLAlchemy Integration
Testing Strategies for Type-Safe Code
Comprehensive Test Suite
Error Handling and Debugging
Advanced Error Handling
Debugging and Monitoring
Production Deployment and Monitoring
Configuration Management
Health Checks and Monitoring
Conclusion: The Future of Type-Safe Python
As I reflect on my journey from writing dynamically-typed Python to embracing type hints and Pydantic models, I can confidently say that this transformation has made me a more effective and confident developer. The combination of static type checking and runtime validation provides the perfect balance between Python's flexibility and the reliability needed for production applications.
Key Takeaways
Gradual Adoption: You don't need to convert your entire codebase overnight. Start with new features and critical components.
Tool Integration: Leverage IDE support, mypy, and Pydantic together for comprehensive type safety.
Performance Considerations: Type hints and validation have minimal runtime overhead when implemented correctly.
Testing Benefits: Type-safe code is easier to test and debug, leading to higher quality applications.
Team Productivity: Type hints serve as living documentation, making codebases more accessible to team members.
Looking Forward
The Python ecosystem continues to evolve toward better type safety:
PEP 695 (Python 3.12): Simplified generic type syntax
PEP 647: User-Defined Type Guards
Pydantic V2: Rust-based core for improved performance
FastAPI Evolution: Deeper integration with modern Python type features
Final Recommendations
For developers considering adopting type hints and Pydantic:
Start Small: Begin with a single module or new feature
Invest in Tooling: Set up mypy, configure your IDE, and use type-aware linters
Focus on Boundaries: Prioritize type safety at API boundaries and data models
Embrace the Learning Curve: The initial investment pays dividends in code quality and maintainability
Stay Updated: The type system is rapidly evolving with new features and improvements
Type hints and Pydantic models have fundamentally changed how I approach Python development. They've made my code more robust, my debugging sessions shorter, and my confidence in deploying to production higher. I encourage every Python developer to explore these tools and experience the benefits of type-safe Python development.
Have you implemented type hints and Pydantic models in your Python projects? Share your experiences and challenges in the comments below. Let's continue the conversation about building better, more reliable Python applications together.
```python
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Dict, List, Optional
from enum import Enum
import asyncio
import aioredis
import asyncpg
class HealthStatus(str, Enum):
from pydantic import BaseModel, Field, EmailStr, validator
from typing import Optional, List
from datetime import datetime
from enum import Enum
class UserRole(str, Enum):
ADMIN = "admin"
USER = "user"
MODERATOR = "moderator"
class User(BaseModel):
id: int
name: str = Field(..., min_length=1, max_length=100)
email: EmailStr
age: int = Field(..., ge=0, le=150)
role: UserRole = UserRole.USER
created_at: datetime = Field(default_factory=datetime.now)
tags: Optional[List[str]] = None
@validator('name')
def validate_name(cls, v):
if not v.strip():
raise ValueError('Name cannot be empty or just whitespace')
return v.title()
class Config:
# Enable validation for assignment
validate_assignment = True
# Use enum values for serialization
use_enum_values = True
# Usage example
try:
user_data = {
"id": "123", # Will be converted to int
"name": "john doe", # Will be title-cased
"email": "[email protected]",
"age": 30,
"role": "admin"
}
user = User(**user_data)
print(f"Created user: {user.name} ({user.email})")
print(f"User JSON: {user.json()}")
except ValueError as e:
print(f"Validation error: {e}")
from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.responses import JSONResponse
from typing import List, Dict
from uuid import uuid4, UUID
from datetime import datetime
import asyncio
app = FastAPI(title="User Management API", version="1.0.0")
# In-memory storage for demo (use database in production)
users_db: Dict[UUID, UserResponse] = {}
@app.post("/users/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(user_data: UserCreate) -> UserResponse:
"""Create a new user with comprehensive validation."""
# Check if email already exists
for existing_user in users_db.values():
if existing_user.email == user_data.email:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered"
)
# Create user
user_id = uuid4()
new_user = UserResponse(
id=user_id,
name=user_data.name,
email=user_data.email,
age=user_data.age,
role=user_data.role,
created_at=datetime.now(),
profile=user_data.profile
)
users_db[user_id] = new_user
return new_user
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: UUID) -> UserResponse:
"""Retrieve a user by ID."""
if user_id not in users_db:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return users_db[user_id]
@app.put("/users/{user_id}", response_model=UserResponse)
async def update_user(user_id: UUID, user_update: UserUpdate) -> UserResponse:
"""Update an existing user."""
if user_id not in users_db:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
current_user = users_db[user_id]
# Apply updates only for provided fields
update_data = user_update.dict(exclude_unset=True)
updated_user = current_user.copy(update=update_data)
updated_user.updated_at = datetime.now()
users_db[user_id] = updated_user
return updated_user
@app.get("/users/", response_model=List[UserResponse])
async def list_users(
role: Optional[UserRole] = None,
limit: int = Field(10, ge=1, le=100)
) -> List[UserResponse]:
"""List users with optional filtering."""
users = list(users_db.values())
if role:
users = [user for user in users if user.role == role]
return users[:limit]
from pydantic import BaseModel, validator, root_validator, Field, EmailStr
from typing import List, Optional, Dict
from datetime import datetime
import re
class AdvancedUser(BaseModel):
username: str = Field(..., min_length=3, max_length=20)
email: EmailStr
phone: Optional[str] = None
social_handles: Optional[Dict[str, str]] = None
tags: List[str] = Field(default_factory=list)
@validator('username')
def validate_username(cls, v):
"""Ensure username contains only alphanumeric characters and underscores."""
if not re.match(r'^[a-zA-Z0-9_]+$', v):
raise ValueError('Username must contain only letters, numbers, and underscores')
return v.lower()
@validator('phone')
def validate_phone(cls, v):
"""Validate phone number format."""
if v is None:
return v
# Remove all non-digit characters
digits = re.sub(r'\D', '', v)
if len(digits) != 10:
raise ValueError('Phone number must be 10 digits')
# Format as (XXX) XXX-XXXX
return f"({digits[:3]}) {digits[3:6]}-{digits[6:]}"
@validator('tags', each_item=True)
def validate_tags(cls, v):
"""Ensure tags are lowercase and alphanumeric."""
if not isinstance(v, str):
raise ValueError('Tags must be strings')
return v.lower().strip()
@root_validator
def validate_social_handles(cls, values):
"""Validate social media handles format."""
social_handles = values.get('social_handles')
if not social_handles:
return values
valid_platforms = {'twitter', 'linkedin', 'github', 'instagram'}
for platform, handle in social_handles.items():
if platform not in valid_platforms:
raise ValueError(f'Unsupported platform: {platform}')
if not handle.startswith('@'):
social_handles[platform] = f'@{handle}'
return values
class Config:
validate_assignment = True
json_encoders = {
datetime: lambda v: v.isoformat()
}
from pydantic import BaseModel, Field
from typing import List, Dict, Union, Optional
from datetime import datetime
from enum import Enum
from uuid import UUID
class EventType(str, Enum):
LOGIN = "login"
LOGOUT = "logout"
PURCHASE = "purchase"
VIEW = "view"
class EventMetadata(BaseModel):
ip_address: str
user_agent: str
session_id: str
timestamp: datetime = Field(default_factory=datetime.now)
class BaseEvent(BaseModel):
event_type: EventType
user_id: UUID
metadata: EventMetadata
class LoginEvent(BaseEvent):
event_type: EventType = EventType.LOGIN
success: bool
failure_reason: Optional[str] = None
class PurchaseEvent(BaseEvent):
event_type: EventType = EventType.PURCHASE
product_id: str
amount: float = Field(..., gt=0)
currency: str = Field(default="USD", min_length=3, max_length=3)
discount_applied: Optional[float] = Field(None, ge=0, le=1)
class EventProcessor(BaseModel):
"""Process different types of events with proper validation."""
@staticmethod
def process_event(event_data: Dict) -> Union[LoginEvent, PurchaseEvent, BaseEvent]:
"""Factory method to create appropriate event type."""
event_type = event_data.get('event_type')
if event_type == EventType.LOGIN:
return LoginEvent(**event_data)
elif event_type == EventType.PURCHASE:
return PurchaseEvent(**event_data)
else:
return BaseEvent(**event_data)
from pydantic import BaseModel, Field, validator
from typing import List, Optional, ClassVar
import time
from functools import lru_cache
class OptimizedUser(BaseModel):
"""Optimized Pydantic model for high-performance scenarios."""
# Use class variables for constants to avoid repeated validation
MAX_TAGS: ClassVar[int] = 10
id: int
name: str = Field(..., min_length=1, max_length=100)
email: str # Use str instead of EmailStr if validation is done elsewhere
tags: List[str] = Field(default_factory=list, max_items=10)
@validator('tags', pre=True)
def optimize_tags_validation(cls, v):
"""Pre-process tags to avoid expensive operations later."""
if isinstance(v, str):
return [tag.strip().lower() for tag in v.split(',') if tag.strip()]
return v
@lru_cache(maxsize=128)
@classmethod
def get_schema(cls):
"""Cache schema generation for better performance."""
return cls.schema()
class Config:
# Disable validation for assignment to improve performance
validate_assignment = False
# Use faster JSON serialization
json_encoders = {
datetime: lambda v: int(v.timestamp())
}
# Allow population by field name for ORM integration
allow_population_by_field_name = True
from pydantic import BaseModel
from typing import List, Iterator
import sys
class EfficientBatchProcessor:
"""Process large datasets efficiently with Pydantic."""
@staticmethod
def process_batch(data_batch: List[dict], model_class: BaseModel) -> Iterator[BaseModel]:
"""Process data in batches to manage memory usage."""
for item in data_batch:
try:
yield model_class(**item)
except ValueError as e:
# Log error and continue processing
print(f"Validation error for item {item.get('id', 'unknown')}: {e}")
continue
@staticmethod
def measure_memory_usage():
"""Monitor memory usage during processing."""
return sys.getsizeof([])
# Usage example for large datasets
def process_large_dataset(file_path: str, batch_size: int = 1000):
"""Process large CSV file with memory-efficient batching."""
import csv
with open(file_path, 'r') as file:
reader = csv.DictReader(file)
batch = []
for row in reader:
batch.append(row)
if len(batch) >= batch_size:
# Process batch
processed_items = list(
EfficientBatchProcessor.process_batch(batch, OptimizedUser)
)
# Clear batch to free memory
batch.clear()
# Yield processed items
yield from processed_items
from fastapi import FastAPI, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from typing import Optional, List
from enum import Enum
app = FastAPI(
title="Advanced User API",
description="Production-ready API with comprehensive type validation",
version="2.0.0"
)
class SortOrder(str, Enum):
ASC = "asc"
DESC = "desc"
class UserFilter(BaseModel):
"""Query parameters for user filtering."""
role: Optional[UserRole] = None
min_age: Optional[int] = Field(None, ge=0, le=150)
max_age: Optional[int] = Field(None, ge=0, le=150)
search: Optional[str] = Field(None, min_length=1, max_length=100)
sort_by: str = Field(default="created_at")
sort_order: SortOrder = SortOrder.ASC
page: int = Field(default=1, ge=1)
page_size: int = Field(default=20, ge=1, le=100)
@app.get("/users/search", response_model=List[UserResponse])
async def search_users(
filter_params: UserFilter = Depends()
) -> List[UserResponse]:
"""Search users with advanced filtering and pagination."""
# Apply filters (implementation would use actual database)
filtered_users = []
for user in users_db.values():
if filter_params.role and user.role != filter_params.role:
continue
if filter_params.min_age and user.age < filter_params.min_age:
continue
if filter_params.max_age and user.age > filter_params.max_age:
continue
if filter_params.search and filter_params.search.lower() not in user.name.lower():
continue
filtered_users.append(user)
# Apply sorting
reverse = filter_params.sort_order == SortOrder.DESC
filtered_users.sort(
key=lambda u: getattr(u, filter_params.sort_by),
reverse=reverse
)
# Apply pagination
start_idx = (filter_params.page - 1) * filter_params.page_size
end_idx = start_idx + filter_params.page_size
return filtered_users[start_idx:end_idx]
from sqlalchemy import Column, Integer, String, DateTime, Enum as SQLEnum
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from pydantic import BaseModel
from datetime import datetime
Base = declarative_base()
class UserDB(Base):
"""SQLAlchemy model for database storage."""
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), nullable=False)
email = Column(String(255), unique=True, index=True, nullable=False)
age = Column(Integer, nullable=False)
role = Column(SQLEnum(UserRole), default=UserRole.USER)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, onupdate=datetime.utcnow)
class UserService:
"""Service layer with type-safe database operations."""
def __init__(self, db_session):
self.db = db_session
def create_user(self, user_data: UserCreate) -> UserResponse:
"""Create user with automatic validation and conversion."""
db_user = UserDB(
name=user_data.name,
email=user_data.email,
age=user_data.age,
role=user_data.role
)
self.db.add(db_user)
self.db.commit()
self.db.refresh(db_user)
# Convert SQLAlchemy model to Pydantic model
return UserResponse.from_orm(db_user)
def get_user(self, user_id: int) -> Optional[UserResponse]:
"""Retrieve user with type-safe conversion."""
db_user = self.db.query(UserDB).filter(UserDB.id == user_id).first()
if db_user:
return UserResponse.from_orm(db_user)
return None
import pytest
from pydantic import ValidationError
from typing import List
from datetime import datetime
from uuid import uuid4
```python
import pytest
from pydantic import ValidationError
from typing import List
from datetime import datetime
from uuid import uuid4
class TestUserValidation:
"""Test suite for user model validation."""
def test_valid_user_creation(self):
"""Test creating a user with valid data."""
user_data = {
"name": "John Doe",
"email": "[email protected]",
"age": 30,
"role": UserRole.USER
}
user = UserCreate(**user_data)
assert user.name == "John Doe"
assert user.email == "[email protected]"
assert user.age == 30
assert user.role == UserRole.USER
def test_invalid_email_format(self):
"""Test validation error for invalid email format."""
user_data = {
"name": "John Doe",
"email": "invalid-email",
"age": 30
}
with pytest.raises(ValidationError) as exc_info:
UserCreate(**user_data)
errors = exc_info.value.errors()
assert any(error['loc'] == ('email',) for error in errors)
def test_age_constraints(self):
"""Test age validation constraints."""
# Test minimum age
with pytest.raises(ValidationError):
UserCreate(
name="Child",
email="[email protected]",
age=12, # Below minimum age of 13
password="password123"
)
# Test maximum age
with pytest.raises(ValidationError):
UserCreate(
name="Ancient",
email="[email protected]",
age=200, # Above maximum age of 150
password="password123"
)
def test_nested_model_validation(self):
"""Test validation of nested address model."""
user_data = {
"name": "John Doe",
"email": "[email protected]",
"age": 30,
"password": "password123",
"profile": {
"bio": "Software developer",
"address": {
"street": "123 Main St",
"city": "New York",
"state": "NY",
"zip_code": "10001"
}
}
}
user = UserCreate(**user_data)
assert user.profile.address.city == "New York"
assert user.profile.address.zip_code == "10001"
@pytest.mark.parametrize("invalid_zip", [
"123", # Too short
"1234567", # Too long
"abcde", # Non-numeric
"12345-abc" # Invalid extended format
])
def test_invalid_zip_codes(self, invalid_zip):
"""Test various invalid zip code formats."""
user_data = {
"name": "John Doe",
"email": "[email protected]",
"age": 30,
"password": "password123",
"profile": {
"address": {
"street": "123 Main St",
"city": "New York",
"state": "NY",
"zip_code": invalid_zip
}
}
}
with pytest.raises(ValidationError) as exc_info:
UserCreate(**user_data)
errors = exc_info.value.errors()
assert any('zip_code' in str(error['loc']) for error in errors)
class TestTypeHintIntegration:
"""Test type hint integration with runtime validation."""
def test_function_with_type_hints(self):
"""Test function that uses type hints with Pydantic models."""
def process_user_batch(users: List[UserCreate]) -> List[UserResponse]:
"""Process a batch of users with type validation."""
processed = []
for user_data in users:
# Convert to response model (simulating database save)
response_user = UserResponse(
id=uuid4(),
name=user_data.name,
email=user_data.email,
age=user_data.age,
role=user_data.role,
created_at=datetime.now()
)
processed.append(response_user)
return processed
# Test with valid data
users_data = [
UserCreate(
name="User 1",
email="[email protected]",
age=25,
password="password123"
),
UserCreate(
name="User 2",
email="[email protected]",
age=30,
password="password456"
)
]
result = process_user_batch(users_data)
assert len(result) == 2
assert all(isinstance(user, UserResponse) for user in result)
from pydantic import ValidationError, BaseModel
from fastapi import HTTPException, status
from typing import List, Dict, Any
import logging
logger = logging.getLogger(__name__)
class ErrorDetail(BaseModel):
"""Structured error detail for API responses."""
field: str
message: str
invalid_value: Any
error_type: str
class ValidationErrorResponse(BaseModel):
"""Comprehensive validation error response."""
message: str
errors: List[ErrorDetail]
error_count: int
def handle_validation_error(error: ValidationError) -> ValidationErrorResponse:
"""Convert Pydantic validation errors to structured response."""
error_details = []
for error_detail in error.errors():
field_path = ".".join(str(loc) for loc in error_detail['loc'])
error_details.append(
ErrorDetail(
field=field_path,
message=error_detail['msg'],
invalid_value=error_detail.get('input', 'N/A'),
error_type=error_detail['type']
)
)
return ValidationErrorResponse(
message="Validation failed",
errors=error_details,
error_count=len(error_details)
)
# Usage in FastAPI
@app.exception_handler(ValidationError)
async def validation_exception_handler(request, exc: ValidationError):
"""Global validation error handler."""
logger.warning(f"Validation error for {request.url}: {exc}")
error_response = handle_validation_error(exc)
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content=error_response.dict()
)
from pydantic import BaseSettings, BaseModel, Field
from typing import Optional
import os
class DatabaseConfig(BaseModel):
"""Database configuration with validation."""
host: str = Field(..., env='DB_HOST')
port: int = Field(5432, env='DB_PORT', ge=1, le=65535)
username: str = Field(..., env='DB_USERNAME')
password: str = Field(..., env='DB_PASSWORD')
database: str = Field(..., env='DB_NAME')
@property
def connection_string(self) -> str:
"""Generate database connection string."""
return f"postgresql://{self.username}:{self.password}@{self.host}:{self.port}/{self.database}"
class RedisConfig(BaseModel):
"""Redis configuration with validation."""
host: str = Field("localhost", env='REDIS_HOST')
port: int = Field(6379, env='REDIS_PORT', ge=1, le=65535)
database: int = Field(0, env='REDIS_DB', ge=0, le=15)
password: Optional[str] = Field(None, env='REDIS_PASSWORD')
class AppSettings(BaseSettings):
"""Application settings with environment variable support."""
# Application settings
app_name: str = Field("User Management API", env='APP_NAME')
debug: bool = Field(False, env='DEBUG')
version: str = Field("1.0.0", env='APP_VERSION')
# Server settings
host: str = Field("0.0.0.0", env='HOST')
port: int = Field(8000, env='PORT', ge=1, le=65535)
workers: int = Field(1, env='WORKERS', ge=1, le=8)
# Security settings
secret_key: str = Field(..., env='SECRET_KEY', min_length=32)
jwt_algorithm: str = Field("HS256", env='JWT_ALGORITHM')
jwt_expiration: int = Field(3600, env='JWT_EXPIRATION', ge=300) # seconds
# External services
database: DatabaseConfig
redis: RedisConfig
# Feature flags
enable_user_registration: bool = Field(True, env='ENABLE_USER_REGISTRATION')
enable_email_verification: bool = Field(True, env='ENABLE_EMAIL_VERIFICATION')
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
case_sensitive = False
# Global settings instance
settings = AppSettings()
# Usage in application
@app.on_event("startup")
async def startup_event():
"""Initialize application with validated settings."""
logger.info(f"Starting {settings.app_name} v{settings.version}")
logger.info(f"Debug mode: {settings.debug}")
logger.info(f"Database: {settings.database.host}:{settings.database.port}")
# Validate critical settings
if len(settings.secret_key) < 32:
raise ValueError("SECRET_KEY must be at least 32 characters long")
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Dict, List, Optional
from enum import Enum
import asyncio
import aioredis
import asyncpg
class HealthStatus(str, Enum):
HEALTHY = "healthy"
UNHEALTHY = "unhealthy"
DEGRADED = "degraded"
class ComponentHealth(BaseModel):
"""Health status of individual system components."""
name: str
status: HealthStatus
response_time_ms: Optional[float] = None
error_message: Optional[str] = None
last_checked: datetime = Field(default_factory=datetime.now)
class SystemHealth(BaseModel):
"""Overall system health status."""
status: HealthStatus
timestamp: datetime = Field(default_factory=datetime.now)
version: str
uptime_seconds: float
components: List[ComponentHealth]
metrics: Dict[str, float] = Field(default_factory=dict)
class HealthChecker:
"""Comprehensive health checking system."""
def __init__(self, settings: AppSettings):
self.settings = settings
self.start_time = datetime.now()
async def check_database(self) -> ComponentHealth:
"""Check database connectivity and performance."""
start_time = datetime.now()
try:
conn = await asyncpg.connect(
self.settings.database.connection_string
)
await conn.execute("SELECT 1")
await conn.close()
response_time = (datetime.now() - start_time).total_seconds() * 1000
return ComponentHealth(
name="database",
status=HealthStatus.HEALTHY,
response_time_ms=response_time
)
except Exception as e:
return ComponentHealth(
name="database",
status=HealthStatus.UNHEALTHY,
error_message=str(e)
)
async def check_redis(self) -> ComponentHealth:
"""Check Redis connectivity and performance."""
start_time = datetime.now()
try:
redis = await aioredis.from_url(
f"redis://{self.settings.redis.host}:{self.settings.redis.port}"
)
await redis.ping()
await redis.close()
response_time = (datetime.now() - start_time).total_seconds() * 1000
return ComponentHealth(
name="redis",
status=HealthStatus.HEALTHY,
response_time_ms=response_time
)
except Exception as e:
return ComponentHealth(
name="redis",
status=HealthStatus.UNHEALTHY,
error_message=str(e)
)
async def get_system_health(self) -> SystemHealth:
"""Get comprehensive system health status."""
components = await asyncio.gather(
self.check_database(),
self.check_redis(),
return_exceptions=True
)
# Filter out exceptions and create default unhealthy status
component_health = []
for comp in components:
if isinstance(comp, Exception):
component_health.append(
ComponentHealth(
name="unknown",
status=HealthStatus.UNHEALTHY,
error_message=str(comp)
)
)
else:
component_health.append(comp)
# Determine overall system status
unhealthy_count = sum(
1 for comp in component_health
if comp.status == HealthStatus.UNHEALTHY
)
if unhealthy_count == 0:
overall_status = HealthStatus.HEALTHY
elif unhealthy_count == len(component_health):
overall_status = HealthStatus.UNHEALTHY
else:
overall_status = HealthStatus.DEGRADED
uptime = (datetime.now() - self.start_time).total_seconds()
return SystemHealth(
status=overall_status,
version=self.settings.version,
uptime_seconds=uptime,
components=component_health,
metrics={
"total_components": len(component_health),
"healthy_components": len(component_health) - unhealthy_count,
"unhealthy_components": unhealthy_count
}
)
# Health check endpoint
health_checker = HealthChecker(settings)
@app.get("/health", response_model=SystemHealth)
async def health_check() -> SystemHealth:
"""Comprehensive health check endpoint."""
return await health_checker.get_system_health()
