feat(workflows)!: add FastAPI router factory + WDL dir to WorkflowSet

bento_lib/workflows/fastapi.py

@@ -0,0 +1,40 @@
+from fastapi import status
+from fastapi.exceptions import HTTPException
+from fastapi.responses import FileResponse
+from fastapi.routing import APIRouter
+
+from ..auth.middleware.fastapi import FastApiAuthMiddleware
+from .models import WorkflowDefinition
+from .workflow_set import WorkflowSet
+
+__all__ = [
+    "build_workflow_router",
+]
+
+
+def build_workflow_router(authz_middleware: FastApiAuthMiddleware, workflow_set: WorkflowSet) -> APIRouter:
+    workflow_router = APIRouter(prefix="/workflows")
+
+    @workflow_router.get("", dependencies=[authz_middleware.dep_public_endpoint()])
+    def workflow_list():
+        return workflow_set.workflow_dicts_by_type_and_id()
+
+    @workflow_router.get("/", dependencies=[authz_middleware.dep_public_endpoint()])
+    def workflow_list_trailing_slash():
+        return workflow_set.workflow_dicts_by_type_and_id()
+
+    # The endpoint with the .wdl suffix needs to come first, since we match in order and otherwise the whole thing,
+    # including the suffix, would get passed as {workflow_id}.
+    @workflow_router.get("/{workflow_id}.wdl", dependencies=[authz_middleware.dep_public_endpoint()])
+    def workflow_file(workflow_id: str):
+        if (wdl := workflow_set.get_workflow_wdl_path(workflow_id)) is not None:
+            return FileResponse(wdl)
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"No workflow with ID {workflow_id}")
+
+    @workflow_router.get("/{workflow_id}", dependencies=[authz_middleware.dep_public_endpoint()])
+    def workflow_item(workflow_id: str) -> WorkflowDefinition:
+        if (wf := workflow_set.get_workflow(workflow_id)) is not None:
+            return wf
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"No workflow with ID {workflow_id}")
+
+    return workflow_router

bento_lib/workflows/workflow_set.py

@@ -1,5 +1,6 @@
 import werkzeug.utils
 from collections import defaultdict
+from pathlib import Path
 from .models import WorkflowDefinition
 
 __all__ = [
@@ -12,8 +13,9 @@ class WorkflowSet:
     A class for constructing a singleton object that stores all workflow descriptions in a particular service.
     """
 
-    def __init__(self):
+    def __init__(self, wdl_directory: Path):
         self._defs_by_id: dict[str, WorkflowDefinition] = {}
+        self._wdl_directory: Path = wdl_directory
 
     def add_workflow(self, id_: str, definition: WorkflowDefinition):
         if id_ in self._defs_by_id:
@@ -29,6 +31,11 @@ def get_workflow_resource(self, id_: str) -> str | None:
             return werkzeug.utils.secure_filename(wd.file)
         return None
 
+    def get_workflow_wdl_path(self, id_: str) -> Path | None:
+        if (wfr := self.get_workflow_resource(id_)) is not None:
+            return self._wdl_directory / wfr
+        return None
+
     def workflow_exists(self, id_: str) -> bool:
         return id_ in self._defs_by_id