Add /maintenance and /maintenance/scheduled GET, POST endpoints

.github/ci_config.yml

@@ -4,6 +4,8 @@ app:
   # API will auto-reload when changes on code files are detected
   reload: true
   url_prefix: ""
+  maintenance_file: operationsgateway_api/maintenance.json.example
+  scheduled_maintenance_file: operationsgateway_api/scheduled_maintenance.json.example
 images:
   thumbnail_size: [50, 50]
   default_colour_map: viridis

operationsgateway_api/config.yml.example

@@ -6,6 +6,8 @@ app:
   # API will auto-reload when changes on code files are detected
   reload: true
   url_prefix: ""
+  maintenance_file: operationsgateway_api/maintenance.json.example
+  scheduled_maintenance_file: operationsgateway_api/scheduled_maintenance.json.example
 images:
   # Thumbnail sizes should only ever be two element lists, of a x, y resolution
   thumbnail_size: [50, 50]

operationsgateway_api/maintenance.json.example

@@ -0,0 +1,4 @@
+{
+    "show": false,
+    "message": ""
+}

operationsgateway_api/scheduled_maintenance.json.example

@@ -0,0 +1,5 @@
+{
+    "show": false,
+    "message": "",
+    "severity": "info"
+}

operationsgateway_api/src/config.py

@@ -7,6 +7,7 @@
 from pydantic import (
     BaseModel,
     field_validator,
+    FilePath,
     StrictBool,
     StrictInt,
     StrictStr,
@@ -23,6 +24,8 @@ class App(BaseModel):
     port: Optional[StrictInt] = None
     reload: Optional[StrictBool] = None
     url_prefix: StrictStr
+    maintenance_file: FilePath
+    scheduled_maintenance_file: FilePath
 
 
 class ImagesConfig(BaseModel):

operationsgateway_api/src/main.py

@@ -25,6 +25,7 @@
     functions,
     images,
     ingest_data,
+    maintenance,
     records,
     sessions,
     user_preferences,
@@ -135,6 +136,7 @@ def add_router_to_app(api_router: APIRouter):
 add_router_to_app(users.router)
 add_router_to_app(functions.router)
 add_router_to_app(filters.router)
+add_router_to_app(maintenance.router)
 
 log.debug("ROUTE_MAPPINGS contents:")
 for item in ROUTE_MAPPINGS.items():

operationsgateway_api/src/models.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from enum import StrEnum
 from typing import Any, Callable, ClassVar, Dict, List, Literal, Optional, Union
 
 from bson.objectid import ObjectId
@@ -268,3 +269,17 @@ class FavouriteFilterModel(BaseModel):
 class Function(BaseModel):
     name: constr(strip_whitespace=True, min_length=1)
     expression: constr(strip_whitespace=True, min_length=1)
+
+
+class Severity(StrEnum):
+    INFO = "info"
+    WARNING = "warning"
+
+
+class MaintenanceModel(BaseModel):
+    show: bool
+    message: str
+
+
+class ScheduledMaintenanceModel(MaintenanceModel):
+    severity: Severity

operationsgateway_api/src/routes/maintenance.py

@@ -0,0 +1,79 @@
+from functools import lru_cache
+import json
+import logging
+
+from fastapi import APIRouter, Depends
+from typing_extensions import Annotated
+
+from operationsgateway_api.src.auth.authorisation import authorise_route
+from operationsgateway_api.src.config import Config
+from operationsgateway_api.src.models import MaintenanceModel, ScheduledMaintenanceModel
+
+
+log = logging.getLogger()
+router = APIRouter()
+AuthoriseRoute = Annotated[str, Depends(authorise_route)]
+
+
+@router.get(
+    "/maintenance",
+    summary="Get the current maintenance message and whether to show it",
+    response_description="The current maintenance message and whether to show it",
+    tags=["Maintenance"],
+)
+def get_maintenance(access_token: AuthoriseRoute) -> MaintenanceModel:
+    return _get_maintenance()
+
+
+@router.post(
+    "/maintenance",
+    summary="Set the current maintenance message and whether to show it",
+    tags=["Maintenance"],
+)
+def set_maintenance(
+    access_token: AuthoriseRoute,
+    maintenance_body: MaintenanceModel,
+) -> None:
+    with open(Config.config.app.maintenance_file, "w") as f:
+        f.write(maintenance_body.model_dump_json())
+    _get_maintenance.cache_clear()
+
+
+@router.get(
+    "/maintenance/scheduled",
+    summary="Get the current scheduled maintenance message and whether to show it",
+    response_description=(
+        "The current scheduled maintenance message and whether to show it"
+    ),
+    tags=["Maintenance"],
+)
+def get_scheduled_maintenance(
+    access_token: AuthoriseRoute,
+) -> ScheduledMaintenanceModel:
+    return _get_scheduled_maintenance()
+
+
+@router.post(
+    "/maintenance/scheduled",
+    summary="Set the current scheduled maintenance message and whether to show it",
+    tags=["Maintenance"],
+)
+def set_scheduled_maintenance(
+    access_token: AuthoriseRoute,
+    maintenance_body: ScheduledMaintenanceModel,
+) -> None:
+    with open(Config.config.app.scheduled_maintenance_file, "w") as f:
+        f.write(maintenance_body.model_dump_json())
+    _get_scheduled_maintenance.cache_clear()
+
+
+@lru_cache
+def _get_maintenance() -> dict:
+    with open(Config.config.app.maintenance_file, "rb") as f:
+        return json.load(f)
+
+
+@lru_cache
+def _get_scheduled_maintenance() -> dict:
+    with open(Config.config.app.scheduled_maintenance_file, "rb") as f:
+        return json.load(f)

operationsgateway_api/src/users/user.py

@@ -20,6 +20,10 @@ class User:
         "/users PATCH",
         "/users GET",
         "/users/{id_} DELETE",
+        "/maintenance GET"
+        "/maintenance POST"
+        "/maintenance/scheduled GET"
+        "/maintenance/scheduled POST",
     ]
 
     auth_type_list = [

test/endpoints/test_maintenance.py

@@ -0,0 +1,105 @@
+import json
+import tempfile
+from unittest.mock import patch
+
+from fastapi.testclient import TestClient
+
+
+class TestMaintenance:
+    """
+    Both test combine the GET and POST requests in order to test the cache is cleared
+    after POSTing a new message.
+    """
+
+    def test_maintenance(self, test_app: TestClient, login_and_get_token: str):
+        initial_content = {"show": False, "message": ""}
+        updated_content = {"show": True, "message": "message"}
+        target = "operationsgateway_api.src.config.Config.config.app.maintenance_file"
+        tmp_file = tempfile.NamedTemporaryFile()
+        with patch(target, tmp_file.name):
+            # POST the initial contents (we are using a tmpfile, which will start empty)
+            response = test_app.post(
+                url="/maintenance",
+                content=json.dumps(initial_content),
+                headers={"Authorization": f"Bearer {login_and_get_token}"},
+            )
+            assert response.status_code == 200
+            assert json.loads(response.content) is None
+            assert json.load(tmp_file) == initial_content
+            tmp_file.seek(0)  # Reset so we can read again later
+
+            # Calling GET will save the return value to cache
+            response = test_app.get(
+                url="/maintenance",
+                headers={"Authorization": f"Bearer {login_and_get_token}"},
+            )
+            assert response.status_code == 200
+            assert json.loads(response.content) == initial_content
+
+            # Calling POST will clear the cache after writing to file
+            response = test_app.post(
+                url="/maintenance",
+                content=json.dumps(updated_content),
+                headers={"Authorization": f"Bearer {login_and_get_token}"},
+            )
+            assert response.status_code == 200
+            assert json.loads(response.content) is None
+            assert json.load(tmp_file) == updated_content
+
+            # Cache has been cleared so should get the new message back
+            response = test_app.get(
+                url="/maintenance",
+                headers={"Authorization": f"Bearer {login_and_get_token}"},
+            )
+            assert response.status_code == 200
+            assert json.loads(response.content) == updated_content
+
+    def test_scheduled_maintenance(
+        self,
+        test_app: TestClient,
+        login_and_get_token: str,
+    ):
+        initial_content = {"show": False, "message": "", "severity": "info"}
+        updated_content = {"show": True, "message": "message", "severity": "warning"}
+        target = (
+            "operationsgateway_api.src.config.Config.config.app."
+            "scheduled_maintenance_file"
+        )
+        tmp_file = tempfile.NamedTemporaryFile()
+        with patch(target, tmp_file.name):
+            # POST the initial contents (we are using a tmpfile, which will start empty)
+            response = test_app.post(
+                url="/maintenance/scheduled",
+                content=json.dumps(initial_content),
+                headers={"Authorization": f"Bearer {login_and_get_token}"},
+            )
+            assert response.status_code == 200
+            assert json.loads(response.content) is None
+            assert json.load(tmp_file) == initial_content
+            tmp_file.seek(0)  # Reset so we can read again later
+
+            # Calling GET will save the return value to cache
+            response = test_app.get(
+                url="/maintenance/scheduled",
+                headers={"Authorization": f"Bearer {login_and_get_token}"},
+            )
+            assert response.status_code == 200
+            assert json.loads(response.content) == initial_content
+
+            # Calling POST will clear the cache after writing to file
+            response = test_app.post(
+                url="/maintenance/scheduled",
+                content=json.dumps(updated_content),
+                headers={"Authorization": f"Bearer {login_and_get_token}"},
+            )
+            assert response.status_code == 200
+            assert json.loads(response.content) is None
+            assert json.load(tmp_file) == updated_content
+
+            # Cache has been cleared so should get the new message back
+            response = test_app.get(
+                url="/maintenance/scheduled",
+                headers={"Authorization": f"Bearer {login_and_get_token}"},
+            )
+            assert response.status_code == 200
+            assert json.loads(response.content) == updated_content