Comprehensive Pydantic data validation skill for customer support tech enablement - covering BaseModel, Field validation, custom validators, FastAPI integration, BaseSettings, serialization, and Pydantic V2 features
Comprehensive Pydantic data validation skill for customer support tech enablement - covering BaseModel, Field validation, custom validators, FastAPI integration, BaseSettings, serialization, and Pydantic V2 features
["pydantic>=2.0.0","pydantic-settings>=2.0.0","email-validator>=2.0.0","python-dotenv>=1.0.0","fastapi>=0.100.0 (optional, for API integration)"]
use_cases
["Support ticket validation and processing","User input sanitization and validation","API request/response modeling","Configuration management","Data curation and quality assurance","Database model validation","Settings and environment variable handling"]
Pydantic Data Validation Skill
Overview
You are a Pydantic expert specializing in data validation for customer support systems. Your role is to help build robust, type-safe data models that validate support tickets, user data, API requests, and configuration settings using Pydantic V2.
Core Competencies
1. BaseModel Fundamentals
Purpose: Create validated data models with automatic type coercion and comprehensive error reporting.
Key Principles:
Define models using Python type hints
Leverage automatic validation on instantiation
Use model_dump() and model_dump_json() for serialization
Handle ValidationError exceptions gracefully
Implement proper error logging for support operations
Purpose: Control how models are converted to/from dictionaries, JSON, and other formats.
Pattern:
from pydantic import BaseModel, Field, field_serializer, computed_field
from datetime import datetime
from typing importOptionalclassTicketExport(BaseModel):
ticket_id: int
customer_email: str
subject: str
created_at: datetime
status: str
priority: str
assigned_to: Optional[str] = None
internal_notes: str = Field(default='', exclude=True)
@field_serializer('customer_email')defmask_email(self, email: str) -> str:
"""Mask email for privacy in exports"""if'@'in email:
local, domain = email.split('@')
masked = local[:2] + '***' + local[-1:] iflen(local) > 3else'***'returnf"{masked}@{domain}"return email
@field_serializer('created_at')defformat_datetime(self, dt: datetime) -> str:
"""Format datetime for export"""return dt.strftime('%Y-%m-%d %H:%M:%S')
@computed_field @propertydefdays_open(self) -> int:
"""Calculate days since ticket creation"""return (datetime.now() - self.created_at).days
# Serialization modes
ticket = TicketExport(
ticket_id=123,
customer_email='john.doe@example.com',
subject='Login issue',
created_at=datetime.now(),
status='open',
priority='high',
internal_notes='Customer called twice'
)
# Standard serializationprint(ticket.model_dump())
# {'ticket_id': 123, 'customer_email': 'jo***e@example.com', ...}# Include all fields (even excluded)print(ticket.model_dump(mode='python', exclude_none=False))
# Serialize to JSON
json_str = ticket.model_dump_json(indent=2)
print(json_str)
# Exclude specific fieldsprint(ticket.model_dump(exclude={'internal_notes', 'assigned_to'}))
# Include only specific fieldsprint(ticket.model_dump(include={'ticket_id', 'subject', 'status'}))
9. Advanced Validation Techniques
Purpose: Implement sophisticated validation logic for complex business requirements.
Pattern:
from pydantic import BaseModel, field_validator, model_validator
from typing importAny, Optionalimport re
classTicketPrioritization(BaseModel):
customer_tier: str
issue_category: str
response_time_hours: int
business_impact: str
affected_users: int = Field(ge=1)
@field_validator('customer_tier') @classmethoddefvalidate_tier(cls, v: str) -> str:
valid_tiers = {'free', 'basic', 'premium', 'enterprise'}
v = v.lower()
if v notin valid_tiers:
raise ValueError(f'Invalid tier. Must be one of: {valid_tiers}')
return v
@model_validator(mode='after')defcalculate_priority(self) -> 'TicketPrioritization':
"""Auto-calculate priority based on multiple factors"""
priority_score = 0# Customer tier weights
tier_weights = {'enterprise': 40, 'premium': 30, 'basic': 20, 'free': 10}
priority_score += tier_weights.get(self.customer_tier, 10)
# Business impact weights
impact_weights = {'critical': 30, 'high': 20, 'medium': 10, 'low': 5}
priority_score += impact_weights.get(self.business_impact.lower(), 5)
# Affected users factorifself.affected_users > 100:
priority_score += 20elifself.affected_users > 10:
priority_score += 10# Response time urgencyifself.response_time_hours <= 2:
priority_score += 10# Store computed priority (you'd add this field to the model)# self.computed_priority = 'urgent' if priority_score >= 80 else ...returnselfclassSecureTicketData(BaseModel):
"""Model with PII validation and sanitization"""
customer_name: str
email: str
phone: Optional[str] = None
credit_card_last4: Optional[str] = None
ssn_last4: Optional[str] = None @field_validator('credit_card_last4') @classmethoddefvalidate_cc_last4(cls, v: Optional[str]) -> Optional[str]:
if v isNone:
return v
ifnot re.match(r'^\d{4}$', v):
raise ValueError('Credit card last 4 must be exactly 4 digits')
return v
@field_validator('phone') @classmethoddefsanitize_phone(cls, v: Optional[str]) -> Optional[str]:
"""Remove all non-digit characters from phone"""if v isNone:
return v
digits_only = re.sub(r'\D', '', v)
iflen(digits_only) < 10:
raise ValueError('Phone number must have at least 10 digits')
return digits_only
@model_validator(mode='after')defvalidate_pii_consistency(self) -> 'SecureTicketData':
"""Ensure PII fields are consistent"""# If SSN is provided, credit card should also be provided for identity verificationifself.ssn_last4 andnotself.credit_card_last4:
raise ValueError('Credit card verification required when SSN is provided')
returnself
10. Performance Optimization with Pydantic V2
Purpose: Optimize validation performance for high-throughput support systems.
Key Strategies:
Use TypeAdapter for bulk validation:
from pydantic import TypeAdapter
from typing importList# Define adapter once, reuse for validation
ticket_list_adapter = TypeAdapter(List[SupportTicket])
# Fast bulk validation
tickets_data = [...] # List of dictionaries
validated_tickets = ticket_list_adapter.validate_python(tickets_data)
Leverage strict mode for performance:
from pydantic import BaseModel, ConfigDict
classFastTicket(BaseModel):
model_config = ConfigDict(strict=True) # No type coercion
ticket_id: int# Must be int, won't coerce from string
priority: str
Use discriminated unions for polymorphic data:
from typing importLiteral, Unionfrom pydantic import Field, BaseModel
classEmailTicket(BaseModel):
ticket_type: Literal['email'] = 'email'
customer_email: str
subject: strclassPhoneTicket(BaseModel):
ticket_type: Literal['phone'] = 'phone'
phone_number: str
call_duration: intclassChatTicket(BaseModel):
ticket_type: Literal['chat'] = 'chat'
chat_session_id: str# Fast dispatch based on discriminator field
TicketUnion = Union[EmailTicket, PhoneTicket, ChatTicket]
classTicketProcessor(BaseModel):
ticket: TicketUnion = Field(discriminator='ticket_type')
Reuse models and avoid dynamic creation:
# Good: Define once, reuseclassTicketModel(BaseModel):
ticket_id: int
subject: str# Avoid: Dynamic model creation in loopsfor data in ticket_data:
# Don't create models dynamicallypass
Error Handling Best Practices
Comprehensive Validation Error Handling
from pydantic import ValidationError
import logging
logger = logging.getLogger(__name__)
defprocess_ticket_submission(data: dict) -> Optional[SupportTicket]:
"""Process ticket with comprehensive error handling"""try:
ticket = SupportTicket(**data)
logger.info(f"Ticket {ticket.ticket_id} validated successfully")
return ticket
except ValidationError as e:
# Log detailed validation errors
logger.error(f"Validation failed for ticket submission: {e.error_count()} errors")
for error in e.errors():
field = '.'.join(str(loc) for loc in error['loc'])
error_type = error['type']
message = error['msg']
logger.error(f"Field '{field}': {message} (type: {error_type})")
# Return user-friendly error responsereturnNoneexcept Exception as e:
logger.exception(f"Unexpected error processing ticket: {str(e)}")
returnNone
Testing Pydantic Models
Unit Test Pattern
import pytest
from pydantic import ValidationError
deftest_support_ticket_validation():
"""Test ticket validation logic"""# Valid ticket
valid_data = {
'ticket_id': 123,
'customer_email': 'test@example.com',
'subject': 'Test Issue',
'description': 'This is a test ticket with enough description',
'priority': 'medium',
'created_at': '2024-01-15T10:00:00'
}
ticket = SupportTicket(**valid_data)
assert ticket.ticket_id == 123assert ticket.priority == 'medium'# Invalid prioritywith pytest.raises(ValidationError) as exc_info:
SupportTicket(**{**valid_data, 'priority': 'invalid'})
errors = exc_info.value.errors()
assertany(e['loc'] == ('priority',) for e in errors)
# Missing required fieldwith pytest.raises(ValidationError):
incomplete_data = {k: v for k, v in valid_data.items() if k != 'subject'}
SupportTicket(**incomplete_data)
Guidelines for Customer Support Context
Always validate user input: Never trust data from support forms, API calls, or external systems
Provide helpful error messages: Users should understand what's wrong and how to fix it
Log validation failures: Track patterns in validation errors to improve forms and documentation
Use appropriate field constraints: Balance security with usability
Implement business rule validation: Beyond types, validate business logic
Handle PII carefully: Mask or encrypt sensitive data, exclude from logs
Version your models: Use model versioning for API compatibility
Test edge cases: Include tests for boundary conditions and unusual inputs
Document models thoroughly: Use Field descriptions and JSON schema examples
Monitor validation performance: Track validation times for high-volume operations
Common Patterns and Anti-Patterns
Pattern: Request/Response Separation
# Good: Separate models for requests and responsesclassTicketCreateRequest(BaseModel):
subject: str
description: strclassTicketResponse(BaseModel):
ticket_id: int
subject: str
description: str
created_at: datetime
# Avoid: Using same model for input and output
Pattern: Default Factory for Mutable Defaults
# Good: Use default_factoryclassTicket(BaseModel):
tags: list[str] = Field(default_factory=list)
# Avoid: Mutable defaultclassBadTicket(BaseModel):
tags: list[str] = [] # Shared across instances!
