Add global URN resolver

Author: uittenbroekrobbert

task_registry/api/main.py

@@ -1,8 +1,9 @@
 from fastapi import APIRouter
-from task_registry.api.routes import health, instruments, measures, requirements
+from task_registry.api.routes import health, instruments, measures, requirements, urns
 
 api_router = APIRouter()
 api_router.include_router(health.router, prefix="/health", tags=["health"])
 api_router.include_router(instruments.router, prefix="/instruments", tags=["instruments"])
 api_router.include_router(measures.router, prefix="/measures", tags=["measures"])
 api_router.include_router(requirements.router, prefix="/requirements", tags=["requirements"])
+api_router.include_router(urns.router, prefix="/urns", tags=["urn"])

task_registry/api/routes/urns.py

@@ -0,0 +1,27 @@
+import logging
+
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import JSONResponse
+from task_registry.data import Index, TaskType
+from task_registry.lifespan import CACHED_REGISTRY
+
+router = APIRouter()
+
+logger = logging.getLogger(__name__)
+
+
+@router.get(
+    "/{urn}",
+    summary="Get the contents of the specific URN.",
+    description="This endpoint returns a JSON with the contents of a specific URN"
+    " and version.",
+    responses={
+        200: {"description": "JSON with the specific contents of the URN."},
+        400: {"description": "The URN does not exist or is not valid."},
+    },
+)
+async def get_urn(urn: str, version: str = "latest") -> JSONResponse:
+    for taskType in TaskType:
+        if CACHED_REGISTRY.has_urn(urn, taskType):
+            return JSONResponse(content=CACHED_REGISTRY.get_task(urn, taskType))
+    raise HTTPException(status_code=400, detail=f"Unknown urn: {urn}")

task_registry/data.py

@@ -77,6 +77,9 @@ def get_tasks_index(self, tasks: TaskType) -> Index:
     def get_task(self, urn: str, tasks: TaskType) -> dict[str, Any]:
         return self.tasks_cache[(urn, tasks)]
 
+    def has_urn(self, urn: str, tasks: TaskType) -> bool:
+        return (urn, tasks) in self.tasks_cache
+
 
 def generate_index(
     tasks: TaskType,

tests/api/routes/test_urns.py

@@ -0,0 +1,29 @@
+from fastapi.testclient import TestClient
+
+
+def test_get_main_page(client: TestClient) -> None:
+    # when
+    response = client.get("/urns/")
+
+    # then
+    assert response.status_code == 404
+
+
+def test_get_urn(client: TestClient) -> None:
+    # when
+    response = client.get("/urns/test_urn")
+
+    # then
+    assert response.status_code == 200
+    assert response.headers["content-type"] == "application/json"
+    assert b'{"urn":"test_urn"}' in response.content.lower()
+
+
+def test_get_nonexistent_urn(client: TestClient) -> None:
+    # when
+    response = client.get("/urns/idonotexist")
+
+    # then
+    assert response.status_code == 400
+    assert response.headers["content-type"] == "application/json"
+    assert b'{"detail":"unknown urn: idonotexist"}' in response.content.lower()