5e2094f2ad
FastAPI service in infra/media-worker implementing every MediaInferenceProvider operation behind HTTP API v1, plus capabilities. Bearer auth, MEDIA_ROOT realpath enforcement, strict request schemas, cancellation on disconnect, GPU/CPU gates with 503 overflow, input size/duration limits, lazy SHA256-pinned model downloads, fail-closed redaction with atomic output, offline pytest suite with fake backends, CPU/CUDA/ROCm Dockerfile and a license review of every model. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0138YqDj3ttfayRwuMKKrqVj
63 lines
1.7 KiB
Python
63 lines
1.7 KiB
Python
"""Error envelope shared by every route: {"error": {"code", "message", "operation"}}."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi.responses import JSONResponse
|
|
|
|
STATUS_BY_CODE = {
|
|
"bad_request": 400,
|
|
"invalid_path": 400,
|
|
"unauthorized": 401,
|
|
"too_large": 413,
|
|
"unsupported": 501,
|
|
"busy": 503,
|
|
"internal": 500,
|
|
}
|
|
|
|
|
|
class MediaError(Exception):
|
|
def __init__(self, code: str, message: str, operation: str | None = None) -> None:
|
|
super().__init__(message)
|
|
self.code = code
|
|
self.message = message
|
|
self.operation = operation
|
|
|
|
@property
|
|
def status(self) -> int:
|
|
return STATUS_BY_CODE[self.code]
|
|
|
|
|
|
class Unsupported(MediaError):
|
|
def __init__(self, reason: str, operation: str | None = None) -> None:
|
|
super().__init__("unsupported", reason, operation)
|
|
|
|
|
|
class InvalidPath(MediaError):
|
|
def __init__(self, message: str = "path is outside MEDIA_ROOT") -> None:
|
|
super().__init__("invalid_path", message)
|
|
|
|
|
|
class BadRequest(MediaError):
|
|
def __init__(self, message: str) -> None:
|
|
super().__init__("bad_request", message)
|
|
|
|
|
|
class Busy(MediaError):
|
|
def __init__(self) -> None:
|
|
super().__init__("busy", "inference queue is full")
|
|
|
|
|
|
class Cancelled(Exception):
|
|
"""The client went away or the request ran past MEDIA_MAX_SECONDS."""
|
|
|
|
def __init__(self, reason: str = "client disconnected") -> None:
|
|
super().__init__(reason)
|
|
self.reason = reason
|
|
|
|
|
|
def error_response(code: str, message: str, operation: str | None, status: int | None = None) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=status or STATUS_BY_CODE[code],
|
|
content={"error": {"code": code, "message": message, "operation": operation}},
|
|
)
|