""" Pydantic Models for API All request and response models for API endpoints. Provides validation and documentation for API contracts. """ from typing import Dict, List, Optional, Any from pydantic import BaseModel, Field, field_validator from datetime import datetime # ============================================================================ # AUTHENTICATION MODELS # ============================================================================ class LoginRequest(BaseModel): """Login request model""" username: str = Field(..., min_length=1, max_length=50) password: str = Field(..., min_length=1) rememberMe: bool = False class LoginResponse(BaseModel): """Login response model""" success: bool token: Optional[str] = None username: Optional[str] = None role: Optional[str] = None expires_at: Optional[str] = None message: Optional[str] = None class ChangePasswordRequest(BaseModel): """Password change request""" current_password: str = Field(..., min_length=1) new_password: str = Field(..., min_length=8) class UserPreferences(BaseModel): """User preferences model""" theme: Optional[str] = Field(None, pattern=r'^(light|dark|system)$') notifications_enabled: Optional[bool] = None default_platform: Optional[str] = None # ============================================================================ # DOWNLOAD MODELS # ============================================================================ class DownloadResponse(BaseModel): """Single download record""" id: int platform: str source: str content_type: Optional[str] = None filename: Optional[str] = None file_path: Optional[str] = None file_size: Optional[int] = None download_date: str post_date: Optional[str] = None status: Optional[str] = None width: Optional[int] = None height: Optional[int] = None class StatsResponse(BaseModel): """Statistics response""" total_downloads: int by_platform: Dict[str, int] total_size: int recent_24h: int duplicates_prevented: int review_queue_count: int recycle_bin_count: int = 0 class TriggerRequest(BaseModel): """Manual download trigger request""" username: Optional[str] = None content_types: Optional[List[str]] = None # ============================================================================ # PLATFORM MODELS # ============================================================================ class PlatformStatus(BaseModel): """Platform status model""" platform: str enabled: bool last_run: Optional[str] = None next_run: Optional[str] = None status: str class PlatformConfig(BaseModel): """Platform configuration""" enabled: bool = False username: Optional[str] = None interval_hours: int = Field(24, ge=1, le=168) randomize: bool = True randomize_minutes: int = Field(30, ge=0, le=180) # ============================================================================ # CONFIGURATION MODELS # ============================================================================ class PushoverConfig(BaseModel): """Pushover notification configuration""" enabled: bool user_key: Optional[str] = Field(None, min_length=30, max_length=30, pattern=r'^[A-Za-z0-9]+$') api_token: Optional[str] = Field(None, min_length=30, max_length=30, pattern=r'^[A-Za-z0-9]+$') priority: int = Field(0, ge=-2, le=2) sound: str = Field("pushover", pattern=r'^[a-z_]+$') class SchedulerConfig(BaseModel): """Scheduler configuration""" enabled: bool interval_hours: int = Field(24, ge=1, le=168) randomize: bool = True randomize_minutes: int = Field(30, ge=0, le=180) class RecycleBinConfig(BaseModel): """Recycle bin configuration""" enabled: bool = True path: str = "/opt/immich/recycle" retention_days: int = Field(30, ge=1, le=365) max_size_gb: int = Field(50, ge=1, le=1000) auto_cleanup: bool = True class ConfigUpdate(BaseModel): """Configuration update request""" config: Dict[str, Any] class Config: extra = "allow" # ============================================================================ # HEALTH MODELS # ============================================================================ class ServiceHealth(BaseModel): """Individual service health status""" status: str = Field(..., pattern=r'^(healthy|unhealthy|unknown)$') message: Optional[str] = None last_check: Optional[str] = None details: Optional[Dict[str, Any]] = None class HealthStatus(BaseModel): """Overall health status""" status: str services: Dict[str, ServiceHealth] last_check: str version: str # ============================================================================ # MEDIA MODELS # ============================================================================ class MediaItem(BaseModel): """Media item in gallery""" id: int file_path: str filename: str platform: str source: str content_type: str file_size: Optional[int] = None width: Optional[int] = None height: Optional[int] = None post_date: Optional[str] = None download_date: Optional[str] = None face_match: Optional[str] = None face_confidence: Optional[float] = None class BatchDeleteRequest(BaseModel): """Batch delete request""" file_paths: List[str] = Field(..., min_length=1) permanent: bool = False class BatchMoveRequest(BaseModel): """Batch move request""" file_paths: List[str] = Field(..., min_length=1) destination: str # ============================================================================ # REVIEW MODELS # ============================================================================ class ReviewItem(BaseModel): """Review queue item""" id: int file_path: str filename: str platform: str source: str content_type: str file_size: Optional[int] = None width: Optional[int] = None height: Optional[int] = None detected_faces: Optional[int] = None best_match: Optional[str] = None match_confidence: Optional[float] = None class ReviewKeepRequest(BaseModel): """Keep review item request""" file_path: str destination: str new_name: Optional[str] = None # ============================================================================ # FACE RECOGNITION MODELS # ============================================================================ class FaceReference(BaseModel): """Face reference model""" id: str name: str created_at: str encoding_count: int = 1 thumbnail: Optional[str] = None class AddReferenceRequest(BaseModel): """Add face reference request""" file_path: str name: str # ============================================================================ # RECYCLE BIN MODELS # ============================================================================ class RecycleItem(BaseModel): """Recycle bin item""" id: str original_path: str original_filename: str recycle_path: str file_size: Optional[int] = None deleted_at: str deleted_from: str metadata: Optional[Dict[str, Any]] = None class RestoreRequest(BaseModel): """Restore from recycle bin request""" recycle_id: str restore_to: Optional[str] = None # Original path if None # ============================================================================ # VIDEO DOWNLOAD MODELS # ============================================================================ class VideoInfoRequest(BaseModel): """Video info request""" url: str = Field(..., pattern=r'^https?://') class VideoDownloadRequest(BaseModel): """Video download request""" url: str = Field(..., pattern=r'^https?://') format: Optional[str] = None quality: Optional[str] = None class VideoStatus(BaseModel): """Video download status""" video_id: str platform: str status: str progress: Optional[float] = None title: Optional[str] = None error: Optional[str] = None # ============================================================================ # SCHEDULER MODELS # ============================================================================ class TaskStatus(BaseModel): """Scheduler task status""" task_id: str platform: str source: Optional[str] = None status: str last_run: Optional[str] = None next_run: Optional[str] = None error_count: int = 0 last_error: Optional[str] = None class SchedulerStatus(BaseModel): """Overall scheduler status""" running: bool tasks: List[TaskStatus] current_activity: Optional[Dict[str, Any]] = None # ============================================================================ # NOTIFICATION MODELS # ============================================================================ class NotificationItem(BaseModel): """Notification item""" id: int platform: str source: str content_type: str message: str title: Optional[str] = None sent_at: str download_count: int status: str # ============================================================================ # SEMANTIC SEARCH MODELS # ============================================================================ class SemanticSearchRequest(BaseModel): """Semantic search request""" query: str = Field(..., min_length=1, max_length=500) limit: int = Field(50, ge=1, le=200) threshold: float = Field(0.3, ge=0.0, le=1.0) class SimilarImagesRequest(BaseModel): """Find similar images request""" file_id: int limit: int = Field(20, ge=1, le=100) threshold: float = Field(0.5, ge=0.0, le=1.0) # ============================================================================ # TAG MODELS # ============================================================================ class TagCreate(BaseModel): """Create tag request""" name: str = Field(..., min_length=1, max_length=50) color: Optional[str] = Field(None, pattern=r'^#[0-9A-Fa-f]{6}$') class TagUpdate(BaseModel): """Update tag request""" name: Optional[str] = Field(None, min_length=1, max_length=50) color: Optional[str] = Field(None, pattern=r'^#[0-9A-Fa-f]{6}$') class BulkTagRequest(BaseModel): """Bulk tag operation request""" file_ids: List[int] = Field(..., min_length=1) tag_ids: List[int] = Field(..., min_length=1) operation: str = Field(..., pattern=r'^(add|remove)$') # ============================================================================ # COLLECTION MODELS # ============================================================================ class CollectionCreate(BaseModel): """Create collection request""" name: str = Field(..., min_length=1, max_length=100) description: Optional[str] = Field(None, max_length=500) class CollectionUpdate(BaseModel): """Update collection request""" name: Optional[str] = Field(None, min_length=1, max_length=100) description: Optional[str] = Field(None, max_length=500) class CollectionBulkAdd(BaseModel): """Bulk add files to collection""" file_ids: List[int] = Field(..., min_length=1) # ============================================================================ # SMART FOLDER MODELS # ============================================================================ class SmartFolderCreate(BaseModel): """Create smart folder request""" name: str = Field(..., min_length=1, max_length=100) query_rules: Dict[str, Any] class SmartFolderUpdate(BaseModel): """Update smart folder request""" name: Optional[str] = Field(None, min_length=1, max_length=100) query_rules: Optional[Dict[str, Any]] = None # ============================================================================ # SCRAPER MODELS # ============================================================================ class ScraperUpdate(BaseModel): """Update scraper settings""" enabled: Optional[bool] = None proxy: Optional[str] = None interval_hours: Optional[int] = Field(None, ge=1, le=168) settings: Optional[Dict[str, Any]] = None class CookieUpload(BaseModel): """Cookie upload for scraper""" cookies: str # JSON string of cookies source: str = Field(..., pattern=r'^(browser|manual|extension)$') # ============================================================================ # STANDARD RESPONSE MODELS # ============================================================================ class SuccessResponse(BaseModel): """Standard success response""" success: bool = True message: str = "Operation completed successfully" class DataResponse(BaseModel): """Standard data response wrapper""" success: bool = True data: Any class MessageResponse(BaseModel): """Response with just a message""" message: str class CountResponse(BaseModel): """Response with count of affected items""" message: str count: int class IdResponse(BaseModel): """Response with created resource ID""" id: int message: str = "Resource created successfully" class PaginatedResponse(BaseModel): """Paginated list response base""" items: List[Any] total: int limit: int offset: int has_more: bool = False @classmethod def create(cls, items: List[Any], total: int, limit: int, offset: int): """Helper to create paginated response""" return cls( items=items, total=total, limit=limit, offset=offset, has_more=(offset + len(items)) < total )