feat(api): Implement v2 campaign graph api

Author: tcjennings

src/lsst/cmservice/common/graph.py

@@ -1,4 +1,5 @@
 from collections.abc import Iterable, Mapping, MutableSet, Sequence
+from typing import Literal
 
 import networkx as nx
 from sqlalchemy import select
@@ -39,14 +40,31 @@ async def graph_from_edge_list(
 
 async def graph_from_edge_list_v2(
     edges: Sequence[Edge],
-    node_type: type[Node],
     session: AnyAsyncSession,
+    node_type: type[Node] = Node,
+    node_view: Literal["simple", "model"] = "model",
 ) -> nx.DiGraph:
     """Given a sequence of Edges, create a directed graph for these
     edges with nodes derived from database lookups of the related objects.
+
+    Parameters
+    ----------
+    edges: Sequence[Edge]
+        The list of edges forming the graph
+
+    node_type: type
+        The pydantic or sqlmodel class representing the graph node model
+
+    node_view: "simple" or "model"
+        Whether the node metadata in the graph should be simplified (dict) or
+        using the full expunged model form.
+
+    session
+        An async database session
     """
     g = nx.DiGraph()
     g.add_edges_from([(e.source, e.target) for e in edges])
+    relabel_mapping = {}
 
     # The graph understands the nodes in terms of the IDs used in the edges,
     # but we want to hydrate the entire Node model for subsequent users of this
@@ -59,7 +77,19 @@ async def graph_from_edge_list_v2(
         # SQLAlchemy baggage along, so we expunge it from the session before
         # adding it to the graph.
         session.expunge(db_node)
-        g.nodes[node]["model"] = db_node
+        if node_view == "simple":
+            # for the simple node view, the goal is to minimize the amount of
+            # data attached to the node and ensure that this data is json-
+            # serializable and otherwise appropriate for an API response
+            g.nodes[node]["id"] = str(db_node.id)
+            g.nodes[node]["status"] = db_node.status.name
+            g.nodes[node]["kind"] = db_node.kind.name
+            relabel_mapping[node] = db_node.name
+        else:
+            g.nodes[node]["model"] = db_node
+
+    if relabel_mapping:
+        g = nx.relabel_nodes(g, mapping=relabel_mapping, copy=False)
 
     # TODO validate graph now raise exception, or leave it to the caller?
     return g
@@ -68,6 +98,11 @@ async def graph_from_edge_list_v2(
 def graph_to_dict(g: nx.DiGraph) -> Mapping:
     """Renders a networkx directed graph to a mapping format suitable for JSON
     serialization.
+
+    Notes
+    -----
+    The "edges" attribute name in the node link data is "edges" instead of the
+    default "links".
     """
     return nx.node_link_data(g, edges="edges")
 

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

@@ -4,7 +4,7 @@
 representing campaign objects within CM-Service.
 """
 
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Annotated
 from uuid import UUID, uuid5
 
@@ -13,6 +13,7 @@
 from sqlmodel import col, select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
+from ...common.graph import graph_from_edge_list_v2, graph_to_dict
 from ...common.logging import LOGGER
 from ...db.campaigns_v2 import Campaign, CampaignUpdate, Edge, Node
 from ...db.manifests_v2 import CampaignManifest
@@ -40,7 +41,9 @@ async def read_campaign_collection(
     limit: Annotated[int, Query(le=100)] = 10,
     offset: Annotated[int, Query()] = 0,
 ) -> Sequence[Campaign]:
-    """..."""
+    """A paginated API returning a list of all Campaigns known to the
+    application.
+    """
     try:
         campaigns = await session.exec(select(Campaign).offset(offset).limit(limit))
 
@@ -187,7 +190,9 @@ async def read_campaign_node_collection(
     limit: Annotated[int, Query(le=100)] = 10,
     offset: Annotated[int, Query()] = 0,
 ) -> Sequence[Node]:
-    # This is a convenience api that could also be `/nodes?campaign=...
+    """A paginated API returning a list of all Nodes in the namespace of a
+    single Campaign.
+    """
 
     # The input could be a campaign UUID or it could be a literal name.
     # TODO this could just as well be a campaign query with a join to nodes
@@ -222,7 +227,10 @@ async def read_campaign_edge_collection(
     *,
     resolve_names: bool = False,
 ) -> Sequence[Edge]:
-    # This is a convenience api that could also be `/edges?campaign=...
+    """A paginated API returning a list of all Edges in the namespace of a
+    single Campaign. This list of Edges can be used to construct the Campaign
+    graph.
+    """
 
     # The input could be a campaign UUID or it could be a literal name.
     # This is why raw SQL is better than ORMs
@@ -301,6 +309,7 @@ async def create_campaign_resource(
     session: Annotated[AsyncSession, Depends(db_session_dependency)],
     manifest: CampaignManifest,
 ) -> Campaign:
+    """An API to create a Campaign from an appropriate Manifest."""
     # Create a campaign spec from the manifest, delegating the creation of new
     # dynamic fields to the model validation method, -OR- create new dynamic
     # fields here.
@@ -333,3 +342,46 @@ async def create_campaign_resource(
     )
 
     return campaign
+
+
+@router.get(
+    "/{campaign_name_or_id}/graph",
+    status_code=200,
+    summary="Construct and return a Campaign's graph of nodes",
+)
+async def read_campaign_graph(
+    request: Request,
+    response: Response,
+    campaign_name_or_id: str,
+    session: Annotated[AsyncSession, Depends(db_session_dependency)],
+) -> Mapping:
+    """Reads the graph resource for a campaign and returns its JSON represent-
+    ation as serialized by the ``networkx.node_link_data()` function, i.e, the
+    "node-link format".
+    """
+
+    # The input could be a campaign UUID or it could be a literal name.
+    campaign_id: UUID | None
+    try:
+        campaign_id = UUID(campaign_name_or_id)
+    except ValueError:
+        s = select(Campaign.id).where(Campaign.name == campaign_name_or_id)
+        campaign_id = (await session.exec(s)).one_or_none()
+
+    if campaign_id is None:
+        raise HTTPException(status_code=404, detail="No such campaign found.")
+
+    # Fetch the Edges for the campaign
+    statement = select(Edge).filter_by(namespace=campaign_id)
+    edges = (await session.exec(statement)).all()
+
+    # Organize the edges into a graph. The graph nodes are annotated with their
+    # current database attributes.
+    # TODO it makes sense for the graph to include expunged Nodes in the meta-
+    # data for campaign processing, but for the purposes of this api route,
+    # only the most relevant information should be associated with each node,
+    # e.g., its name, status, id, and its URL
+    graph = await graph_from_edge_list_v2(edges=edges, node_type=Node, session=session, node_view="simple")
+
+    response.headers["Self"] = ""
+    return graph_to_dict(graph)

tests/v2/test_graph.py

@@ -11,7 +11,7 @@
 from lsst.cmservice.common.enums import StatusEnum
 from lsst.cmservice.common.graph import graph_from_edge_list_v2, processable_graph_nodes, validate_graph
 from lsst.cmservice.common.types import AnyAsyncSession
-from lsst.cmservice.db.campaigns_v2 import Edge, Node
+from lsst.cmservice.db.campaigns_v2 import Edge
 
 pytestmark = pytest.mark.asyncio(loop_scope="module")
 """All tests in this module will run in the same event loop."""
@@ -132,7 +132,7 @@ async def test_build_and_walk_graph(
     ```
     """
     edge_list = [Edge.model_validate(edge) for edge in (await aclient.get(test_campaign)).json()]
-    graph = await graph_from_edge_list_v2(edge_list, Node, session)
+    graph = await graph_from_edge_list_v2(edge_list, session)
 
     # the START node should be the only processable Node
     for node in processable_graph_nodes(graph):
@@ -146,7 +146,7 @@ async def test_build_and_walk_graph(
 
     # Repeat the graph building and traversal, this time expecting a single
     # node that is not "START"
-    graph = await graph_from_edge_list_v2(edge_list, Node, session)
+    graph = await graph_from_edge_list_v2(edge_list, session)
     for node in processable_graph_nodes(graph):
         assert node.name != "START"
         assert node.status is StatusEnum.waiting
@@ -158,7 +158,7 @@ async def test_build_and_walk_graph(
 
     # Repeat the graph building and traversal, this time expecting a pair of
     # nodes processable in parallel
-    graph = await graph_from_edge_list_v2(edge_list, Node, session)
+    graph = await graph_from_edge_list_v2(edge_list, session)
     count = 0
     for node in processable_graph_nodes(graph):
         count += 1
@@ -172,7 +172,7 @@ async def test_build_and_walk_graph(
     assert count == 2
 
     # Finally, expect the END node
-    graph = await graph_from_edge_list_v2(edge_list, Node, session)
+    graph = await graph_from_edge_list_v2(edge_list, session)
     for node in processable_graph_nodes(graph):
         assert node.name == "END"
         assert node.status is StatusEnum.waiting
@@ -215,3 +215,16 @@ def test_validate_graph() -> None:
     # remove the unneeded node
     g.remove_node("CC")
     assert validate_graph(g, "A", "F")
+
+
+async def test_campaign_graph_route(aclient: AsyncClient, test_campaign: str) -> None:
+    """Tests the acquisition of a serialized graph from a REST endpoint and
+    the subsequent reconstruction of a valid graph from the node-link data.
+    """
+    graph_url = test_campaign.replace("/edges", "/graph")
+    x = await aclient.get(graph_url)
+    assert x.is_success
+
+    # Test reconstruction of the serialized graph
+    graph = nx.node_link_graph(x.json(), edges="edges")
+    assert validate_graph(graph)