WIP: nodes and edges

Author: tcjennings

src/lsst/cmservice/routers/v2/nodes.py

@@ -5,14 +5,15 @@
 """
 
 from collections.abc import Sequence
-from typing import Annotated
+from typing import TYPE_CHECKING, Annotated
 from uuid import UUID, uuid5
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
 from sqlmodel import col, select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from ...common.enums import ManifestKind
+from ...common.jsonpatch import JSONPatch, JSONPatchError, apply_json_patch
 from ...common.logging import LOGGER
 from ...db.campaigns_v2 import Campaign, Node
 from ...db.manifests_v2 import ManifestWrapper
@@ -33,18 +34,31 @@
     "/",
     summary="Get a list of nodes",
 )
-async def read_node_collection(
+async def read_nodes_collection(
     request: Request,
     response: Response,
     session: Annotated[AsyncSession, Depends(db_session_dependency)],
     offset: Annotated[int, Query(ge=0)] = 0,
     limit: Annotated[int, Query(le=100)] = 10,
 ) -> Sequence[Node]:
-    """Gets all nodes"""
-    # TODO add paginated links to response header
-    response.headers["Link"] = ""
+    """Fetches and returns all nodes known to the service.
+
+    Notes
+    -----
+    For campaign-scoped nodes, one should use the /campaigns/{}/nodes route.
+    """
     try:
         nodes = await session.exec(select(Node).offset(offset).limit(limit))
+        response.headers["Next"] = (
+            request.url_for("read_nodes_collection")
+            .include_query_params(offset=(offset + limit), limit=limit)
+            .__str__()
+        )
+        response.headers["Previous"] = (
+            request.url_for("read_nodes_collection")
+            .include_query_params(offset=(offset - limit), limit=limit)
+            .__str__()
+        )
         return nodes.all()
     except Exception as msg:
         logger.exception()
@@ -72,10 +86,14 @@ async def read_node_resource(
     except ValueError:
         s = s.where(Node.name == node_name)
 
-    campaign = (await session.exec(s)).one_or_none()
-    if campaign is None:
+    node = (await session.exec(s)).one_or_none()
+    if node is None:
         raise HTTPException(status_code=404)
-    return campaign
+    response.headers["Self"] = request.url_for("read_node_resource", node_name=node.id).__str__()
+    response.headers["Campaign"] = request.url_for(
+        "read_campaign_resource", campaign_name=node.namespace
+    ).__str__()
+    return node
 
 
 @router.post(
@@ -88,26 +106,19 @@ async def create_node_resource(
     manifest: ManifestWrapper,
     session: Annotated[AsyncSession, Depends(db_session_dependency)],
 ) -> Node:
-    # TODO should support query parameters that scope the namespace, such that
-    #      response headers from a campaign-create operation can immediately
-    #      follow a link to node-create for that campaign.
-
     # Validate the input by checking the "kind" of manifest is a node
     if manifest.kind is not ManifestKind.node:
         raise HTTPException(status_code=422, detail="Nodes may only be created from a 'node' manifest")
     # and that the manifest includes any required fields, though this could
     # just as well be a try/except ValueError around `_.model_validate()`
-    elif (node_name := manifest.metadata_.pop("name")) is None:
+    elif (node_name := manifest.metadata_.pop("name", None)) is None:
         raise HTTPException(status_code=400, detail="Nodes must have a name set in '.metadata.name'")
 
-    # A node's spec must be a valid node spec
-    # TODO match node with jsonschema and validate
-
     # A node must exist in the namespace of an existing campaign
-    node_namespace: str = manifest.metadata_.pop("namespace")
+    node_namespace: str = manifest.metadata_.pop("namespace", None)
     if node_namespace is None:
         raise HTTPException(
-            status_code=400, detail="Nodes must have a namespace set in '.metadata.namespace'"
+            status_code=400, detail="Nodes must have a campaign namespace set in '.metadata.namespace'"
         )
 
     try:
@@ -150,4 +161,89 @@ async def create_node_resource(
     session.add(node)
     await session.commit()
     await session.refresh(node)
+    response.headers["Self"] = request.url_for("read_node_resource", node_name=node.id).__str__()
+    response.headers["Campaign"] = request.url_for(
+        "read_campaign_resource", campaign_name=node.namespace
+    ).__str__()
     return node
+
+
+@router.patch(
+    "/{node_name_or_id}",
+    summary="Update node detail",
+    status_code=202,
+)
+async def update_node_resource(
+    request: Request,
+    response: Response,
+    session: Annotated[AsyncSession, Depends(db_session_dependency)],
+    node_name_or_id: str,
+    patch_data: Sequence[JSONPatch],
+) -> Node:
+    """Partial update method for nodes.
+
+    A Nodes's spec or metadata may be updated with this PATCH operation. All
+    updates to a Node creates a new version of the Node instead of
+    updating an existing record in-place. This preserves history and keeps
+    previous node versions available.
+
+    A Node's name, id, kind, or namespace may not be modified by this
+    method, and attempts to do so will produce a 4XX client error.
+
+    This PATCH endpoint supports RFC6902 json-patch and RFCrequests.
+
+    Notes
+    -----
+    - This API always targets the latest version of a manifest when applying
+      a patch. This requires and maintains a "linear" sequence of versions;
+      it is not permissible to "patch" a previous version and create a "tree"-
+      like history of manifests. For exmaple, every manifest may be diffed
+      against any previous version without having to consider branches.
+    """
+    use_rfc6902 = False
+    if request.headers["Content-Type"] == "application/json-patch+json":
+        use_rfc6902 = True
+    else:
+        raise HTTPException(status_code=406, detail="Unsupported Content-Type")
+
+    if TYPE_CHECKING:
+        assert use_rfc6902
+
+    s = select(Node)
+    # The input could be a UUID or it could be a literal name.
+    try:
+        if _id := UUID(node_name_or_id):
+            s = s.where(Node.id == _id)
+    except ValueError:
+        s = s.where(Node.name == node_name_or_id)
+
+    # we want to order and sort by version, in descending order, so we always
+    # fetch only the most recent version of manifest
+    # FIXME this implies that when a node ID is provided, it should be an
+    # error if it is not the most recent version.
+    s = s.order_by(col(Node.version).desc()).limit(1)
+
+    old_manifest = (await session.exec(s)).one_or_none()
+    if old_manifest is None:
+        raise HTTPException(status_code=404, detail="No such node")
+
+    new_manifest = old_manifest.model_dump(by_alias=True)
+    new_manifest["version"] += 1
+    new_manifest["id"] = uuid5(new_manifest["namespace"], f"{new_manifest['name']}.{new_manifest['version']}")
+
+    for patch in patch_data:
+        try:
+            apply_json_patch(patch, new_manifest)
+        except JSONPatchError as e:
+            raise HTTPException(
+                status_code=422,
+                detail=f"Unable to process one or more patch operations: {e}",
+            )
+
+    # create Manifest from new_manifest, add to session, and commit
+    new_manifest_db = Node.model_validate(new_manifest)
+    session.add(new_manifest_db)
+    await session.commit()
+
+    # TODO response headers
+    return new_manifest_db

tests/v2/test_edge_routes.py

@@ -1,4 +1,4 @@
-"""Tests v2 fastapi manifest routes"""
+"""Tests v2 fastapi edge routes"""
 
 from uuid import uuid4
 

tests/v2/test_node_routes.py

@@ -0,0 +1,128 @@
+"""Tests v2 fastapi node routes"""
+
+from uuid import uuid4
+
+import pytest
+from httpx import AsyncClient
+
+pytestmark = pytest.mark.asyncio(loop_scope="module")
+"""All tests in this module will run in the same event loop."""
+
+
+async def test_list_delete_no_nodes(aclient: AsyncClient) -> None:
+    """Tests listing nodes when there are no nodes available."""
+    x = await aclient.get("/cm-service/v2/nodes")
+    assert x.is_success
+    assert len(x.json()) == 0
+
+    x = await aclient.get(f"/cm-service/v2/nodes/{uuid4()}")
+    assert x.status_code == 404
+
+    # there is no delete api for nodes
+    x = await aclient.delete(f"/cm-service/v2/nodes/{uuid4()}")
+    assert x.status_code == 405
+
+
+async def test_node_negative(aclient: AsyncClient) -> None:
+    """Tests node route negative outcomes."""
+
+    # negative test: failing request model validation
+    x = await aclient.post(
+        "cm-service/v2/nodes",
+        json={
+            "kind": "no_such_kind",
+        },
+    )
+    assert x.is_client_error
+
+    # negative test: using wrong manifest kind
+    x = await aclient.post(
+        "cm-service/v2/nodes",
+        json={
+            "kind": "other",
+        },
+    )
+    assert x.is_client_error
+
+    # negative test: missing node name
+    x = await aclient.post(
+        "cm-service/v2/nodes",
+        json={
+            "kind": "node",
+            "metadata": {},
+            "spec": {},
+        },
+    )
+    assert x.is_client_error
+
+    # negative test: missing campaign namespace
+    x = await aclient.post(
+        "cm-service/v2/nodes",
+        json={
+            "kind": "node",
+            "metadata": {"name": uuid4().hex[8:]},
+            "spec": {},
+        },
+    )
+    assert x.is_client_error
+    assert "campaign namespace" in x.text
+
+
+async def test_node_lifecycle(aclient: AsyncClient) -> None:
+    """Tests node lifecycle."""
+    campaign_name = uuid4().hex[:8]
+
+    # Create a campaign for edges. Campaigns come with START and END nodes.
+    x = await aclient.post(
+        "/cm-service/v2/campaigns",
+        json={
+            "kind": "campaign",
+            "metadata": {"name": campaign_name},
+        },
+    )
+    assert x.is_success
+    campaign_id = x.json()["id"]
+
+    # Create a new campaign node
+    x = await aclient.post(
+        "/cm-service/v2/nodes",
+        json={
+            "kind": "node",
+            "metadata": {"name": uuid4().hex[8:], "namespace": campaign_id},
+            "spec": {
+                "handler": "lsst.cmservice.handlers.element_handler.ElementHandler",
+                "pipeline_yaml": "${DRP_PIPE_DIR}/pipelines/HSC/DRP-RC2.yaml#step1",
+                "query": [{"instrument": "HSC"}, {"skymap": "hsc_rings_v1"}],
+                "split": {
+                    "dataset": "raw",
+                    "field": "exposure",
+                    "method": "query",
+                    "minGroups": 3,
+                    "maxInFlight": 3,
+                },
+            },
+        },
+    )
+    assert x.is_success
+    node = x.json()
+    assert node["version"] == 1
+    node_url = x.headers["Self"]
+
+    # Edit a Node using RFC6902 json-patch
+    x = await aclient.patch(
+        node_url,
+        headers={"Content-Type": "application/json-patch+json"},
+        json=[
+            {"op": "replace", "path": "/configuration/split/maxInFlight", "value": 4},
+            {
+                "op": "add",
+                "path": "/configuration/query/-",
+                "value": {"dimension": "fact"},
+            },
+        ],
+    )
+    assert x.is_success
+    node = x.json()
+
+    # Edit a Node using RFCxxxx json-merge
+    ...