Merge pull request #210 from lsst-sqre/tickets/DM-45138

DM-45138: Move availability handler into UWS library

src/vocutouts/handlers/external.py

@@ -5,15 +5,12 @@
 application knows the job parameters.
 """
 
-from typing import Annotated
-
-from fastapi import APIRouter, Depends, Request, Response
+from fastapi import APIRouter, Request, Response
 from safir.metadata import get_metadata
 from safir.slack.webhook import SlackRouteErrorHandler
 
 from ..config import config
 from ..models.index import Index
-from ..uws.dependencies import UWSFactory, uws_dependency
 
 router = APIRouter(route_class=SlackRouteErrorHandler)
 """FastAPI router for all external handlers."""
@@ -75,22 +72,6 @@ async def get_index() -> Index:
     return Index(metadata=metadata)
 
 
-@router.get(
-    "/availability",
-    description="VOSI-availability resource for the image cutout service",
-    responses={200: {"content": {"application/xml": {}}}},
-    summary="IVOA service availability",
-)
-async def get_availability(
-    request: Request,
-    uws_factory: Annotated[UWSFactory, Depends(uws_dependency)],
-) -> Response:
-    job_service = uws_factory.create_job_service()
-    availability = await job_service.availability()
-    templates = uws_factory.create_templates()
-    return templates.availability(request, availability)
-
-
 @router.get(
     "/capabilities",
     description="VOSI-capabilities resource for the image cutout service",

src/vocutouts/uws/app.py

@@ -22,6 +22,7 @@
 from .exceptions import UWSError
 from .handlers import (
     install_async_post_handler,
+    install_availability_handler,
     install_sync_get_handler,
     install_sync_post_handler,
     uws_router,
@@ -169,6 +170,7 @@ def install_handlers(self, router: APIRouter) -> None:
         ``/sync`` to create a sync job will be added.
         """
         router.include_router(uws_router, prefix="/jobs")
+        install_availability_handler(router)
         if route := self._config.sync_get_route:
             install_sync_get_handler(router, route)
         if route := self._config.sync_post_route:

src/vocutouts/uws/handlers.py

@@ -37,6 +37,7 @@
 
 __all__ = [
     "install_async_post_handler",
+    "install_availability_handler",
     "install_sync_get_handler",
     "install_sync_post_handler",
     "uws_router",
@@ -476,6 +477,31 @@ async def get_job_results(
     return Response(content=xml, media_type="application/xml")
 
 
+def install_availability_handler(router: APIRouter) -> None:
+    """Construct a default handler for the VOSI ``/availability`` interface.
+
+    Parameters
+    ----------
+    router
+        Router into which to install the handler.
+    """
+
+    @router.get(
+        "/availability",
+        description="VOSI-availability resource for the service",
+        responses={200: {"content": {"application/xml": {}}}},
+        summary="IVOA service availability",
+    )
+    async def get_availability(
+        request: Request,
+        uws_factory: Annotated[UWSFactory, Depends(uws_dependency)],
+    ) -> Response:
+        job_service = uws_factory.create_job_service()
+        availability = await job_service.availability()
+        xml = availability.to_xml(skip_empty=True)
+        return Response(content=xml, media_type="application/xml")
+
+
 def install_async_post_handler(router: APIRouter, route: UWSRoute) -> None:
     """Construct the POST handler for creating an async job.
 

src/vocutouts/uws/models.py

@@ -23,7 +23,6 @@
 
 __all__ = [
     "ACTIVE_PHASES",
-    "Availability",
     "ErrorCode",
     "UWSJob",
     "UWSJobDescription",
@@ -34,17 +33,6 @@
 ]
 
 
-@dataclass
-class Availability:
-    """Availability information (from VOSI)."""
-
-    available: bool
-    """Whether the service appears to be available."""
-
-    note: str | None = None
-    """Supplemental information, usually when the service is not available."""
-
-
 ACTIVE_PHASES = {
     ExecutionPhase.PENDING,
     ExecutionPhase.QUEUED,

src/vocutouts/uws/responses.py

@@ -7,7 +7,7 @@
 from fastapi.templating import Jinja2Templates
 from safir.datetime import isodatetime
 
-from .models import Availability, UWSJob, UWSJobError
+from .models import UWSJob, UWSJobError
 from .results import ResultStore
 
 __all__ = ["UWSTemplates"]
@@ -30,17 +30,6 @@ class UWSTemplates:
     def __init__(self, result_store: ResultStore) -> None:
         self._result_store = result_store
 
-    def availability(
-        self, request: Request, availability: Availability
-    ) -> Response:
-        """Return the availability of a service as an XML response."""
-        return _templates.TemplateResponse(
-            request,
-            "availability.xml",
-            {"availability": availability},
-            media_type="application/xml",
-        )
-
     def error(self, request: Request, error: UWSJobError) -> Response:
         """Return the error of a job as an XML response."""
         return _templates.TemplateResponse(

src/vocutouts/uws/service.py

@@ -9,6 +9,7 @@
 from safir.datetime import current_datetime, isodatetime
 from structlog.stdlib import BoundLogger
 from vo_models.uws.types import ExecutionPhase
+from vo_models.vosi.availability import Availability
 
 from .config import ParametersModel, UWSConfig
 from .constants import JOB_STOP_TIMEOUT
@@ -21,7 +22,6 @@
 )
 from .models import (
     ACTIVE_PHASES,
-    Availability,
     UWSJob,
     UWSJobDescription,
     UWSJobParameter,
@@ -103,13 +103,12 @@ async def abort(self, user: str, job_id: str) -> None:
     async def availability(self) -> Availability:
         """Check whether the service is up.
 
-        Used for ``/availability`` endpoints.  Currently this only checks the
-        database.  Eventually it should push an end-to-end test through the
-        job execution system.
+        Used for ``/availability`` endpoints. Currently this only checks the
+        database.
 
         Returns
         -------
-        vocutouts.uws.models.Availability
+        Availability
             Service availability information.
         """
         return await self._storage.availability()

src/vocutouts/uws/storage.py

@@ -15,10 +15,10 @@
 from sqlalchemy.ext.asyncio import async_scoped_session
 from sqlalchemy.future import select
 from vo_models.uws.types import ErrorType, ExecutionPhase
+from vo_models.vosi.availability import Availability
 
 from .exceptions import TaskError, UnknownJobError
 from .models import (
-    Availability,
     ErrorCode,
     UWSJob,
     UWSJobDescription,

tests/handlers/external_test.py

@@ -8,12 +8,6 @@
 
 from vocutouts.config import config
 
-AVAILABILITY = """
-<vosi:availability xmlns:vosi="http://www.ivoa.net/xml/VOSIAvailability/v1.0">
-  <vosi:available>true</vosi:available>
-</vosi:availability>
-"""
-
 CAPABILITIES = """
 <?xml version="1.0"?>
 <capabilities
@@ -60,13 +54,6 @@ async def test_get_index(client: AsyncClient) -> None:
     assert isinstance(metadata["documentation_url"], str)
 
 
-@pytest.mark.asyncio
-async def test_availability(client: AsyncClient) -> None:
-    r = await client.get("/api/cutout/availability")
-    assert r.status_code == 200
-    assert r.text == AVAILABILITY.strip()
-
-
 @pytest.mark.asyncio
 async def test_capabilities(client: AsyncClient) -> None:
     r = await client.get("/api/cutout/capabilities")

tests/uws/vosi_test.py

@@ -0,0 +1,20 @@
+"""Tests for VOSI functionality provided by the UWS library."""
+
+from __future__ import annotations
+
+import pytest
+from httpx import AsyncClient
+from vo_models.vosi.availability import Availability
+
+AVAILABILITY = """
+<vosi:availability xmlns:vosi="http://www.ivoa.net/xml/VOSIAvailability/v1.0">
+  <vosi:available>true</vosi:available>
+</vosi:availability>
+"""
+
+
+@pytest.mark.asyncio
+async def test_availability(client: AsyncClient) -> None:
+    r = await client.get("/test/availability")
+    assert r.status_code == 200
+    assert Availability.from_xml(r.text) == Availability.from_xml(AVAILABILITY)