src/app/schemas/README.md

API Schemas Documentation

This directory contains Pydantic schemas for the FastAPI video generation backend. The schemas are organized into three main categories:

Structure

• requests.py - Request validation schemas for API endpoints
• responses.py - Response formatting schemas for API endpoints
• common.py - Shared schemas for pagination, filtering, and error handling
• examples.py - Comprehensive examples for API documentation
• __init__.py - Module exports and imports

Request Schemas

VideoGenerationRequest

Primary schema for video generation requests with comprehensive validation:

from app.schemas.requests import VideoGenerationRequest

request = VideoGenerationRequest(
    topic="Machine Learning Basics",
    context="Comprehensive introduction to ML concepts...",
    quality="high",
    use_rag=True,
    tags=["ml", "education"]
)

Key Features:

• Field validation with custom validators
• Comprehensive examples for API documentation
• Support for advanced configuration options
• Automatic data cleaning and normalization

BatchVideoGenerationRequest

Schema for creating multiple video generation jobs:

from app.schemas.requests import BatchVideoGenerationRequest

batch_request = BatchVideoGenerationRequest(
    jobs=[request1, request2, request3],
    batch_priority="high",
    batch_name="Educational Series"
)

JobFilterRequest

Schema for filtering and searching jobs:

from app.schemas.requests import JobFilterRequest

filters = JobFilterRequest(
    status=["processing", "completed"],
    search="mathematics",
    created_after="2024-01-01T00:00:00Z"
)

Response Schemas

JobResponse

Standard response for job operations:

from app.schemas.responses import JobResponse

response = JobResponse(
    job_id="550e8400-e29b-41d4-a716-446655440000",
    status="queued",
    priority="normal",
    progress=JobProgress(percentage=0.0),
    created_at=datetime.now()
)

JobStatusResponse

Detailed job status information:

from app.schemas.responses import JobStatusResponse

status_response = JobStatusResponse(
    job_id="550e8400-e29b-41d4-a716-446655440000",
    user_id="user_123",
    status="processing",
    progress=JobProgress(
        percentage=45.0,
        current_stage="video_generation"
    ),
    metrics=JobMetrics(
        processing_time_seconds=120.5,
        cpu_usage_percent=75.0
    )
)

VideoDownloadResponse

Complete video information with download links:

from app.schemas.responses import VideoDownloadResponse

video_response = VideoDownloadResponse(
    video_id="video_123",
    job_id="job_123",
    filename="educational_video.mp4",
    file_size=25165824,
    file_size_mb=24.0,
    download_url="https://api.example.com/download/video_123",
    thumbnail_urls=["thumb1.jpg", "thumb2.jpg"]
)

Common Schemas

PaginationRequest

Standardized pagination parameters:

from app.schemas.common import PaginationRequest

pagination = PaginationRequest(
    page=1,
    items_per_page=20
)

# Access calculated values
offset = pagination.offset  # 0
limit = pagination.limit    # 20

SortRequest

Sorting parameters with validation:

from app.schemas.common import SortRequest

sort = SortRequest(
    sort_by="created_at",
    sort_order="desc"
)

ErrorResponse

Consistent error formatting:

from app.schemas.common import ErrorResponse
from app.models.common import ErrorDetail, ErrorCode

error = ErrorResponse(
    message="Job not found",
    error=ErrorDetail(
        message="The requested job was not found",
        code=ErrorCode.NOT_FOUND,
        details={"job_id": "123"}
    )
)

Usage in FastAPI Endpoints

Basic Endpoint with Request/Response Schemas

from fastapi import APIRouter, HTTPException
from app.schemas.requests import VideoGenerationRequest
from app.schemas.responses import JobResponse
from app.schemas.common import ErrorResponse

router = APIRouter()

@router.post(
    "/videos/generate",
    response_model=JobResponse,
    responses={
        422: {"model": ValidationErrorResponse},
        401: {"model": UnauthorizedResponse},
        500: {"model": InternalServerErrorResponse}
    }
)
async def generate_video(request: VideoGenerationRequest):
    """
    Generate a new educational video.
    
    Creates a new video generation job with the provided configuration.
    Returns job information including ID and initial status.
    """
    # Business logic here
    pass

List Endpoint with Pagination and Filtering

from app.schemas.common import PaginationRequest, SortRequest
from app.schemas.requests import JobFilterRequest
from app.schemas.responses import JobListResponse

@router.get(
    "/jobs",
    response_model=JobListResponse
)
async def list_jobs(
    pagination: PaginationRequest = Depends(),
    sort: SortRequest = Depends(),
    filters: JobFilterRequest = Depends()
):
    """
    List jobs with pagination, sorting, and filtering.
    
    Supports comprehensive filtering by status, priority, date ranges,
    and text search across job topics and context.
    """
    # Implementation here
    pass

Error Handling with Consistent Responses

from app.schemas.common import NotFoundResponse, ValidationErrorResponse

@router.get(
    "/jobs/{job_id}",
    response_model=JobStatusResponse,
    responses={
        404: {"model": NotFoundResponse},
        422: {"model": ValidationErrorResponse}
    }
)
async def get_job_status(job_id: str):
    """Get detailed job status information."""
    if not job_exists(job_id):
        raise HTTPException(
            status_code=404,
            detail=NotFoundResponse(
                resource="Job",
                resource_id=job_id
            ).dict()
        )

Validation Features

Custom Validators

The schemas include comprehensive validation:

# Topic validation
@validator("topic")
def validate_topic(cls, v):
    if len(v.strip()) < 3:
        raise ValueError("Topic must be at least 3 characters long")
    return v.strip()

# Date range validation
@validator("created_before")
def validate_date_range(cls, v, values):
    created_after = values.get("created_after")
    if created_after and v and v <= created_after:
        raise ValueError("created_before must be after created_after")
    return v

Field Constraints

• String length limits with min_length and max_length
• Numeric ranges with ge, le, gt, lt
• Pattern matching with pattern for formats
• List size limits with min_items and max_items
• Custom validation logic with @validator

API Documentation Examples

The schemas include comprehensive examples for automatic API documentation:

model_config = ConfigDict(
    json_schema_extra={
        "example": {
            "topic": "Pythagorean Theorem",
            "context": "Explain the mathematical proof...",
            "quality": "medium",
            "use_rag": True
        }
    }
)

These examples are automatically used by FastAPI to generate:

• Interactive Swagger UI documentation
• ReDoc documentation
• OpenAPI specification with examples
• Client code generation with proper examples

Best Practices

1. Consistent Naming: Use clear, descriptive field names
2. Comprehensive Validation: Include both type and business logic validation
3. Rich Examples: Provide realistic examples for documentation
4. Error Details: Include specific error codes and field-level errors
5. Backward Compatibility: Use optional fields for new features
6. Performance: Keep validation efficient for high-throughput endpoints

Testing Schemas

def test_video_generation_request():
    # Valid request
    request = VideoGenerationRequest(
        topic="Test Topic",
        context="This is a valid context with sufficient length"
    )
    assert request.topic == "Test Topic"
    
    # Invalid request should raise ValidationError
    with pytest.raises(ValidationError):
        VideoGenerationRequest(
            topic="",  # Empty topic should fail
            context="short"  # Too short context should fail
        )

This schema system provides a robust foundation for API validation, documentation, and client generation while maintaining type safety and comprehensive error handling.