zhongwei/gh-greyhaven-ai-claude-code-config-grey-haven-plugins-data-quality
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
Zhongwei Li
2025-11-29 18:29:10 +08:00
commit 657f1e3da3
29 changed files with 2738 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
skills/data-validation/SKILL.md Normal file
View File

@@ -0,0 +1,25 @@
# Data Validation Skill
Comprehensive data validation using Pydantic v2 with data quality monitoring and schema alignment for PlanetScale PostgreSQL.
## Description
Implement robust data validation, quality monitoring, and schema contracts using Pydantic v2 models.
## What's Included
- **Examples**: Pydantic v2 models, validation patterns
- **Reference**: Schema design, validation strategies
- **Templates**: Pydantic model templates
## Use When
- API request/response validation
- Database schema alignment
- Data quality assurance
## Related Agents
- `data-validator`
**Skill Version**: 1.0

346
skills/data-validation/checklists/data-validation-checklist.md Normal file
View File

@@ -0,0 +1,346 @@
# Data Validation Checklist
Comprehensive checklist for implementing robust data validation in TypeScript (Zod) and Python (Pydantic) applications.
## Pre-Validation Setup
- [ ] **Identify all input sources** (API requests, forms, file uploads, external APIs)
- [ ] **Choose validation library** (Zod for TypeScript, Pydantic for Python)
- [ ] **Define validation strategy** (fail-fast vs collect all errors)
- [ ] **Set up error handling** (consistent error response format)
- [ ] **Document validation requirements** (business rules, constraints)
## TypeScript + Zod Validation
### Schema Definition
- [ ] **All API endpoints have Zod schemas** defined
- [ ] **Schema types exported** for use in frontend
- [ ] **Schemas colocated** with route handlers or in shared location
- [ ] **Schema composition used** (z.object, z.array, z.union)
- [ ] **Reusable schemas extracted** (common patterns like email, UUID)
### Basic Validations
- [ ] **String validations** applied:
- [ ] `.min()` for minimum length
- [ ] `.max()` for maximum length
- [ ] `.email()` for email addresses
- [ ] `.url()` for URLs
- [ ] `.uuid()` for UUIDs
- [ ] `.regex()` for custom patterns
- [ ] `.trim()` to remove whitespace
- [ ] **Number validations** applied:
- [ ] `.int()` for integers
- [ ] `.positive()` for positive numbers
- [ ] `.min()` / `.max()` for ranges
- [ ] `.finite()` to exclude Infinity/NaN
- [ ] **Array validations** applied:
- [ ] `.min()` for minimum items
- [ ] `.max()` for maximum items
- [ ] `.nonempty()` for required arrays
- [ ] **Date validations** applied:
- [ ] `.min()` for earliest date
- [ ] `.max()` for latest date
- [ ] Proper date parsing (z.coerce.date())
### Advanced Validations
- [ ] **Custom refinements** for complex rules:
```typescript
z.object({
password: z.string(),
confirmPassword: z.string()
}).refine(data => data.password === data.confirmPassword, {
message: "Passwords don't match",
path: ["confirmPassword"]
})
```
- [ ] **Conditional validations** with `.superRefine()`
- [ ] **Transform validations** to normalize data (`.transform()`)
- [ ] **Discriminated unions** for polymorphic data
- [ ] **Branded types** for domain-specific values
### Error Handling
- [ ] **Validation errors caught** and formatted consistently
- [ ] **Error messages user-friendly** (not technical jargon)
- [ ] **Field-level errors** returned (which field failed)
- [ ] **Multiple errors collected** (not just first error)
- [ ] **Error codes standardized** (e.g., "INVALID_EMAIL")
### Multi-Tenant Context
- [ ] **tenant_id validated** on all requests requiring tenant context
- [ ] **UUID format verified** for tenant_id
- [ ] **Tenant existence checked** (tenant must exist in database)
- [ ] **User-tenant relationship verified** (user belongs to tenant)
- [ ] **Admin permissions validated** for admin-only operations
## Python + Pydantic Validation
### Model Definition
- [ ] **All API request models** inherit from BaseModel
- [ ] **All response models** defined with Pydantic
- [ ] **SQLModel used** for database models (includes Pydantic)
- [ ] **Field validators** used for custom validation
- [ ] **Model validators** used for cross-field validation
### Basic Field Validators
- [ ] **EmailStr** used for email fields
- [ ] **HttpUrl** used for URL fields
- [ ] **UUID4** used for UUID fields
- [ ] **Field()** used with constraints:
- [ ] `min_length` / `max_length` for strings
- [ ] `ge` / `le` for number ranges (greater/less than or equal)
- [ ] `gt` / `lt` for strict ranges
- [ ] `regex` for pattern matching
```python
from pydantic import BaseModel, EmailStr, Field
class UserCreate(BaseModel):
email: EmailStr
name: str = Field(..., min_length=1, max_length=100)
age: int = Field(..., ge=0, le=150)
```
### Advanced Validation
- [ ] **@field_validator** for single field custom validation:
```python
@field_validator('password')
@classmethod
def validate_password(cls, v):
if len(v) < 8:
raise ValueError('Password must be at least 8 characters')
return v
```
- [ ] **@model_validator** for cross-field validation:
```python
@model_validator(mode='after')
def check_passwords_match(self):
if self.password != self.confirm_password:
raise ValueError('Passwords do not match')
return self
```
- [ ] **Custom validators** handle edge cases
- [ ] **Mode='before'** used for preprocessing
- [ ] **Mode='after'** used for post-validation checks
### Error Handling
- [ ] **ValidationError caught** in API endpoints
- [ ] **Errors formatted** to match frontend expectations
- [ ] **HTTPException raised** with 422 status for validation errors
- [ ] **Error details included** in response body
- [ ] **Logging added** for validation failures (security monitoring)
### Multi-Tenant Context
- [ ] **tenant_id field** on all multi-tenant request models
- [ ] **Tenant UUID validated** before database queries
- [ ] **Repository pattern enforces** tenant filtering
- [ ] **Admin flag validated** for privileged operations
- [ ] **RLS policies configured** on database tables
## Database Constraints
### Schema Constraints
- [ ] **NOT NULL constraints** on required fields
- [ ] **UNIQUE constraints** on unique fields (email, username)
- [ ] **CHECK constraints** for value ranges
- [ ] **FOREIGN KEY constraints** for relationships
- [ ] **Default values** defined where appropriate
### Index Support
- [ ] **Indexes created** on frequently queried fields
- [ ] **Composite indexes** for multi-field queries
- [ ] **Partial indexes** for filtered queries (WHERE clauses)
- [ ] **tenant_id indexed** on all multi-tenant tables
## File Upload Validation
- [ ] **File size limits** enforced (e.g., 10MB max)
- [ ] **File type validation** (MIME type checking)
- [ ] **File extension validation** (whitelist allowed extensions)
- [ ] **Virus scanning** (if handling untrusted uploads)
- [ ] **Content validation** (parse and validate file content)
### Image Uploads
- [ ] **Image dimensions validated** (max width/height)
- [ ] **Image format verified** (PNG, JPEG, etc.)
- [ ] **EXIF data stripped** (security concern)
- [ ] **Thumbnails generated** for large images
### CSV/JSON Uploads
- [ ] **Parse errors handled gracefully**
- [ ] **Schema validation** applied to each row/object
- [ ] **Batch validation** with error collection
- [ ] **Maximum rows/objects** limit enforced
## External API Integration
- [ ] **Response schemas defined** for external APIs
- [ ] **Validation applied** to external data
- [ ] **Graceful degradation** when validation fails
- [ ] **Retry logic** for transient failures
- [ ] **Timeout limits** configured
## Security Validations
### Input Sanitization
- [ ] **HTML/script tags stripped** from text inputs
- [ ] **SQL injection prevented** (use ORM, not raw SQL)
- [ ] **XSS prevention** (escape output in templates)
- [ ] **Path traversal prevented** (validate file paths)
- [ ] **Command injection prevented** (no shell execution of user input)
### Authentication & Authorization
- [ ] **JWT tokens validated** (signature, expiration)
- [ ] **Session tokens verified** against database
- [ ] **User existence checked** before operations
- [ ] **Permissions verified** for protected resources
- [ ] **Rate limiting applied** to prevent abuse
### Sensitive Data
- [ ] **Passwords never logged** or returned in responses
- [ ] **Credit card numbers validated** with Luhn algorithm
- [ ] **SSN/Tax ID formats validated**
- [ ] **PII handling compliant** with regulations (GDPR, CCPA)
- [ ] **Encryption applied** to sensitive stored data
## Testing Validation Logic
### Unit Tests
- [ ] **Valid inputs pass** validation
- [ ] **Invalid inputs fail** with correct error messages
- [ ] **Edge cases tested** (empty strings, null, undefined)
- [ ] **Boundary values tested** (min/max lengths, ranges)
- [ ] **Error messages verified** (correct field, message)
### Integration Tests
- [ ] **API endpoints validated** in integration tests
- [ ] **Database constraints tested** (violate constraint, expect error)
- [ ] **Multi-tenant isolation tested** (cross-tenant access blocked)
- [ ] **File upload validation tested**
- [ ] **External API mocking** with invalid responses
### Test Coverage
- [ ] **Validation logic 100% covered**
- [ ] **Error paths tested** (not just happy path)
- [ ] **Custom validators tested** independently
- [ ] **Refinements tested** with failing cases
## Performance Considerations
- [ ] **Validation performance measured** (avoid expensive validations in hot paths)
- [ ] **Async validation** for I/O-bound checks (database lookups)
- [ ] **Caching applied** to repeated validations (e.g., tenant existence)
- [ ] **Batch validation** for arrays/lists
- [ ] **Early returns** for fail-fast scenarios
## Documentation
- [ ] **Validation rules documented** in API docs
- [ ] **Error responses documented** (status codes, error formats)
- [ ] **Examples provided** (valid and invalid requests)
- [ ] **Schema exported** for frontend consumption (TypeScript types)
- [ ] **Changelog updated** when validation changes
## Grey Haven Specific
### TanStack Start (Frontend)
- [ ] **Form validation** with Zod + TanStack Form
- [ ] **Server function validation** (all server functions validate input)
- [ ] **Type safety** maintained (Zod.infer<> for types)
- [ ] **Error display** in UI components
- [ ] **Client-side validation** mirrors server-side
### FastAPI (Backend)
- [ ] **Request models** use Pydantic
- [ ] **Response models** use Pydantic
- [ ] **Repository methods** validate before database operations
- [ ] **Service layer** handles business rule validation
- [ ] **Dependency injection** for validation context (tenant_id)
### Database (Drizzle/SQLModel)
- [ ] **Drizzle schemas** include validation constraints
- [ ] **SQLModel fields** use Pydantic validators
- [ ] **Migration scripts** add database constraints
- [ ] **Indexes support** validation queries
## Monitoring & Alerting
- [ ] **Validation failure metrics** tracked
- [ ] **High failure rate alerts** configured
- [ ] **Unusual validation patterns** logged (potential attacks)
- [ ] **Performance metrics** for validation operations
- [ ] **Error logs** structured for analysis
## Scoring
- **80+ items checked**: Excellent - Comprehensive validation ✅
- **60-79 items**: Good - Most validation covered ⚠️
- **40-59 items**: Fair - Significant gaps exist 🔴
- **<40 items**: Poor - Inadequate validation ❌
## Priority Items
Address these first:
1. **All API endpoints validated** - Prevent invalid data entry
2. **Multi-tenant isolation** - Security-critical
3. **SQL injection prevention** - Use ORM, not raw SQL
4. **File upload validation** - Common attack vector
5. **Error handling** - User experience and debugging
## Common Pitfalls
❌ **Don't:**
- Trust client-side validation alone (always validate server-side)
- Use overly complex regex (hard to maintain, performance issues)
- Return technical error messages to users
- Skip validation on internal endpoints (defense in depth)
- Log sensitive data in validation errors
✅ **Do:**
- Validate at boundaries (API endpoints, file uploads, external APIs)
- Use standard validators (email, URL, UUID) from libraries
- Provide clear, actionable error messages
- Test validation logic thoroughly
- Document validation requirements
## Related Resources
- [Zod Documentation](https://zod.dev)
- [Pydantic Documentation](https://docs.pydantic.dev)
- [OWASP Input Validation](https://cheatsheetseries.owasp.org/cheatsheets/Input_Validation_Cheat_Sheet.html)
- [data-validation skill](../SKILL.md)
---
**Total Items**: 120+ validation checks
**Critical Items**: API validation, Multi-tenant, Security, File uploads
**Coverage**: TypeScript, Python, Database, Security
**Last Updated**: 2025-11-10

104
skills/data-validation/examples/INDEX.md Normal file
View File

@@ -0,0 +1,104 @@
# Data Validation Examples
Complete examples for Pydantic v2 validation, database schema alignment, and data quality monitoring.
## Examples Overview
### User Validation Example
**File**: [user-validation-example.md](user-validation-example.md)
Complete user validation workflow with:
- Pydantic v2 model with field validators
- Email, username, age validation
- Cross-field validation (admin age requirement)
- SQLModel database alignment
- FastAPI integration
- Validation error formatting
- Pytest test suite
**Use when**: Building user registration, profile management, or authentication.
---
### Order Validation Example
**File**: [order-validation-example.md](order-validation-example.md)
Complex order validation with:
- Nested object validation (order items, shipping address)
- Currency and pricing validation
- Inventory quantity checks
- Multi-tenant validation (tenant_id)
- Custom validators for business rules
- Database transaction validation
**Use when**: Building e-commerce, order management, or invoicing systems.
---
### Nested Validation Example
**File**: [nested-validation.md](nested-validation.md)
Advanced nested validation patterns:
- Nested Pydantic models
- List validation with min/max items
- Discriminated unions for polymorphic data
- Recursive validation (tree structures)
- Forward references
- Validation context passing
**Use when**: Working with complex JSON structures, API payloads, or hierarchical data.
---
## Quick Reference
| Example | Complexity | Key Patterns |
|---------|-----------|--------------|
| **User** | Basic | Field validators, cross-field validation |
| **Order** | Intermediate | Nested objects, business rules, transactions |
| **Nested** | Advanced | Discriminated unions, recursion, context |
## Common Patterns
### Basic Validation
```python
from pydantic import BaseModel, Field, EmailStr
class User(BaseModel):
email: EmailStr
age: int = Field(ge=13, le=120)
```
### Field Validator
```python
from pydantic import field_validator
class User(BaseModel):
username: str
@field_validator('username')
@classmethod
def validate_username(cls, v: str) -> str:
if len(v) < 3:
raise ValueError('Username too short')
return v
```
### Model Validator
```python
from pydantic import model_validator
class User(BaseModel):
role: str
age: int
@model_validator(mode='after')
def check_admin_age(self):
if self.role == 'admin' and self.age < 18:
raise ValueError('Admins must be 18+')
return self
```
---
Return to [main agent](../data-validator.md)

84
skills/data-validation/reference/INDEX.md Normal file
View File

@@ -0,0 +1,84 @@
# Data Validation Reference
Comprehensive reference documentation for Pydantic v2, SQLModel, and data quality patterns.
## Reference Materials
### Pydantic v2 Reference
**File**: [pydantic-v2-reference.md](pydantic-v2-reference.md)
Complete Pydantic v2 guide covering:
- Core concepts and migration from v1
- Field types and constraints (EmailStr, constr, conint, HttpUrl)
- Configuration with model_config
- Serialization modes (mode='json', mode='python')
- Error handling and custom error messages
- Performance optimization tips
**Use when**: Need comprehensive Pydantic v2 documentation or migrating from v1.
---
### Validators Reference
**File**: [validators-reference.md](validators-reference.md)
Complete guide to Pydantic validators:
- @field_validator for single-field validation
- @model_validator for cross-field validation
- Validator modes ('before', 'after', 'wrap')
- Accessing other field values with ValidationInfo
- Reusable validator functions
- Common validation patterns
**Use when**: Implementing custom validation logic beyond built-in constraints.
---
### SQLModel Alignment
**File**: [sqlmodel-alignment.md](sqlmodel-alignment.md)
Ensuring Pydantic schemas align with database models:
- Schema alignment validation patterns
- Type mapping (Pydantic ↔ SQLModel ↔ PostgreSQL)
- Multi-tenant patterns (tenant_id, RLS)
- Migration strategies
- Testing alignment
- Common misalignment issues
**Use when**: Building APIs with database persistence, ensuring data contracts match.
---
### Data Quality Monitoring
**File**: [data-quality-monitoring.md](data-quality-monitoring.md)
Monitoring data quality in production:
- Validation metrics tracking
- Error rate monitoring
- Data profiling patterns
- Alerting on validation failures
- Quality dashboards
- Integration with observability tools
**Use when**: Setting up production monitoring for data validation.
---
## Quick Reference
| Topic | Key Concepts | Common Use Cases |
|-------|-------------|------------------|
| **Pydantic v2** | Fields, validators, config | Request/response schemas |
| **Validators** | @field_validator, @model_validator | Custom business rules |
| **SQLModel Alignment** | Type mapping, migrations | Database persistence |
| **Data Quality** | Metrics, monitoring, alerts | Production reliability |
## Navigation
- **Examples**: [Examples Index](../examples/INDEX.md)
- **Templates**: [Templates Index](../templates/INDEX.md)
- **Main Agent**: [data-validator.md](../data-validator.md)
---
Return to [main agent](../data-validator.md)

66
skills/data-validation/templates/INDEX.md Normal file
View File

@@ -0,0 +1,66 @@
# Data Validation Templates
Copy-paste templates for common data validation patterns.
## Available Templates
### Pydantic Model Template
**File**: [pydantic-model.py](pydantic-model.py)
Complete Pydantic v2 model template with:
- Field definitions with constraints
- Custom validators (@field_validator, @model_validator)
- model_config configuration
- Nested models
- Documentation
**Use when**: Starting a new API request/response schema.
---
### SQLModel Template
**File**: [sqlmodel-model.py](sqlmodel-model.py)
Complete SQLModel database template with:
- Table configuration
- Field definitions with PostgreSQL types
- Multi-tenant pattern (tenant_id)
- Timestamps (created_at, updated_at)
- Indexes and constraints
- Relationships
**Use when**: Creating a new database table.
---
### FastAPI Endpoint Template
**File**: [fastapi-endpoint.py](fastapi-endpoint.py)
Complete FastAPI endpoint template with:
- Router configuration
- Pydantic request/response schemas
- Dependency injection (session, tenant_id, user_id)
- Validation error handling
- Database operations
- Multi-tenant isolation
**Use when**: Creating a new API endpoint with validation.
---
## Quick Start
1. **Copy template** to your project
2. **Rename** model/endpoint appropriately
3. **Customize** fields and validators
4. **Test** with comprehensive test cases
## Navigation
- **Examples**: [Examples Index](../examples/INDEX.md)
- **Reference**: [Reference Index](../reference/INDEX.md)
- **Main Agent**: [data-validator.md](../data-validator.md)
---
Return to [main agent](../data-validator.md)

311
skills/data-validation/templates/fastapi-endpoint.py Normal file
View File

@@ -0,0 +1,311 @@
"""
FastAPI Endpoint Template
Copy this template to create new API endpoints with validation.
Replace {ModelName}, {endpoint_prefix}, and {table_name} with your actual values.
"""
from fastapi import APIRouter, HTTPException, Depends, status
from sqlmodel import Session, select
from pydantic import ValidationError
from typing import List
from uuid import UUID
# Import your models and schemas
from app.models.{model_name} import {ModelName}
from app.schemas.{model_name} import (
{ModelName}CreateSchema,
{ModelName}UpdateSchema,
{ModelName}ResponseSchema
)
from app.database import get_session
from app.auth import get_current_user, get_current_tenant_id
# Create router
# -------------
router = APIRouter(
prefix="/api/{endpoint_prefix}",
tags=["{endpoint_prefix}"]
)
# Create Endpoint
# --------------
@router.post(
"/",
response_model={ModelName}ResponseSchema,
status_code=status.HTTP_201_CREATED,
summary="Create new {ModelName}",
description="Create a new {ModelName} with validation"
)
async def create_{model_name}(
data: {ModelName}CreateSchema,
session: Session = Depends(get_session),
user_id: UUID = Depends(get_current_user),
tenant_id: UUID = Depends(get_current_tenant_id)
):
"""
Create new {ModelName}.
Validates:
- Request data against Pydantic schema
- Business rules (if any)
- Uniqueness constraints
- Multi-tenant isolation
Returns:
{ModelName}ResponseSchema: Created {ModelName}
Raises:
HTTPException 409: If duplicate exists
HTTPException 422: If validation fails
"""
# Check for duplicates (if applicable)
existing = session.exec(
select({ModelName})
.where({ModelName}.email == data.email)
.where({ModelName}.tenant_id == tenant_id)
).first()
if existing:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="{ ModelName} with this email already exists"
)
# Create instance
instance = {ModelName}(
**data.model_dump(),
user_id=user_id,
tenant_id=tenant_id
)
session.add(instance)
session.commit()
session.refresh(instance)
return {ModelName}ResponseSchema.model_validate(instance)
# Read Endpoints
# -------------
@router.get(
"/{id}",
response_model={ModelName}ResponseSchema,
summary="Get {ModelName} by ID"
)
async def get_{model_name}(
id: UUID,
session: Session = Depends(get_session),
tenant_id: UUID = Depends(get_current_tenant_id)
):
"""
Get {ModelName} by ID with tenant isolation.
Returns:
{ModelName}ResponseSchema: Found {ModelName}
Raises:
HTTPException 404: If not found
"""
instance = session.exec(
select({ModelName})
.where({ModelName}.id == id)
.where({ModelName}.tenant_id == tenant_id)
).first()
if not instance:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="{ModelName} not found"
)
return {ModelName}ResponseSchema.model_validate(instance)
@router.get(
"/",
response_model=List[{ModelName}ResponseSchema],
summary="List {ModelName}s"
)
async def list_{model_name}s(
skip: int = 0,
limit: int = 100,
status: str | None = None,
session: Session = Depends(get_session),
tenant_id: UUID = Depends(get_current_tenant_id)
):
"""
List {ModelName}s with pagination and filtering.
Args:
skip: Number of records to skip (default 0)
limit: Maximum records to return (default 100, max 1000)
status: Filter by status (optional)
Returns:
List[{ModelName}ResponseSchema]: List of {ModelName}s
"""
# Build query
statement = (
select({ModelName})
.where({ModelName}.tenant_id == tenant_id)
.offset(skip)
.limit(min(limit, 1000))
)
# Apply filters
if status:
statement = statement.where({ModelName}.status == status)
# Execute
results = session.exec(statement).all()
return [
{ModelName}ResponseSchema.model_validate(item)
for item in results
]
# Update Endpoint
# --------------
@router.patch(
"/{id}",
response_model={ModelName}ResponseSchema,
summary="Update {ModelName}"
)
async def update_{model_name}(
id: UUID,
data: {ModelName}UpdateSchema,
session: Session = Depends(get_session),
tenant_id: UUID = Depends(get_current_tenant_id)
):
"""
Update {ModelName} with partial data.
Only provided fields are updated.
Returns:
{ModelName}ResponseSchema: Updated {ModelName}
Raises:
HTTPException 404: If not found
HTTPException 422: If validation fails
"""
# Get existing
instance = session.exec(
select({ModelName})
.where({ModelName}.id == id)
.where({ModelName}.tenant_id == tenant_id)
).first()
if not instance:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="{ModelName} not found"
)
# Update fields
update_data = data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(instance, field, value)
# Update timestamp
instance.updated_at = datetime.utcnow()
session.add(instance)
session.commit()
session.refresh(instance)
return {ModelName}ResponseSchema.model_validate(instance)
# Delete Endpoint
# --------------
@router.delete(
"/{id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete {ModelName}"
)
async def delete_{model_name}(
id: UUID,
session: Session = Depends(get_session),
tenant_id: UUID = Depends(get_current_tenant_id)
):
"""
Delete {ModelName}.
Soft delete by setting is_active = False.
For hard delete, use session.delete() instead.
Raises:
HTTPException 404: If not found
"""
instance = session.exec(
select({ModelName})
.where({ModelName}.id == id)
.where({ModelName}.tenant_id == tenant_id)
).first()
if not instance:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="{ModelName} not found"
)
# Soft delete
instance.is_active = False
instance.updated_at = datetime.utcnow()
session.add(instance)
session.commit()
# For hard delete:
# session.delete(instance)
# session.commit()
# Error Handling
# -------------
@router.exception_handler(ValidationError)
async def validation_exception_handler(request, exc: ValidationError):
"""Handle Pydantic validation errors."""
errors = {}
for error in exc.errors():
field = '.'.join(str(loc) for loc in error['loc'])
message = error['msg']
if field not in errors:
errors[field] = []
errors[field].append(message)
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={
'success': False,
'error': 'validation_error',
'message': 'Request validation failed',
'errors': errors
}
)
# Register Router
# --------------
# In your main FastAPI app:
# from app.api.{endpoint_prefix} import router as {model_name}_router
# app.include_router({model_name}_router)

231
skills/data-validation/templates/pydantic-model.py Normal file
View File

@@ -0,0 +1,231 @@
"""
Pydantic Model Template
Copy this template to create new Pydantic validation models.
Replace {ModelName}, {field_name}, and {FieldType} with your actual values.
"""
from pydantic import BaseModel, Field, field_validator, model_validator, EmailStr, HttpUrl
from typing import Optional, List, Literal
from datetime import datetime, date
from decimal import Decimal
from uuid import UUID
from enum import Enum
# Optional: Define enums for categorical fields
class {ModelName}Status(str, Enum):
"""Status enum for {ModelName}."""
ACTIVE = 'active'
INACTIVE = 'inactive'
PENDING = 'pending'
class {ModelName}CreateSchema(BaseModel):
"""
Schema for creating a {ModelName}.
This schema defines the data contract for API requests.
All validation rules are enforced here.
"""
# Required string field with length constraints
name: str = Field(
...,
min_length=1,
max_length=100,
description="Display name",
examples=["Example Name"]
)
# Email field with built-in validation
email: EmailStr = Field(
...,
description="Email address"
)
# Optional string field
description: Optional[str] = Field(
None,
max_length=500,
description="Optional description"
)
# Integer with range constraints
quantity: int = Field(
...,
ge=1,
le=999,
description="Quantity (1-999)"
)
# Decimal for currency (recommended over float)
price: Decimal = Field(
...,
gt=0,
max_digits=10,
decimal_places=2,
description="Price in USD"
)
# Date field
start_date: date = Field(
...,
description="Start date"
)
# Enum field
status: {ModelName}Status = Field(
default={ModelName}Status.PENDING,
description="Current status"
)
# List field with constraints
tags: List[str] = Field(
default_factory=list,
max_length=10,
description="Associated tags"
)
# Literal type (inline enum)
priority: Literal['low', 'medium', 'high'] = Field(
default='medium',
description="Priority level"
)
# Field Validators
# ---------------
@field_validator('name')
@classmethod
def validate_name(cls, v: str) -> str:
"""Validate name format."""
if not v.strip():
raise ValueError('Name cannot be empty')
return v.strip()
@field_validator('tags')
@classmethod
def validate_unique_tags(cls, v: List[str]) -> List[str]:
"""Ensure tags are unique."""
if len(v) != len(set(v)):
raise ValueError('Duplicate tags not allowed')
return [tag.lower() for tag in v]
# Model Validators (cross-field validation)
# ----------------------------------------
@model_validator(mode='after')
def validate_business_rules(self):
"""Validate business rules across fields."""
# Example: Ensure high-priority items have descriptions
if self.priority == 'high' and not self.description:
raise ValueError('High priority items must have description')
return self
# Configuration
# ------------
model_config = {
# Strip whitespace from strings
'str_strip_whitespace': True,
# Validate on assignment
'validate_assignment': True,
# JSON schema examples
'json_schema_extra': {
'examples': [{
'name': 'Example {ModelName}',
'email': 'example@company.com',
'description': 'An example description',
'quantity': 5,
'price': '99.99',
'start_date': '2024-01-01',
'status': 'active',
'tags': ['tag1', 'tag2'],
'priority': 'medium'
}]
}
}
class {ModelName}UpdateSchema(BaseModel):
"""
Schema for updating a {ModelName}.
All fields are optional for partial updates.
"""
name: Optional[str] = Field(None, min_length=1, max_length=100)
email: Optional[EmailStr] = None
description: Optional[str] = Field(None, max_length=500)
quantity: Optional[int] = Field(None, ge=1, le=999)
price: Optional[Decimal] = Field(None, gt=0, max_digits=10, decimal_places=2)
status: Optional[{ModelName}Status] = None
tags: Optional[List[str]] = Field(None, max_length=10)
priority: Optional[Literal['low', 'medium', 'high']] = None
model_config = {
'str_strip_whitespace': True,
'validate_assignment': True
}
class {ModelName}ResponseSchema(BaseModel):
"""
Schema for {ModelName} responses.
Includes all fields plus auto-generated ones (id, timestamps).
"""
id: UUID
name: str
email: str
description: Optional[str]
quantity: int
price: Decimal
start_date: date
status: str
tags: List[str]
priority: str
# Auto-generated fields
created_at: datetime
updated_at: datetime
model_config = {
# Enable ORM mode for SQLModel compatibility
'from_attributes': True
}
# Usage Example
# -------------
if __name__ == '__main__':
# Valid data
data = {
'name': 'Test Item',
'email': 'test@example.com',
'quantity': 10,
'price': '49.99',
'start_date': '2024-01-01',
'status': 'active',
'tags': ['electronics', 'featured']
}
# Create instance
item = {ModelName}CreateSchema(**data)
print(f"Created: {item.model_dump_json()}")
# Validation will raise errors for invalid data
try:
invalid_data = {**data, 'quantity': 0} # Invalid quantity
{ModelName}CreateSchema(**invalid_data)
except ValidationError as e:
print(f"Validation error: {e.errors()}")

246
skills/data-validation/templates/sqlmodel-model.py Normal file
View File

@@ -0,0 +1,246 @@
"""
SQLModel Database Model Template
Copy this template to create new database models.
Replace {ModelName} and {table_name} with your actual values.
"""
from sqlmodel import SQLModel, Field, Relationship
from datetime import datetime
from uuid import UUID, uuid4
from decimal import Decimal
from typing import Optional, List
from enum import Enum
# Optional: Define enum for status field
class {ModelName}Status(str, Enum):
"""Status enum for {ModelName}."""
ACTIVE = 'active'
INACTIVE = 'inactive'
PENDING = 'pending'
class {ModelName}(SQLModel, table=True):
"""
{ModelName} database model.
Represents {table_name} table in PostgreSQL.
Includes multi-tenant isolation via tenant_id.
"""
__tablename__ = '{table_name}'
# Primary Key
# -----------
id: UUID = Field(
default_factory=uuid4,
primary_key=True,
description="Unique identifier"
)
# Multi-Tenant Isolation (REQUIRED)
# ---------------------------------
tenant_id: UUID = Field(
foreign_key="tenants.id",
index=True,
description="Tenant for RLS isolation"
)
# Foreign Keys
# -----------
user_id: UUID = Field(
foreign_key="users.id",
index=True,
description="Owner user ID"
)
# Data Fields
# ----------
# String fields with length constraints
name: str = Field(
max_length=100,
index=True,
description="Display name"
)
email: str = Field(
max_length=255,
unique=True,
index=True,
description="Email address"
)
# Optional text field
description: Optional[str] = Field(
default=None,
max_length=500,
description="Optional description"
)
# Integer field
quantity: int = Field(
description="Quantity"
)
# Decimal for currency
price: Decimal = Field(
max_digits=10,
decimal_places=2,
description="Price in USD"
)
# Enum/Status field
status: str = Field(
default='pending',
max_length=20,
index=True,
description="Current status"
)
# Boolean field
is_active: bool = Field(
default=True,
description="Active flag"
)
# JSON field (stored as JSONB in PostgreSQL)
metadata: Optional[dict] = Field(
default=None,
sa_column_kwargs={"type_": "JSONB"},
description="Additional metadata"
)
# Timestamps (REQUIRED)
# --------------------
created_at: datetime = Field(
default_factory=datetime.utcnow,
nullable=False,
description="Creation timestamp"
)
updated_at: datetime = Field(
default_factory=datetime.utcnow,
nullable=False,
description="Last update timestamp"
)
# Relationships
# ------------
# One-to-many: This model has many related items
# items: List["RelatedItem"] = Relationship(back_populates="{model_name}")
# Many-to-one: This model belongs to a user
# user: Optional["User"] = Relationship(back_populates="{model_name}s")
# Related model example
# ---------------------
class {ModelName}Item(SQLModel, table=True):
"""Related items for {ModelName}."""
__tablename__ = '{table_name}_items'
id: UUID = Field(default_factory=uuid4, primary_key=True)
# Foreign key to parent
{model_name}_id: UUID = Field(
foreign_key="{table_name}.id",
index=True
)
# Item fields
name: str = Field(max_length=100)
quantity: int = Field(gt=0)
unit_price: Decimal = Field(max_digits=10, decimal_places=2)
created_at: datetime = Field(default_factory=datetime.utcnow)
# Relationship back to parent
{model_name}: Optional[{ModelName}] = Relationship(back_populates="items")
# Update parent model with relationship
{ModelName}.items = Relationship(back_populates="{model_name}")
# Database Migration
# ------------------
"""
After creating this model, generate a migration:
```bash
# Using Alembic
alembic revision --autogenerate -m "create {table_name} table"
alembic upgrade head
```
Migration will create:
- Table {table_name}
- Indexes on tenant_id, user_id, name, email, status
- Unique constraint on email
- Foreign key constraints
- Timestamps with defaults
"""
# Row-Level Security (RLS)
# ------------------------
"""
Enable RLS for multi-tenant isolation:
```sql
-- Enable RLS
ALTER TABLE {table_name} ENABLE ROW LEVEL SECURITY;
-- Create policy
CREATE POLICY tenant_isolation ON {table_name}
USING (tenant_id = current_setting('app.tenant_id')::UUID);
-- Grant access
GRANT SELECT, INSERT, UPDATE, DELETE ON {table_name} TO app_user;
```
"""
# Usage Example
# -------------
if __name__ == '__main__':
from sqlmodel import Session, create_engine, select
# Create engine
engine = create_engine('postgresql://user:pass@localhost/db')
# Create tables
SQLModel.metadata.create_all(engine)
# Insert data
with Session(engine) as session:
item = {ModelName}(
tenant_id=uuid4(),
user_id=uuid4(),
name='Test Item',
email='test@example.com',
quantity=10,
price=Decimal('49.99'),
status='active'
)
session.add(item)
session.commit()
session.refresh(item)
print(f"Created {ModelName}: {item.id}")
# Query data
with Session(engine) as session:
statement = select({ModelName}).where({ModelName}.status == 'active')
results = session.exec(statement).all()
print(f"Found {len(results)} active items")