Spaces:
Sleeping
Sleeping
File size: 4,464 Bytes
92fd1a7 acb0ec6 92fd1a7 acb0ec6 92fd1a7 acb0ec6 92fd1a7 acb0ec6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 |
# interfaces/api/dependencies.py
"""FastAPI dependency injection configuration."""
from typing import Annotated, Optional
from fastapi import Depends, UploadFile, Form, HTTPException, Request, Header
from pydantic import BaseModel, Field, validator
import re
from application.use_cases.container import UseCaseContainer
from infrastructure.services.container import ServiceContainer
from infrastructure.services.jwt_validation_service import JWTValidationService
class ExtractionRequest(BaseModel):
"""Request model for audio extraction."""
output_format: str = Field(default="mp3", description="Output audio format")
quality: str = Field(default="medium", description="Audio quality level")
@validator('output_format')
def validate_format(cls, v):
allowed = ['mp3', 'aac', 'wav', 'flac', 'm4a', 'ogg']
if v.lower() not in allowed:
raise ValueError(f"Format must be one of: {', '.join(allowed)}")
return v.lower()
@validator('quality')
def validate_quality(cls, v):
allowed = ['high', 'medium', 'low']
if v.lower() not in allowed:
raise ValueError(f"Quality must be one of: {', '.join(allowed)}")
return v.lower()
def extraction_params(
output_format: str = Form("mp3"),
quality: str = Form("medium")
) -> ExtractionRequest:
"""Parse and validate extraction parameters."""
return ExtractionRequest(output_format=output_format, quality=quality)
async def validate_video_file(video: UploadFile) -> UploadFile:
"""Validate uploaded video file."""
if not video.filename:
raise HTTPException(400, "No filename provided")
# Check file extension
allowed_extensions = ['.mp4', '.avi', '.mov', '.mkv', '.webm', '.flv', '.wmv', '.m4v']
file_ext = '.' + video.filename.lower().split('.')[-1] if '.' in video.filename else ''
if file_ext not in allowed_extensions:
raise HTTPException(
400,
f"Unsupported video format. Allowed: {', '.join(allowed_extensions)}"
)
# Check content type (basic validation)
if video.content_type and not video.content_type.startswith(('video/', 'application/octet-stream')):
raise HTTPException(400, "File must be a video")
return video
async def validate_bearer_token(
authorization: Optional[str] = Header(None, description="Bearer token for authentication")
) -> str:
"""
Extract and validate bearer token from Authorization header.
Args:
authorization: Authorization header value
Returns:
str: The validated bearer token
Raises:
HTTPException: 401 if token is missing or invalid
"""
if not authorization:
raise HTTPException(
status_code=401,
detail="Missing Authorization header",
headers={"WWW-Authenticate": "Bearer"}
)
# Check if it starts with "Bearer "
if not authorization.startswith("Bearer "):
raise HTTPException(
status_code=401,
detail="Invalid Authorization header format. Expected: Bearer <token>",
headers={"WWW-Authenticate": "Bearer"}
)
# Extract token
token = authorization[7:] # Remove "Bearer " prefix
if not token:
raise HTTPException(
status_code=401,
detail="Empty bearer token",
headers={"WWW-Authenticate": "Bearer"}
)
# Validate JWT structure
jwt_service = JWTValidationService()
if not jwt_service.validate_structure(token):
raise HTTPException(
status_code=401,
detail="Invalid JWT token structure",
headers={"WWW-Authenticate": "Bearer"}
)
return token
def get_services(request: Request) -> ServiceContainer:
"""Get service container from app state."""
return request.app.state.get_services()
def get_use_cases(request: Request) -> UseCaseContainer:
"""Get use case container from app state."""
return request.app.state.get_use_cases()
# Type aliases for dependency injection
ValidatedVideo = Annotated[UploadFile, Depends(validate_video_file)]
ExtractionParams = Annotated[ExtractionRequest, Depends(extraction_params)]
Services = Annotated[ServiceContainer, Depends(get_services)]
UseCases = Annotated[UseCaseContainer, Depends(get_use_cases)]
BearerToken = Annotated[str, Depends(validate_bearer_token)]
|