refactor: Change _elkjs typed dicts to pydantic elements

.github/workflows/lint.yml

@@ -39,4 +39,4 @@ jobs:
           python -m pip install pylint==2.13.9
       - name: Run pylint
         run: |-
-          pylint -dfixme -- capellambse_context_diagrams || exit $(($? & ~24))
+          pylint -dfixme -- capellambse_context_diagrams tests || exit $(($? & ~24))

capellambse_context_diagrams/_elkjs.py

@@ -19,7 +19,7 @@
 from pathlib import Path
 
 import capellambse
-import typing_extensions as te
+import pydantic
 
 __all__ = [
     "call_elkjs",
@@ -89,121 +89,144 @@
 """Options for labels to configure ELK layouting."""
 
 
-class ELKInputData(te.TypedDict, total=False):
+class BaseELKModel(pydantic.BaseModel):
+    """Base class for ELK models."""
+
+    model_config = pydantic.ConfigDict(
+        extra="allow",
+        arbitrary_types_allowed=True,
+    )
+
+
+class ELKInputData(BaseELKModel):
     """Data that can be fed to ELK."""
 
-    id: te.Required[str]
-    layoutOptions: LayoutOptions
-    children: cabc.MutableSequence[ELKInputChild]  # type: ignore
-    edges: cabc.MutableSequence[ELKInputEdge]
+    id: str
+    layoutOptions: LayoutOptions = {}
+    children: cabc.MutableSequence[ELKInputChild] = pydantic.Field(
+        default_factory=list
+    )
+    edges: cabc.MutableSequence[ELKInputEdge] = pydantic.Field(
+        default_factory=list
+    )
 
 
-class ELKInputChild(ELKInputData, total=False):
+class ELKInputChild(ELKInputData):
     """Children of either `ELKInputData` or `ELKInputChild`."""
 
-    labels: cabc.MutableSequence[ELKInputLabel]
-    ports: cabc.MutableSequence[ELKInputPort]
+    labels: cabc.MutableSequence[ELKInputLabel] = pydantic.Field(
+        default_factory=list
+    )
+    ports: cabc.MutableSequence[ELKInputPort] = pydantic.Field(
+        default_factory=list
+    )
 
-    width: t.Union[int, float]
-    height: t.Union[int, float]
+    width: t.Union[int, float] = 0
+    height: t.Union[int, float] = 0
 
 
-class ELKInputLabel(te.TypedDict, total=False):
+class ELKInputLabel(BaseELKModel):
     """Label data that can be fed to ELK."""
 
-    text: te.Required[str]
-    layoutOptions: LayoutOptions
-    width: t.Union[int, float]
-    height: t.Union[int, float]
+    text: str
+    layoutOptions: LayoutOptions = pydantic.Field(default_factory=list)
+    width: t.Union[int, float] = 0
+    height: t.Union[int, float] = 0
 
 
-class ELKInputPort(t.TypedDict):
+class ELKInputPort(BaseELKModel):
     """Connector data that can be fed to ELK."""
 
     id: str
+    layoutOptions: LayoutOptions = pydantic.Field(default_factory=dict)
     width: t.Union[int, float]
     height: t.Union[int, float]
 
-    layoutOptions: te.NotRequired[cabc.MutableMapping[str, t.Any]]
 
-
-class ELKInputEdge(te.TypedDict):
+class ELKInputEdge(BaseELKModel):
     """Exchange data that can be fed to ELK."""
 
     id: str
     sources: cabc.MutableSequence[str]
     targets: cabc.MutableSequence[str]
-    labels: te.NotRequired[cabc.MutableSequence[ELKInputLabel]]
+    labels: cabc.MutableSequence[ELKInputLabel] = pydantic.Field(
+        default_factory=list
+    )
 
 
-class ELKPoint(t.TypedDict):
+class ELKPoint(BaseELKModel):
     """Point data in ELK."""
 
     x: t.Union[int, float]
     y: t.Union[int, float]
 
 
-class ELKSize(t.TypedDict):
+class ELKSize(BaseELKModel):
     """Size data in ELK."""
 
     width: t.Union[int, float]
     height: t.Union[int, float]
 
 
-class ELKOutputElement(t.TypedDict):
+class ELKOutputElement(BaseELKModel):
     """Base class for all elements that comes out of ELK."""
 
     id: str
 
-    style: dict[str, t.Any]
+    style: dict[str, t.Any] = pydantic.Field(default_factory=dict)
+
+
+class ELKOutputDiagramElement(ELKOutputElement):
+    """Class for positioned and sized elements that come out of ELK."""
+
+    position: ELKPoint
+    size: ELKSize
 
 
 class ELKOutputData(ELKOutputElement):
     """Data that comes from ELK."""
 
     type: t.Literal["graph"]
-    children: cabc.MutableSequence[ELKOutputChild]  # type: ignore
+    children: cabc.MutableSequence[ELKOutputChild] = pydantic.Field(
+        default_factory=list
+    )
 
 
-class ELKOutputNode(ELKOutputElement):
+class ELKOutputNode(ELKOutputDiagramElement):
     """Node that comes out of ELK."""
 
     type: t.Literal["node"]
-    children: cabc.MutableSequence[ELKOutputChild]  # type: ignore
-
-    position: ELKPoint
-    size: ELKSize
+    children: cabc.MutableSequence[ELKOutputChild] = pydantic.Field(
+        default_factory=list
+    )
 
 
 class ELKOutputJunction(ELKOutputElement):
     """Exchange-Junction that comes out of ELK."""
 
     type: t.Literal["junction"]
-    children: cabc.MutableSequence[ELKOutputLabel]
+    children: cabc.MutableSequence[ELKOutputLabel] = pydantic.Field(
+        default_factory=list
+    )
 
     position: ELKPoint
-    size: ELKSize
 
 
-class ELKOutputPort(ELKOutputElement):
+class ELKOutputPort(ELKOutputDiagramElement):
     """Port that comes out of ELK."""
 
     type: t.Literal["port"]
-    children: cabc.MutableSequence[ELKOutputLabel]
-
-    position: ELKPoint
-    size: ELKSize
+    children: cabc.MutableSequence[ELKOutputLabel] = pydantic.Field(
+        default_factory=list
+    )
 
 
-class ELKOutputLabel(ELKOutputElement):
+class ELKOutputLabel(ELKOutputDiagramElement):
     """Label that comes out of ELK."""
 
     type: t.Literal["label"]
     text: str
 
-    position: ELKPoint
-    size: ELKSize
-
 
 class ELKOutputEdge(ELKOutputElement):
     """Edge that comes out of ELK."""
@@ -213,10 +236,12 @@ class ELKOutputEdge(ELKOutputElement):
     sourceId: str
     targetId: str
     routingPoints: cabc.MutableSequence[ELKPoint]
-    children: cabc.MutableSequence[ELKOutputLabel]
+    children: cabc.MutableSequence[
+        t.Union[ELKOutputLabel, ELKOutputJunction]
+    ] = pydantic.Field(default_factory=list)
 
 
-ELKOutputChild = t.Union[  # type: ignore
+ELKOutputChild = t.Union[
     ELKOutputEdge,
     ELKOutputJunction,
     ELKOutputLabel,
@@ -323,7 +348,7 @@ def _install_required_npm_pkg_versions() -> None:
             _install_npm_package(pkg_name, pkg_version)
 
 
-def call_elkjs(elk_dict: ELKInputData) -> ELKOutputData:
+def call_elkjs(elk_model: ELKInputData) -> ELKOutputData:
     """Call into elk.js to auto-layout the ``diagram``.
 
     Parameters
@@ -339,20 +364,21 @@ def call_elkjs(elk_dict: ELKInputData) -> ELKOutputData:
     _find_node_and_npm()
     _install_required_npm_pkg_versions()
 
+    ELKInputData.model_validate(elk_model, strict=True)
     proc = subprocess.run(
         ["node", str(PATH_TO_ELK_JS)],
         executable=shutil.which("node"),
         capture_output=True,
         check=False,
-        input=json.dumps(elk_dict),
+        input=elk_model.model_dump_json(exclude_defaults=True),
         text=True,
         env={**os.environ, "NODE_PATH": str(NODE_HOME)},
     )
     if proc.returncode:
         log.getChild("node").error("%s", proc.stderr)
         raise NodeJSError("elk.js process failed")
 
-    return json.loads(proc.stdout)
+    return ELKOutputData.model_validate_json(proc.stdout, strict=True)
 
 
 def get_global_layered_layout_options() -> LayoutOptions:

capellambse_context_diagrams/collectors/dataflow_view.py

@@ -73,7 +73,7 @@ def collector_portless(
     )
     made_edges: set[str] = set()
     for act in activities:
-        data["children"].append(act_box := makers.make_box(act))
+        data.children.append(act_box := makers.make_box(act))
         connections = list(portless.get_exchanges(act, filter=filter))
 
         in_act: dict[str, oa.OperationalActivity] = {}
@@ -84,9 +84,9 @@ def collector_portless(
             else:
                 in_act.setdefault(edge.target.uuid, edge.target)
 
-        act_box["height"] += (
-            makers.PORT_SIZE + 2 * makers.PORT_PADDING
-        ) * max(len(in_act), len(out_act))
+        act_box.height += (makers.PORT_SIZE + 2 * makers.PORT_PADDING) * max(
+            len(in_act), len(out_act)
+        )
 
         ex_datas: list[generic.ExchangeData] = []
         for ex in connections:
@@ -124,7 +124,7 @@ def collector_default(
     )
     made_edges: set[str] = set()
     for fnc in functions:
-        data["children"].append(fnc_box := makers.make_box(fnc))
+        data.children.append(fnc_box := makers.make_box(fnc))
         _ports = default.port_collector(fnc, diagram.type)
         connections = default.port_exchange_collector(_ports, filter=filter)
         in_ports: dict[str, fa.FunctionPort] = {}
@@ -135,12 +135,12 @@ def collector_default(
             else:
                 in_ports.setdefault(edge.target.uuid, edge.target)
 
-        fnc_box["ports"] = [
+        fnc_box.ports = [
             makers.make_port(i.uuid) for i in (in_ports | out_ports).values()
         ]
-        fnc_box["height"] += (
-            makers.PORT_SIZE + 2 * makers.PORT_PADDING
-        ) * max(len(in_ports), len(out_ports))
+        fnc_box.height += (makers.PORT_SIZE + 2 * makers.PORT_PADDING) * max(
+            len(in_ports), len(out_ports)
+        )
 
         ex_datas: list[generic.ExchangeData] = []
         for ex in edges: