from __future__ import annotations import time from collections.abc import AsyncIterator from contextlib import asynccontextmanager from uuid import uuid4 from fastapi import FastAPI, HTTPException, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse, ORJSONResponse from starlette.middleware.base import RequestResponseEndpoint from starlette.middleware.cors import CORSMiddleware from starlette.responses import Response from app.ai import api as ai from app.analytics import api as analytics from app.brand import api as brand from app.api import ( api_keys, audio, generation, health, image, media, probe, social, templates, video, whisper, ytdlp, ) from app.container import build_container from app.copilot import api as copilot from app.core.config import Settings, get_settings from app.core.exceptions import MediaAPIError from app.core.logger import configure_logging, get_logger, request_id_context from app.core.response import ErrorBody, ErrorResponse from app.mcp.server import create_mcp_server from app.projects import api as projects from app.security.middleware import APIKeyAuthenticationMiddleware from app.social.workers.scheduler import SocialSchedulerWorker from app.templates import marketplace_api from app.workers.cleanup_worker import CleanupWorker configure_logging() logger = get_logger(__name__) try: __import__("orjson") DefaultJSONResponse = ORJSONResponse except ImportError: # pragma: no cover - production requirements always install orjson DefaultJSONResponse = JSONResponse try: import psutil as _psutil except ImportError: # pragma: no cover - production requirements always install psutil _psutil = None # type: ignore[assignment] def create_app(settings: Settings | None = None) -> FastAPI: active_settings = settings or get_settings() active_settings.ensure_directories() container = build_container(active_settings) cleanup_worker = CleanupWorker(container.cleanup, active_settings.cleanup_interval_seconds) social_worker = SocialSchedulerWorker( container.social, active_settings.social_scheduler_interval_seconds ) mcp_server = create_mcp_server(container) mcp_http_app = mcp_server.streamable_http_app() @asynccontextmanager async def lifespan(application: FastAPI) -> AsyncIterator[None]: application.state.container = container application.state.mcp_server = mcp_server await container.security_database.initialize() if not await container.security_database.schema_ready(): missing = ", ".join(await container.security_database.missing_schema_objects()) raise RuntimeError( "Security schema is unavailable; apply app/security/migrations/ " "and app/projects/migrations/. " f"Missing: {missing}" ) await container.security_database.verify_execution_boundary( expected_role=active_settings.security_database_role, enforce_rls=active_settings.security_enforce_rls, ) await container.generation.initialize() await container.api_keys.ensure_bootstrap_admin() await container.tenants.ensure_all_api_key_principals() await container.social.initialize() await container.analytics.initialize(container.social.ready) await container.social.adopt_legacy_workspaces(await container.tenants.list_principals()) async with mcp_server.session_manager.run(): await cleanup_worker.start() await container.generation_worker.start() await container.render_worker.start() if active_settings.social_enabled and active_settings.social_worker_enabled: await social_worker.start() await container.analytics_worker.start() logger.info( "media API started", extra={"version": active_settings.app_version, "port": active_settings.port}, ) try: yield finally: await social_worker.stop() await container.analytics_worker.stop() await container.generation_worker.stop() await container.render_worker.stop() await cleanup_worker.stop() await container.generation.close() await container.social.close() await container.security_database.close() logger.info("media API stopped") application = FastAPI( title=active_settings.app_name, version=active_settings.app_version, description=( "CPU-optimized FFmpeg, yt-dlp, FFprobe, faster-whisper, MCP, and YAML template API." ), default_response_class=DefaultJSONResponse, lifespan=lifespan, docs_url="/docs", redoc_url="/redoc", openapi_url="/openapi.json", ) # State is also assigned eagerly so ASGI test clients without lifespan support work. application.state.container = container application.state.mcp_server = mcp_server # Added before request_context so the request-ID/logging middleware remains outermost. application.add_middleware( APIKeyAuthenticationMiddleware, settings=active_settings, api_keys=container.api_keys, rate_limiter=container.rate_limiter, audit=container.audit, ) if active_settings.allowed_cors_origins: application.add_middleware( CORSMiddleware, allow_origins=list(active_settings.allowed_cors_origins), allow_credentials=True, allow_methods=["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"], allow_headers=[ "Authorization", "Content-Type", "Idempotency-Key", "X-Request-ID", "X-MediaRouter-Human-Role", ], expose_headers=["Content-Disposition", "X-Request-ID"], ) @application.middleware("http") async def request_context(request: Request, call_next: RequestResponseEndpoint) -> Response: request_id = str(uuid4()) request.state.request_id = request_id request.state.operation = f"{request.method} {request.url.path}" token = request_id_context.set(request_id) started = time.monotonic() process = _psutil.Process() if _psutil else None status_code = 500 try: response = await call_next(request) status_code = response.status_code response.headers["X-Request-ID"] = request_id return response finally: elapsed = round(time.monotonic() - started, 4) logger.info( "request completed", extra={ "operation": request.state.operation, "method": request.method, "path": request.url.path, "status_code": status_code, "duration": elapsed, "cpu_percent": _psutil.cpu_percent(interval=None) if _psutil else None, "memory_bytes": process.memory_info().rss if process else None, }, ) await container.cleanup.complete(request_id) request_id_context.reset(token) @application.exception_handler(MediaAPIError) async def media_error_handler(request: Request, exc: MediaAPIError) -> JSONResponse: logger.warning( "request failed", extra={"error_code": exc.code, "operation": request.state.operation}, ) body = ErrorResponse( request_id=request.state.request_id, error=ErrorBody(code=exc.code, message=exc.message, details=exc.details), ) return DefaultJSONResponse(body.model_dump(), status_code=exc.status_code) @application.exception_handler(RequestValidationError) async def validation_error_handler( request: Request, exc: RequestValidationError ) -> JSONResponse: body = ErrorResponse( request_id=request.state.request_id, error=ErrorBody( code="VALIDATION_ERROR", message="Request validation failed", details=exc.errors(include_url=False, include_context=False), ), ) return DefaultJSONResponse(body.model_dump(), status_code=422) @application.exception_handler(HTTPException) async def http_error_handler(request: Request, exc: HTTPException) -> JSONResponse: body = ErrorResponse( request_id=request.state.request_id, error=ErrorBody(code="HTTP_ERROR", message=str(exc.detail)), ) return DefaultJSONResponse(body.model_dump(), status_code=exc.status_code) @application.exception_handler(Exception) async def unexpected_error_handler(request: Request, exc: Exception) -> JSONResponse: logger.exception("unhandled request error", extra={"operation": request.state.operation}) body = ErrorResponse( request_id=request.state.request_id, error=ErrorBody( code="INTERNAL_ERROR", message="An unexpected internal error occurred", ), ) return DefaultJSONResponse(body.model_dump(), status_code=500) application.include_router(health.router) application.include_router(api_keys.router) application.include_router(media.router) application.include_router(video.router) application.include_router(audio.router) application.include_router(image.router) application.include_router(probe.router) application.include_router(ytdlp.router) application.include_router(whisper.router) application.include_router(marketplace_api.router) # Register the static marketplace prefix before the legacy dynamic # /v1/templates/{template_id} route so "catalog" cannot be captured as a # YAML workflow template ID. application.include_router(templates.router) application.include_router(ai.router) application.include_router(copilot.router) application.include_router(generation.router) application.include_router(projects.router) application.include_router(social.router) application.include_router(analytics.router) application.include_router(brand.router) application.mount("/mcp", mcp_http_app, name="mcp") @application.get("/", tags=["public"], include_in_schema=False) async def root() -> dict[str, str]: return { "name": active_settings.app_name, "version": active_settings.app_version, "status": "healthy", "docs": "/docs", } @application.get("/version", tags=["public"], include_in_schema=False) async def version() -> dict[str, str]: return {"version": active_settings.app_version} original_openapi = application.openapi def secure_openapi() -> dict[str, object]: if application.openapi_schema: return application.openapi_schema # type: ignore[return-value] schema = original_openapi() components = schema.setdefault("components", {}) components.setdefault("securitySchemes", {})["APIKeyBearer"] = { "type": "http", "scheme": "bearer", "bearerFormat": "mp_live_…", "description": "MediaRouter API key", } schema["security"] = [{"APIKeyBearer": []}] # The provider callback cannot send an API key. It is authenticated by # the random, expiring, single-use OAuth state stored server-side. for path in ("/health", "/v1/social/accounts/{provider}/callback"): for operation in schema.get("paths", {}).get(path, {}).values(): if isinstance(operation, dict): operation["security"] = [] application.openapi_schema = schema return schema application.openapi = secure_openapi # type: ignore[method-assign] return application app = create_app()