Add middleware to check service status (e.g., "read-only", "down for maintenance") and provide suitable error responses

Author: apdavison

api/simqueue/main.py

@@ -1,6 +1,7 @@
 import os
 from contextlib import asynccontextmanager
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
 from fastapi.staticfiles import StaticFiles
 from starlette.middleware.sessions import SessionMiddleware
 from starlette.middleware.cors import CORSMiddleware
@@ -50,6 +51,12 @@ async def lifespan(app: FastAPI):
     await database.disconnect()
 
 
+service_status = getattr(settings, "SERVICE_STATUS", "ok")
+if service_status != "ok":
+    warning_message = f"---\n> ⚠️ **_NOTE:_**  _{service_status}_\n---\n\n"
+    description = warning_message + description
+
+
 app = FastAPI(
     title="EBRAINS Neuromorphic Computing Job Queue API",
     description=description,
@@ -66,6 +73,21 @@ async def lifespan(app: FastAPI):
     allow_headers=["*"],
 )
 
+
+@app.middleware("http")
+async def check_service_status(request: Request, call_next):
+    if request.url.path != "/" and (
+        "down" in service_status
+        or ("read-only" in service_status and request.method in ("POST", "PUT", "PATCH"))
+    ):
+        return JSONResponse(
+            status_code=503,
+            content={"error": service_status},
+        )
+    response = await call_next(request)
+    return response
+
+
 app.include_router(for_users.router, tags=["For all users"])
 app.include_router(for_providers.router, tags=["For use by computing system providers"])
 app.include_router(for_admins.router, tags=["For use by administrators"])

api/simqueue/resources/for_users.py

@@ -92,9 +92,11 @@ async def _check_auth_for_list(token, api_key, collab, user_id, hardware_platfor
 
 @router.get("/")
 def about_this_api():
+    service_status = getattr(settings, "SERVICE_STATUS", "ok")
     return {
         "about": "This is the EBRAINS Neuromorphic Computing Job Queue API.",
         "version": "3",
+        "status": service_status,
         "links": {"documentation": "/docs"},
         "authentication": {
             "server": settings.EBRAINS_IAM_SERVICE_URL,

api/simqueue/settings.py

@@ -24,3 +24,7 @@
 EMAIL_SENDER = "[email protected]"
 EMAIL_PASSWORD = os.environ.get("NMPI_EMAIL_PASSWORD", None)
 ADMIN_EMAIL = os.environ.get("NMPI_ADMIN_EMAIL")
+
+# SERVICE_STATUS = "The service is currently in read-only mode for maintenance"
+# SERVICE_STATUS = "The service is currently down for maintenance. We expect service to be restored on 4th December 2025"
+SERVICE_STATUS = os.environ.get("NMPI_SERVICE_STATUS", "ok")