| 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() |
|
|
