Spaces:
Sleeping
Sleeping
File size: 4,475 Bytes
92fd1a7 dbe78dd 3482f6a 92fd1a7 dbe78dd 92fd1a7 3482f6a be8d0ca 57c92c0 be8d0ca 3482f6a 92fd1a7 57c92c0 3482f6a dbe78dd 57c92c0 dbe78dd 3482f6a dbe78dd 94a2fe4 3482f6a dbe78dd 3482f6a dbe78dd 92fd1a7 57c92c0 |
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 124 125 126 127 |
"""FastAPI dependency injection configuration."""
from typing import Annotated, Optional
from fastapi import Depends, UploadFile, Form, HTTPException, Request, Header
from fastapi.security import HTTPBearer
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
# Create HTTPBearer scheme that will be recognized by OpenAPI
bearer_scheme = HTTPBearer(
scheme_name="BearerAuth",
bearerFormat="JWT",
description="JWT Bearer token authentication"
)
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 get_validated_token(
credentials = Depends(bearer_scheme)
) -> str:
"""
Extract and validate bearer token from HTTPBearer scheme.
Args:
credentials: HTTPAuthorizationCredentials from HTTPBearer scheme
Returns:
str: The validated bearer token
Raises:
HTTPException: 401 if token is missing or invalid
"""
from infrastructure.config.settings import settings
if not settings.enforce_authentication:
return "authentication-disabled"
if not credentials:
raise HTTPException(
status_code=401,
detail="Missing Authorization header",
headers={"WWW-Authenticate": "Bearer"}
)
# Extract token from credentials
token = credentials.credentials
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(get_validated_token)]
|