File size: 3,918 Bytes
038574d | 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 | from __future__ import annotations
from fastapi import FastAPI
from fastapi.responses import JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from .config import settings
from .automation import router as automation_router
from .registry import SERVICES, load_service_app, service_enabled
from .security import ApiKeyMiddleware, BodyLimitMiddleware, RateLimitMiddleware, RequestContextMiddleware, require_configured_security
VERSION = "1.0.0-enterprise"
def create_app() -> FastAPI:
settings.ensure_runtime_dirs()
app = FastAPI(
title=settings.app_name,
version=VERSION,
docs_url="/docs" if settings.allow_public_docs or not settings.api_key_required else None,
redoc_url="/redoc" if settings.allow_public_docs or not settings.api_key_required else None,
)
app.add_middleware(RequestContextMiddleware)
app.add_middleware(BodyLimitMiddleware, settings=settings)
app.add_middleware(RateLimitMiddleware, settings=settings)
app.add_middleware(ApiKeyMiddleware, settings=settings)
app.include_router(automation_router)
studio_dir = settings.base_dir / "maester_enterprise" / "studio"
if studio_dir.exists():
app.mount("/studio", StaticFiles(directory=studio_dir, html=True), name="studio")
mounted: list[dict[str, str]] = []
failed: list[dict[str, str]] = []
if settings.mount_services:
for service in SERVICES:
if not service_enabled(service):
continue
try:
service_app = load_service_app(settings.services_dir, service)
app.mount(service.path, service_app)
mounted.append({"key": service.key, "path": service.path, "title": service.title})
except Exception as exc:
failed.append({"key": service.key, "error": str(exc)})
@app.get("/api/meta")
def metadata() -> dict:
return {
"name": settings.app_name,
"version": VERSION,
"environment": settings.environment,
"cpu_safe_mode": settings.cpu_safe_mode,
"security": {
"api_key_required": settings.api_key_required,
"docs_public": settings.allow_public_docs,
},
"services": mounted,
"failed_services": failed,
}
@app.get("/", response_class=RedirectResponse)
def root() -> RedirectResponse:
return RedirectResponse(url="/studio/")
@app.get("/version")
def version() -> dict[str, str]:
return {"version": VERSION}
@app.get("/health")
def health() -> dict:
return {"status": "ok", "services_mounted": len(mounted), "services_failed": len(failed)}
@app.get("/ready")
def ready() -> JSONResponse:
try:
require_configured_security(settings)
except Exception as exc:
return JSONResponse({"status": "not_ready", "detail": str(exc)}, status_code=503)
status_code = 200 if not failed else 207
return JSONResponse({"status": "ready", "mounted": mounted, "failed": failed}, status_code=status_code)
@app.get("/services")
def services() -> dict:
catalog = []
mounted_by_key = {item["key"]: item for item in mounted}
failed_by_key = {item["key"]: item for item in failed}
for service in SERVICES:
catalog.append(
{
"key": service.key,
"title": service.title,
"path": service.path,
"description": service.description,
"enabled": service_enabled(service),
"mounted": service.key in mounted_by_key,
"error": failed_by_key.get(service.key, {}).get("error"),
}
)
return {"services": catalog}
return app
app = create_app()
|