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

DM-45138: Refactor phase support for async jobs

changelog.d/20240715_155338_rra_DM_45138.md

@@ -0,0 +1,3 @@
+### Bug fixes
+
+- Restore support for automatically starting an async job by setting `phase=RUN` in the POST body. The equivalent query parameter was always supported, but POST body support was accidentally dropped in 3.0.0.

src/vocutouts/uws/dependencies.py

@@ -7,16 +7,17 @@
 """
 
 from collections.abc import AsyncIterator
-from typing import Annotated
+from typing import Annotated, Literal
 
-from fastapi import Depends, Form, Request
+from fastapi import Depends, Form, Query, Request
 from safir.arq import ArqMode, ArqQueue, MockArqQueue, RedisArqQueue
 from safir.database import create_async_session, create_database_engine
 from safir.dependencies.logger import logger_dependency
 from sqlalchemy.ext.asyncio import AsyncEngine, async_scoped_session
 from structlog.stdlib import BoundLogger
 
 from .config import UWSConfig
+from .exceptions import ParameterError
 from .models import UWSJobParameter
 from .responses import UWSTemplates
 from .results import ResultStore
@@ -26,6 +27,7 @@
 __all__ = [
     "UWSDependency",
     "UWSFactory",
+    "create_phase_dependency",
     "runid_post_dependency",
     "uws_dependency",
     "uws_post_params_dependency",
@@ -191,6 +193,34 @@ async def uws_post_params_dependency(
     return parameters
 
 
+async def create_phase_dependency(
+    *,
+    get_phase: Annotated[
+        Literal["RUN"] | None,
+        Query(title="Immediately start job", alias="phase"),
+    ] = None,
+    post_phase: Annotated[
+        Literal["RUN"] | None,
+        Form(title="Immediately start job", alias="phase"),
+    ] = None,
+    params: Annotated[
+        list[UWSJobParameter], Depends(uws_post_params_dependency)
+    ],
+) -> Literal["RUN"] | None:
+    """Parse the optional phase parameter to an async job creation.
+
+    Allow ``phase=RUN`` to be specified in either the query or the POST
+    parameters, which says that the job should be immediately started.
+    """
+    for param in params:
+        if param.parameter_id != "phase":
+            continue
+        if param.value != "RUN":
+            raise ParameterError(f"Invalid phase {param.value}")
+        return "RUN"
+    return get_phase
+
+
 async def runid_post_dependency(
     *,
     runid: Annotated[

src/vocutouts/uws/handlers.py

@@ -25,6 +25,7 @@
 from .config import UWSRoute
 from .dependencies import (
     UWSFactory,
+    create_phase_dependency,
     runid_post_dependency,
     uws_dependency,
     uws_post_params_dependency,
@@ -524,7 +525,7 @@ async def create_job(
         *,
         request: Request,
         phase: Annotated[
-            Literal["RUN"] | None, Query(title="Immediately start job")
+            Literal["RUN"] | None, Depends(create_phase_dependency)
         ] = None,
         parameters: Annotated[
             list[UWSJobParameter], Depends(route.dependency)

tests/uws/errors_test.py

@@ -144,5 +144,19 @@ async def test_errors(
         assert r.status_code == 422
         assert r.text.startswith("UsageError")
 
+    # Test bogus phase for async job creation.
+    r = await client.post(
+        "/test/jobs?phase=START",
+        headers={"X-Auth-Request-User": "user"},
+        data={"runid": "some-run-id", "name": "Jane"},
+    )
+    assert r.status_code == 422
+    r = await client.post(
+        "/test/jobs",
+        headers={"X-Auth-Request-User": "user"},
+        data={"runid": "some-run-id", "name": "Jane", "phase": "START"},
+    )
+    assert r.status_code == 422
+
     # None of these errors should have produced Slack errors.
     assert mock_slack.messages == []

tests/uws/job_api_test.py

@@ -309,16 +309,10 @@ async def test_job_abort(
     r = await client.post(
         "/test/jobs",
         headers={"X-Auth-Request-User": "user"},
-        data={"runid": "some-run-id", "name": "Jane"},
+        data={"runid": "some-run-id", "name": "Jane", "phase": "RUN"},
     )
     assert r.status_code == 303
     assert r.headers["Location"] == "https://example.com/test/jobs/2"
-    r = await client.post(
-        "/test/jobs/2/phase",
-        headers={"X-Auth-Request-User": "user"},
-        data={"PHASE": "RUN"},
-    )
-    assert r.status_code == 303
     await runner.mark_in_progress("user", "2")
 
     # Abort that job.
@@ -571,21 +565,12 @@ async def test_presigned_url(
 
     # Create the job.
     r = await client.post(
-        "/test/jobs",
+        "/test/jobs?phase=RUN",
         headers={"X-Auth-Request-User": "user"},
         data={"runid": "some-run-id", "name": "Jane"},
     )
     assert r.status_code == 303
     job = await job_service.get("user", "1")
-
-    # Start the job.
-    r = await client.post(
-        "/test/jobs/1/phase",
-        headers={"X-Auth-Request-User": "user"},
-        data={"PHASE": "RUN"},
-        follow_redirects=True,
-    )
-    assert r.status_code == 200
     await runner.mark_in_progress("user", "1")
 
     # Tell the queue the job is finished, with an https URL.