Spaces:
Running
Running
| 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.encoders import jsonable_encoder | |
| 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, | |
| assets, | |
| 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.request_limits import RequestBodyLimitMiddleware | |
| 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.security.errors import APIKeyConflictError, APIKeyNotFoundError, ForbiddenError, RateLimitError, SecurityError, UnauthorizedError | |
| 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, container.assets) | |
| 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() | |
| 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, | |
| ) | |
| application.add_middleware( | |
| RequestBodyLimitMiddleware, | |
| max_bytes=active_settings.max_upload_size, | |
| ) | |
| 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"], | |
| ) | |
| 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) | |
| 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) | |
| async def validation_error_handler( | |
| request: Request, exc: RequestValidationError | |
| ) -> JSONResponse: | |
| try: | |
| details = exc.errors(include_url=False, include_context=False) | |
| except TypeError: | |
| details = exc.errors() | |
| body = ErrorResponse( | |
| request_id=request.state.request_id, | |
| error=ErrorBody( | |
| code="VALIDATION_ERROR", | |
| message="Request validation failed", | |
| details=jsonable_encoder(details), | |
| ), | |
| ) | |
| return DefaultJSONResponse(body.model_dump(), status_code=422) | |
| 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) | |
| async def security_error_handler(request: Request, exc: SecurityError) -> JSONResponse: | |
| # Security-domain errors can be raised by endpoint services as well as | |
| # the authentication middleware. Map them explicitly so an expected | |
| # authorization failure never leaks out as an opaque HTTP 500. | |
| status_code = 403 | |
| message = "Security policy rejected the request." | |
| headers: dict[str, str] = {} | |
| if isinstance(exc, UnauthorizedError): | |
| status_code = 401 | |
| message = "Invalid or expired API key." | |
| headers["WWW-Authenticate"] = "Bearer" | |
| elif isinstance(exc, ForbiddenError): | |
| status_code = 403 | |
| message = "Missing required scope or active workspace membership." | |
| elif isinstance(exc, RateLimitError): | |
| status_code = 429 | |
| message = "Rate limit exceeded." | |
| headers["Retry-After"] = str(exc.retry_after) | |
| elif isinstance(exc, APIKeyNotFoundError): | |
| status_code = 404 | |
| message = "API key not found." | |
| elif isinstance(exc, APIKeyConflictError): | |
| status_code = 409 | |
| message = str(exc) or "API key conflict." | |
| body = ErrorResponse( | |
| request_id=request.state.request_id, | |
| error=ErrorBody(code=type(exc).__name__.upper(), message=message), | |
| ) | |
| return DefaultJSONResponse(body.model_dump(), status_code=status_code, headers=headers) | |
| 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(assets.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") | |
| async def root() -> dict[str, str]: | |
| return { | |
| "name": active_settings.app_name, | |
| "version": active_settings.app_version, | |
| "status": "healthy", | |
| "docs": "/docs", | |
| } | |
| 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", "/ready", "/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() | |