feat(api): Add edge and node apis

feat(api): Refactor node and edge models

Author: tcjennings

src/lsst/cmservice/db/campaigns_v2.py

@@ -1,8 +1,10 @@
+"""ORM Models for v2 tables and objects."""
+
 from datetime import datetime
 from typing import Any
 from uuid import NAMESPACE_DNS, UUID, uuid5
 
-from pydantic import AliasChoices, ValidationInfo, model_validator
+from pydantic import AliasChoices
 from sqlalchemy.dialects import postgresql
 from sqlalchemy.ext.mutable import MutableDict, MutableList
 from sqlalchemy.types import PickleType
@@ -43,15 +45,6 @@ def jsonb_column(name: str, aliases: list[str] | None = None) -> Any:
     )
 
 
-# NOTES
-# - model validation is not triggered when table=True
-# - Every object model needs to have three flavors:
-#   1. the declarative model of the object's database table
-#   2. the model of the manifest when creating a new object
-#   3. the model of the manifest when updating an object
-#   4. a response model for APIs related to the object
-
-
 class BaseSQLModel(SQLModel):
     __table_args__ = {"schema": config.db.table_schema}
     metadata = metadata
@@ -72,26 +65,7 @@ class CampaignBase(BaseSQLModel):
     configuration: dict = jsonb_column("configuration", aliases=["configuration", "data", "spec"])
 
 
-class CampaignModel(CampaignBase):
-    """model used for resource creation."""
-
-    @model_validator(mode="before")
-    @classmethod
-    def custom_model_validator(cls, data: Any, info: ValidationInfo) -> Any:
-        """Validates the model based on different types of raw inputs,
-        where some default non-optional fields can be auto-populated.
-        """
-        if isinstance(data, dict):
-            if "name" not in data:
-                raise ValueError("'name' must be specified.")
-            if "namespace" not in data:
-                data["namespace"] = _default_campaign_namespace
-            if "id" not in data:
-                data["id"] = uuid5(namespace=data["namespace"], name=data["name"])
-        return data
-
-
-class Campaign(CampaignModel, table=True):
+class Campaign(CampaignBase, table=True):
     """Model used for database operations involving campaigns_v2 table rows"""
 
     __tablename__: str = "campaigns_v2"  # type: ignore[misc]
@@ -127,25 +101,7 @@ class NodeBase(BaseSQLModel):
     configuration: dict = jsonb_column("configuration", aliases=["configuration", "data", "spec"])
 
 
-class NodeModel(NodeBase):
-    """model validating class for Nodes"""
-
-    @model_validator(mode="before")
-    @classmethod
-    def custom_model_validator(cls, data: Any, info: ValidationInfo) -> Any:
-        if isinstance(data, dict):
-            if "version" not in data:
-                data["version"] = 1
-            if "name" not in data:
-                raise ValueError("'name' must be specified.")
-            if "namespace" not in data:
-                data["namespace"] = _default_campaign_namespace
-            if "id" not in data:
-                data["id"] = uuid5(namespace=data["namespace"], name=f"""{data["name"]}.{data["version"]}""")
-        return data
-
-
-class Node(NodeModel, table=True):
+class Node(NodeBase, table=True):
     __tablename__: str = "nodes_v2"  # type: ignore[misc]
 
     machine: UUID | None = Field(foreign_key="machines_v2.id", default=None, ondelete="CASCADE")
@@ -163,28 +119,12 @@ class EdgeBase(BaseSQLModel):
     configuration: dict = jsonb_column("configuration", aliases=["configuration", "data", "spec"])
 
 
-class EdgeModel(EdgeBase):
-    """model validating class for Edges"""
-
-    @model_validator(mode="before")
-    @classmethod
-    def custom_model_validator(cls, data: Any, info: ValidationInfo) -> Any:
-        if isinstance(data, dict):
-            if "name" not in data:
-                raise ValueError("'name' must be specified.")
-            if "namespace" not in data:
-                raise ValueError("Edges may only exist in a 'namespace'.")
-            if "id" not in data:
-                data["id"] = uuid5(namespace=data["namespace"], name=data["name"])
-        return data
-
-
-class EdgeResponseModel(EdgeModel):
+class EdgeResponseModel(EdgeBase):
     source: Any
     target: Any
 
 
-class Edge(EdgeModel, table=True):
+class Edge(EdgeBase, table=True):
     __tablename__: str = "edges_v2"  # type: ignore[misc]
 
 
@@ -216,24 +156,6 @@ class ManifestBase(BaseSQLModel):
     spec: dict = jsonb_column("spec", aliases=["spec", "configuration", "data"])
 
 
-class ManifestModel(ManifestBase):
-    """model validating class for Manifests"""
-
-    @model_validator(mode="before")
-    @classmethod
-    def custom_model_validator(cls, data: Any, info: ValidationInfo) -> Any:
-        if isinstance(data, dict):
-            if "version" not in data:
-                data["version"] = 1
-            if "name" not in data:
-                raise ValueError("'name' must be specified.")
-            if "namespace" not in data:
-                data["namespace"] = _default_campaign_namespace
-            if "id" not in data:
-                data["id"] = uuid5(namespace=data["namespace"], name=f"""{data["name"]}.{data["version"]}""")
-        return data
-
-
 class Manifest(ManifestBase, table=True):
     __tablename__: str = "manifests_v2"  # type: ignore[misc]
 

src/lsst/cmservice/db/manifests_v2.py

@@ -5,6 +5,7 @@
 """
 
 from typing import Self
+from uuid import uuid4
 
 from pydantic import AliasChoices, BaseModel, ConfigDict, Field, ValidationInfo, model_validator
 
@@ -99,3 +100,43 @@ def custom_model_validator(self, info: ValidationInfo) -> Self:
             raise ValueError("Campaigns may only be created from a <campaign> manifest")
 
         return self
+
+
+class EdgeMetadata(ManifestMetadata):
+    """Metadata model for an Edge Manifest.
+
+    A default random alphanumeric 8-byte name is generated if no name provided.
+    """
+
+    name: str = Field(default_factory=lambda: uuid4().hex[:8])
+
+
+class EdgeSpec(ManifestSpec):
+    """Spec model for an Edge Manifest."""
+
+    source: str
+    target: str
+
+
+class EdgeManifest(Manifest[EdgeMetadata, EdgeSpec]):
+    """validating model for Edges"""
+
+    @model_validator(mode="after")
+    def custom_model_validator(self, info: ValidationInfo) -> Self:
+        """Validate an Edge Manifest after a model has been created."""
+        if self.kind is not ManifestKind.edge:
+            raise ValueError("Edges may only be created from an <edge> manifest")
+
+        return self
+
+
+class NodeManifest(Manifest[VersionedMetadata, ManifestSpec]):
+    """validating model for Nodes"""
+
+    @model_validator(mode="after")
+    def custom_model_validator(self, info: ValidationInfo) -> Self:
+        """Validate a Node Manifest after a model has been created."""
+        if self.kind is not ManifestKind.node:
+            raise ValueError("Nodes may only be created from an <node> manifest")
+
+        return self

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

@@ -2,12 +2,16 @@
 
 from . import (
     campaigns,
+    edges,
     manifests,
+    nodes,
 )
 
 router = APIRouter(
     prefix="/v2",
 )
 
 router.include_router(campaigns.router)
+router.include_router(edges.router)
 router.include_router(manifests.router)
+router.include_router(nodes.router)