Skip to main content

project-pdf-platform-technical-design

title: 'Technical Design Document: AI-Powered PDF Analysis Platform' type: reference component_type: reference version: 1.0.0 created: '2025-12-27' updated: '2025-12-27' status: active tags:

  • ai-ml
  • authentication
  • security
  • testing
  • api
  • architecture
  • automation
  • backend summary: 'Technical Design Document: AI-Powered PDF Analysis Platform --- Technical Architecture 1.1 System Component Diagram --- API Specifications 2.1 REST API Endpoints' moe_confidence: 0.950 moe_classified: 2025-12-31

Technical Design Document: AI-Powered PDF Analysis Platform

Version: 1.0
Date: 2025-10-31
Status: APPROVED

1. Technical Architecture​

1.1 System Component Diagram​

2. API Specifications​

2.1 REST API Endpoints​

Upload PDF​

POST /api/v1/documents/upload
Content-Type: multipart/form-data
Authorization: Bearer {token}

{
  "file": <binary>,
  "options": {
    "analysis_type": "full",
    "extract_tables": true,
    "extract_images": false
  }
}

Response 201:
{
  "document_id": "uuid",
  "status": "processing",
  "upload_time": "2025-10-31T10:00:00Z",
  "websocket_channel": "document:uuid"
}

Get Document Status​

GET /api/v1/documents/{document_id}
Authorization: Bearer {token}

Response 200:
{
  "document_id": "uuid",
  "filename": "report.pdf",
  "status": "completed",
  "progress": 100,
  "size_bytes": 1048576,
  "uploaded_at": "2025-10-31T10:00:00Z",
  "processed_at": "2025-10-31T10:01:30Z",
  "results": {
    "text_extraction": "uuid",
    "ai_analysis": "uuid"
  }
}

List Documents​

GET /api/v1/documents?page=1&limit=20&status=completed
Authorization: Bearer {token}

Response 200:
{
  "documents": [...],
  "total": 150,
  "page": 1,
  "limit": 20,
  "has_next": true
}

Get Analysis Results​

GET /api/v1/documents/{document_id}/analysis
Authorization: Bearer {token}

Response 200:
{
  "document_id": "uuid",
  "analysis": {
    "summary": "...",
    "components": [...],
    "confidence_score": 0.95,
    "validation_status": "verified"
  }
}

2.2 WebSocket Protocol​

Connection​

// Client connects
ws://api.example.com/ws?token={jwt_token}

// Server acknowledges
{
  "type": "connection_ack",
  "client_id": "uuid",
  "timestamp": "2025-10-31T10:00:00Z"
}

Subscribe to Document Updates​

// Client subscribes
{
  "type": "subscribe",
  "channel": "document:uuid"
}

// Server confirms
{
  "type": "subscribed",
  "channel": "document:uuid"
}

Processing Events​

// Upload started
{
  "type": "document.upload.started",
  "document_id": "uuid",
  "timestamp": "2025-10-31T10:00:00Z"
}

// Processing progress
{
  "type": "document.processing.progress",
  "document_id": "uuid",
  "stage": "text_extraction",
  "progress": 45,
  "timestamp": "2025-10-31T10:00:15Z"
}

// AI analysis started
{
  "type": "document.analysis.started",
  "document_id": "uuid",
  "analysis_type": "full",
  "timestamp": "2025-10-31T10:00:30Z"
}

// Analysis completed
{
  "type": "document.analysis.completed",
  "document_id": "uuid",
  "results_url": "/api/v1/documents/uuid/analysis",
  "timestamp": "2025-10-31T10:01:30Z"
}

// Error occurred
{
  "type": "document.error",
  "document_id": "uuid",
  "error": {
    "code": "PROCESSING_FAILED",
    "message": "Unable to extract text from page 5"
  },
  "timestamp": "2025-10-31T10:00:45Z"
}

3. Data Models​

3.1 Database Schema​

-- Users table
CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    email VARCHAR(255) UNIQUE NOT NULL,
    name VARCHAR(255) NOT NULL,
    preferences JSONB DEFAULT '{}',
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

-- Documents table
CREATE TABLE documents (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    filename VARCHAR(500) NOT NULL,
    gcs_path VARCHAR(1000) NOT NULL,
    size_bytes BIGINT NOT NULL,
    mime_type VARCHAR(100) NOT NULL,
    status VARCHAR(50) NOT NULL, -- uploaded, processing, completed, failed
    metadata JSONB DEFAULT '{}',
    uploaded_at TIMESTAMP DEFAULT NOW(),
    processed_at TIMESTAMP,
    INDEX idx_user_id (user_id),
    INDEX idx_status (status),
    INDEX idx_uploaded_at (uploaded_at)
);

-- Processing jobs table
CREATE TABLE processing_jobs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    document_id UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
    job_type VARCHAR(50) NOT NULL, -- text_extraction, table_extraction, ai_analysis
    status VARCHAR(50) NOT NULL, -- pending, running, completed, failed
    config JSONB DEFAULT '{}',
    error_message TEXT,
    started_at TIMESTAMP,
    completed_at TIMESTAMP,
    INDEX idx_document_id (document_id),
    INDEX idx_status (status)
);

-- Analysis results table
CREATE TABLE analysis_results (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    job_id UUID NOT NULL REFERENCES processing_jobs(id) ON DELETE CASCADE,
    result_type VARCHAR(50) NOT NULL, -- summary, component, validation
    content JSONB NOT NULL,
    confidence_score FLOAT,
    created_at TIMESTAMP DEFAULT NOW(),
    INDEX idx_job_id (job_id)
);

-- Components table
CREATE TABLE components (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    result_id UUID NOT NULL REFERENCES analysis_results(id) ON DELETE CASCADE,
    component_type VARCHAR(50) NOT NULL, -- text_block, table, figure, header
    content TEXT NOT NULL,
    metadata JSONB DEFAULT '{}',
    page_number INT,
    bounding_box JSONB, -- {x, y, width, height}
    created_at TIMESTAMP DEFAULT NOW(),
    INDEX idx_result_id (result_id),
    INDEX idx_component_type (component_type)
);

-- Audit log
CREATE TABLE audit_log (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id UUID REFERENCES users(id),
    action VARCHAR(100) NOT NULL,
    resource_type VARCHAR(50) NOT NULL,
    resource_id UUID,
    metadata JSONB DEFAULT '{}',
    created_at TIMESTAMP DEFAULT NOW(),
    INDEX idx_user_id (user_id),
    INDEX idx_created_at (created_at)
);

3.2 TypeScript Interfaces​

// Frontend data models

export interface User {
  id: string;
  email: string;
  name: string;
  preferences: Record<string, unknown>;
  createdAt: string;
}

export interface Document {
  id: string;
  userId: string;
  filename: string;
  gcsPath: string;
  sizeBytes: number;
  mimeType: string;
  status: DocumentStatus;
  metadata: DocumentMetadata;
  uploadedAt: string;
  processedAt?: string;
}

export type DocumentStatus = 
  | 'uploaded' 
  | 'processing' 
  | 'completed' 
  | 'failed';

export interface DocumentMetadata {
  pageCount?: number;
  author?: string;
  createdDate?: string;
  title?: string;
}

export interface ProcessingJob {
  id: string;
  documentId: string;
  jobType: JobType;
  status: JobStatus;
  config: Record<string, unknown>;
  errorMessage?: string;
  startedAt?: string;
  completedAt?: string;
}

export type JobType = 
  | 'text_extraction' 
  | 'table_extraction' 
  | 'ai_analysis';

export type JobStatus = 
  | 'pending' 
  | 'running' 
  | 'completed' 
  | 'failed';

export interface AnalysisResult {
  id: string;
  jobId: string;
  resultType: ResultType;
  content: unknown;
  confidenceScore?: number;
  createdAt: string;
}

export type ResultType = 
  | 'summary' 
  | 'component' 
  | 'validation';

export interface Component {
  id: string;
  resultId: string;
  componentType: ComponentType;
  content: string;
  metadata: ComponentMetadata;
  pageNumber?: number;
  boundingBox?: BoundingBox;
  createdAt: string;
}

export type ComponentType = 
  | 'text_block' 
  | 'table' 
  | 'figure' 
  | 'header' 
  | 'footer' 
  | 'list';

export interface ComponentMetadata {
  fontFamily?: string;
  fontSize?: number;
  color?: string;
  style?: string[];
}

export interface BoundingBox {
  x: number;
  y: number;
  width: number;
  height: number;
}

export interface WebSocketMessage {
  type: string;
  documentId?: string;
  timestamp: string;
  data?: unknown;
}

4. AI Integration Architecture​

4.1 Prompt Engineering Framework​

from typing import Protocol, Dict, Any, List
from dataclasses import dataclass
from enum import Enum

class AnalysisMode(Enum):
    STRUCTURE = "structure"
    EXTRACTION = "extraction"
    VALIDATION = "validation"
    SYNTHESIS = "synthesis"

@dataclass
class PromptTemplate:
    """Template for LLM prompts"""
    mode: AnalysisMode
    system_prompt: str
    user_prompt_template: str
    expected_output_schema: Dict[str, Any]
    max_tokens: int
    temperature: float

class PromptRegistry:
    """Registry of prompt templates"""
    
    STRUCTURE_ANALYSIS = PromptTemplate(
        mode=AnalysisMode.STRUCTURE,
        system_prompt="""You are an expert document analyst. Analyze the structure 
        of the provided PDF content and identify key sections, headings, and 
        organizational patterns.""",
        user_prompt_template="""Analyze this document content:

{content}

Return a JSON response with this structure:
{{
  "document_type": "string",
  "sections": [
    {{
      "title": "string",
      "level": number,
      "page_start": number,
      "page_end": number
    }}
  ],
  "has_table_of_contents": boolean,
  "has_appendices": boolean
}}""",
        expected_output_schema={
            "document_type": "string",
            "sections": "array",
            "has_table_of_contents": "boolean",
            "has_appendices": "boolean"
        },
        max_tokens=2000,
        temperature=0.3
    )
    
    COMPONENT_EXTRACTION = PromptTemplate(
        mode=AnalysisMode.EXTRACTION,
        system_prompt="""You are an expert at extracting structured components 
        from documents. Identify and extract tables, figures, code blocks, and 
        other structured elements with high precision.""",
        user_prompt_template="""Extract all structured components from this content:

{content}

For each component, provide:
- Type (table, figure, code_block, equation, list)
- Content or description
- Page number
- Bounding box if available
- Relationships to other components

Return JSON array of components.""",
        expected_output_schema={
            "components": [
                {
                    "id": "string",
                    "type": "string",
                    "content": "string",
                    "page_number": "number",
                    "metadata": "object"
                }
            ]
        },
        max_tokens=4000,
        temperature=0.2
    )
    
    CROSS_VALIDATION = PromptTemplate(
        mode=AnalysisMode.VALIDATION,
        system_prompt="""You are a meticulous validator. Cross-check extracted 
        data for consistency, accuracy, and completeness. Flag any 
        discrepancies or low-confidence extractions.""",
        user_prompt_template="""Validate this extracted data:

Original content:
{original_content}

Extracted data:
{extracted_data}

Check for:
1. Accuracy of text extraction
2. Completeness of tables
3. Correct identification of component types
4. Logical consistency
5. Missing or corrupted data

Return validation report with confidence scores.""",
        expected_output_schema={
            "validation_passed": "boolean",
            "confidence_score": "number",
            "issues": "array",
            "recommendations": "array"
        },
        max_tokens=3000,
        temperature=0.1
    )

class AIService:
    """AI service for PDF analysis"""
    
    def __init__(self, anthropic_client, prompt_registry: PromptRegistry):
        self.client = anthropic_client
        self.registry = prompt_registry
    
    async def analyze_structure(
        self, 
        content: str
    ) -> Dict[str, Any]:
        """Analyze document structure"""
        template = self.registry.STRUCTURE_ANALYSIS
        
        response = await self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=template.max_tokens,
            temperature=template.temperature,
            system=template.system_prompt,
            messages=[
                {
                    "role": "user",
                    "content": template.user_prompt_template.format(
                        content=content
                    )
                }
            ]
        )
        
        return self._parse_response(response)
    
    async def extract_components(
        self, 
        content: str,
        pages: List[str]
    ) -> List[Dict[str, Any]]:
        """Extract structured components"""
        template = self.registry.COMPONENT_EXTRACTION
        
        components = []
        
        # Process in chunks to respect token limits
        for page_num, page_content in enumerate(pages, 1):
            response = await self.client.messages.create(
                model="claude-sonnet-4-20250514",
                max_tokens=template.max_tokens,
                temperature=template.temperature,
                system=template.system_prompt,
                messages=[
                    {
                        "role": "user",
                        "content": template.user_prompt_template.format(
                            content=page_content
                        )
                    }
                ]
            )
            
            page_components = self._parse_response(response)
            
            # Add page number context
            for comp in page_components.get("components", []):
                comp["page_number"] = page_num
            
            components.extend(page_components.get("components", []))
        
        return components
    
    async def validate_extraction(
        self,
        original_content: str,
        extracted_data: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Cross-validate extracted data"""
        template = self.registry.CROSS_VALIDATION
        
        response = await self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=template.max_tokens,
            temperature=template.temperature,
            system=template.system_prompt,
            messages=[
                {
                    "role": "user",
                    "content": template.user_prompt_template.format(
                        original_content=original_content[:10000],  # Truncate
                        extracted_data=json.dumps(extracted_data, indent=2)
                    )
                }
            ]
        )
        
        return self._parse_response(response)
    
    def _parse_response(self, response) -> Dict[str, Any]:
        """Parse Claude API response"""
        content = response.content[0].text
        
        # Extract JSON from response
        try:
            # Remove markdown code blocks if present
            if "```json" in content:
                content = content.split("```json")[1].split("```")[0]
            elif "```" in content:
                content = content.split("```")[1].split("```")[0]
            
            return json.loads(content.strip())
        except json.JSONDecodeError as e:
            raise ValueError(f"Failed to parse AI response: {e}")

4.2 Token Management Strategy​

from dataclasses import dataclass
from typing import Optional

@dataclass
class TokenBudget:
    """Manage token usage across AI calls"""
    total_budget: int = 100_000
    structure_analysis: int = 2_000
    component_extraction_per_page: int = 4_000
    validation: int = 3_000
    synthesis: int = 5_000
    
    def calculate_page_budget(self, page_count: int) -> int:
        """Calculate budget for multi-page document"""
        extraction_budget = page_count * self.component_extraction_per_page
        
        return (
            self.structure_analysis +
            extraction_budget +
            self.validation +
            self.synthesis
        )
    
    def can_process(self, page_count: int) -> bool:
        """Check if document can be processed within budget"""
        required = self.calculate_page_budget(page_count)
        return required <= self.total_budget
    
    def recommend_strategy(
        self, 
        page_count: int
    ) -> Dict[str, Any]:
        """Recommend processing strategy based on page count"""
        if page_count <= 10:
            return {
                "strategy": "full_analysis",
                "description": "Process all pages with full AI analysis"
            }
        elif page_count <= 50:
            return {
                "strategy": "selective_analysis",
                "description": "Analyze key pages, summarize others"
            }
        else:
            return {
                "strategy": "batch_processing",
                "description": "Process in batches with checkpoint recovery"
            }

5. Frontend Architecture​

5.1 React Component Structure​

// app.tsx - Main application
src/
├── app.tsx
├── components/
│   ├── layout/
│   │   ├── AppShell.tsx
│   │   ├── header.tsx
│   │   ├── Sidebar.tsx
│   │   └── footer.tsx
│   ├── upload/
│   │   ├── UploadZone.tsx
│   │   ├── FileInput.tsx
│   │   └── UploadProgress.tsx
│   ├── documents/
│   │   ├── document-list.tsx
│   │   ├── DocumentCard.tsx
│   │   ├── DocumentGrid.tsx
│   │   └── DocumentFilters.tsx
│   ├── analysis/
│   │   ├── AnalysisView.tsx
│   │   ├── ComponentViewer.tsx
│   │   ├── SummaryPanel.tsx
│   │   └── ValidationStatus.tsx
│   └── shared/
│       ├── Button.tsx
│       ├── Modal.tsx
│       ├── Toast.tsx
│       └── Loader.tsx
├── hooks/
│   ├── use-web-socket.ts
│   ├── useDocuments.ts
│   ├── useUpload.ts
│   └── useAnalysis.ts
├── services/
│   ├── api.ts
│   ├── websocket.ts
│   ├── auth.ts
│   └── storage.ts
├── store/
│   ├── document-store.ts
│   ├── uiStore.ts
│   └── auth-store.ts
├── types/
│   ├── document.ts
│   ├── analysis.ts
│   └── api.ts
└── utils/
    ├── formatters.ts
    ├── validators.ts
    └── constants.ts

5.2 State Management with Zustand​

// store/document-store.ts
import { create } from 'zustand';
import { Document, ProcessingJob } from '../types/document';

interface DocumentState {
  documents: Map<string, Document>;
  jobs: Map<string, ProcessingJob>;
  selectedDocumentId: string | null;
  
  // Actions
  addDocument: (doc: Document) => void;
  updateDocument: (id: string, updates: Partial<Document>) => void;
  deleteDocument: (id: string) => void;
  setSelectedDocument: (id: string | null) => void;
  
  addJob: (job: ProcessingJob) => void;
  updateJob: (id: string, updates: Partial<ProcessingJob>) => void;
}

export const useDocumentStore = create<DocumentState>((set) => ({
  documents: new Map(),
  jobs: new Map(),
  selectedDocumentId: null,
  
  addDocument: (doc) => set((state) => {
    const newDocs = new Map(state.documents);
    newDocs.set(doc.id, doc);
    return { documents: newDocs };
  }),
  
  updateDocument: (id, updates) => set((state) => {
    const newDocs = new Map(state.documents);
    const existing = newDocs.get(id);
    if (existing) {
      newDocs.set(id, { ...existing, ...updates });
    }
    return { documents: newDocs };
  }),
  
  deleteDocument: (id) => set((state) => {
    const newDocs = new Map(state.documents);
    newDocs.delete(id);
    return { documents: newDocs };
  }),
  
  setSelectedDocument: (id) => set({ selectedDocumentId: id }),
  
  addJob: (job) => set((state) => {
    const newJobs = new Map(state.jobs);
    newJobs.set(job.id, job);
    return { jobs: newJobs };
  }),
  
  updateJob: (id, updates) => set((state) => {
    const newJobs = new Map(state.jobs);
    const existing = newJobs.get(id);
    if (existing) {
      newJobs.set(id, { ...existing, ...updates });
    }
    return { jobs: newJobs };
  }),
}));

5.3 WebSocket Hook​

// hooks/use-web-socket.ts
import { useEffect, useRef, useCallback } from 'react';
import { useDocumentStore } from '../store/documentStore';
import { WebSocketMessage } from '../types/api';

export function useWebSocket(url: string, token: string) {
  const wsRef = useRef<WebSocket | null>(null);
  const updateDocument = useDocumentStore((s) => s.updateDocument);
  const updateJob = useDocumentStore((s) => s.updateJob);
  
  const connect = useCallback(() => {
    const ws = new WebSocket(`${url}?token=${token}`);
    
    ws.onopen = () => {
      console.log('WebSocket connected');
    };
    
    ws.onmessage = (event) => {
      const message: WebSocketMessage = JSON.parse(event.data);
      
      switch (message.type) {
        case 'document.upload.started':
          updateDocument(message.documentId!, { status: 'processing' });
          break;
          
        case 'document.processing.progress':
          updateJob(message.data.jobId, {
            status: 'running',
            ...message.data
          });
          break;
          
        case 'document.analysis.completed':
          updateDocument(message.documentId!, {
            status: 'completed',
            processedAt: message.timestamp
          });
          break;
          
        case 'document.error':
          updateDocument(message.documentId!, {
            status: 'failed'
          });
          break;
      }
    };
    
    ws.onerror = (error) => {
      console.error('WebSocket error:', error);
    };
    
    ws.onclose = () => {
      console.log('WebSocket disconnected');
      // Attempt reconnection after 3 seconds
      setTimeout(() => connect(), 3000);
    };
    
    wsRef.current = ws;
  }, [url, token, updateDocument, updateJob]);
  
  useEffect(() => {
    connect();
    
    return () => {
      wsRef.current?.close();
    };
  }, [connect]);
  
  const subscribe = useCallback((channel: string) => {
    if (wsRef.current?.readyState === WebSocket.OPEN) {
      wsRef.current.send(JSON.stringify({
        type: 'subscribe',
        channel
      }));
    }
  }, []);
  
  const unsubscribe = useCallback((channel: string) => {
    if (wsRef.current?.readyState === WebSocket.OPEN) {
      wsRef.current.send(JSON.stringify({
        type: 'unsubscribe',
        channel
      }));
    }
  }, []);
  
  return { subscribe, unsubscribe };
}

6. Infrastructure as Code​

6.1 Terraform Configuration Structure​

terraform/
├── main.tf
├── variables.tf
├── outputs.tf
├── modules/
│   ├── gke/
│   │   ├── main.tf
│   │   ├── variables.tf
│   │   └── outputs.tf
│   ├── cloud-sql/
│   │   ├── main.tf
│   │   ├── variables.tf
│   │   └── outputs.tf
│   ├── memorystore/
│   │   ├── main.tf
│   │   ├── variables.tf
│   │   └── outputs.tf
│   └── gcs/
│       ├── main.tf
│       ├── variables.tf
│       └── outputs.tf
└── environments/
    ├── dev/
    │   └── terraform.tfvars
    ├── staging/
    │   └── terraform.tfvars
    └── prod/
        └── terraform.tfvars

7. Monitoring & Observability​

7.1 Metrics Collection​

from prometheus_client import Counter, Histogram, Gauge
import time
from functools import wraps

# Define metrics
pdf_uploads_total = Counter(
    'pdf_uploads_total',
    'Total number of PDF uploads',
    ['status']
)

processing_duration_seconds = Histogram(
    'processing_duration_seconds',
    'Time spent processing PDFs',
    ['job_type'],
    buckets=[1, 5, 10, 30, 60, 120, 300]
)

active_websocket_connections = Gauge(
    'active_websocket_connections',
    'Number of active WebSocket connections'
)

ai_api_calls_total = Counter(
    'ai_api_calls_total',
    'Total Claude API calls',
    ['operation', 'status']
)

ai_tokens_used_total = Counter(
    'ai_tokens_used_total',
    'Total tokens used in AI calls',
    ['operation']
)

def track_processing_time(job_type: str):
    """Decorator to track processing time"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            start_time = time.time()
            try:
                result = await func(*args, **kwargs)
                duration = time.time() - start_time
                processing_duration_seconds.labels(
                    job_type=job_type
                ).observe(duration)
                return result
            except Exception as e:
                duration = time.time() - start_time
                processing_duration_seconds.labels(
                    job_type=job_type
                ).observe(duration)
                raise
        return wrapper
    return decorator

8. Security Implementation​

8.1 JWT Authentication​

from datetime import datetime, timedelta
from typing import Optional
import jwt
from fastapi import HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

SECRET_KEY = "your-secret-key"  # Load from Secret Manager
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(
    data: dict,
    expires_delta: Optional[timedelta] = None
) -> str:
    """Create JWT access token"""
    to_encode = data.copy()
    
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(
            minutes=ACCESS_TOKEN_EXPIRE_MINUTES
        )
    
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    
    return encoded_jwt

async def verify_token(
    credentials: HTTPAuthorizationCredentials = Security(security)
) -> dict:
    """Verify JWT token"""
    try:
        payload = jwt.decode(
            credentials.credentials,
            SECRET_KEY,
            algorithms=[ALGORITHM]
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

9. Performance Optimization​

9.1 Caching Strategy​

from functools import lru_cache
import redis
import json
from typing import Optional, Any

class CacheService:
    """Redis-based caching service"""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.default_ttl = 3600  # 1 hour
    
    async def get(self, key: str) -> Optional[Any]:
        """Get value from cache"""
        value = await self.redis.get(key)
        if value:
            return json.loads(value)
        return None
    
    async def set(
        self,
        key: str,
        value: Any,
        ttl: Optional[int] = None
    ) -> bool:
        """Set value in cache"""
        ttl = ttl or self.default_ttl
        serialized = json.dumps(value)
        return await self.redis.setex(key, ttl, serialized)
    
    async def delete(self, key: str) -> bool:
        """Delete key from cache"""
        return await self.redis.delete(key)
    
    def cache_key(self, prefix: str, *args) -> str:
        """Generate cache key"""
        return f"{prefix}:{':'.join(str(arg) for arg in args)}"

# Usage
cache = CacheService(redis_client)

async def get_document_analysis(document_id: str):
    """Get cached analysis or compute"""
    cache_key = cache.cache_key("analysis", document_id)
    
    # Try cache first
    cached = await cache.get(cache_key)
    if cached:
        return cached
    
    # Compute if not cached
    analysis = await compute_analysis(document_id)
    await cache.set(cache_key, analysis, ttl=7200)
    
    return analysis

10. Error Handling​

10.1 Error Response Format​

from fastapi import HTTPException
from pydantic import BaseModel
from typing import Optional, List

class ErrorDetail(BaseModel):
    """Error detail model"""
    code: str
    message: str
    field: Optional[str] = None

class ErrorResponse(BaseModel):
    """Standard error response"""
    error: str
    details: List[ErrorDetail]
    request_id: str
    timestamp: str

class AppException(Exception):
    """Base application exception"""
    
    def __init__(
        self,
        code: str,
        message: str,
        status_code: int = 500,
        details: Optional[List[ErrorDetail]] = None
    ):
        self.code = code
        self.message = message
        self.status_code = status_code
        self.details = details or []
        super().__init__(message)

# Specific exceptions
class DocumentNotFoundError(AppException):
    def __init__(self, document_id: str):
        super().__init__(
            code="DOCUMENT_NOT_FOUND",
            message=f"Document {document_id} not found",
            status_code=404
        )

class ProcessingError(AppException):
    def __init__(self, message: str):
        super().__init__(
            code="PROCESSING_ERROR",
            message=message,
            status_code=500
        )

class ValidationError(AppException):
    def __init__(self, details: List[ErrorDetail]):
        super().__init__(
            code="VALIDATION_ERROR",
            message="Validation failed",
            status_code=422,
            details=details
        )

Appendices​

A. Development Tools​

  • API Testing: Postman collections
  • Load Testing: k6 scripts
  • Database Migrations: Alembic
  • Code Generation: OpenAPI Generator

B. Performance Benchmarks​

  • Target: 1000 PDF uploads/hour
  • P95 latency: <500ms for API calls
  • WebSocket message latency: <100ms

C. Compliance​

  • GDPR: Data encryption, right to deletion
  • SOC 2: Audit logging, access controls
  • HIPAA (future): PHI handling, BAA requirements