feat(protocols): support generating the SDK core protocols module [IFC-3054]

changelog/+protocols-hierarchical-duplicates.fixed.md

@@ -0,0 +1 @@
+Protocols generated for a hierarchical kind no longer declare `parent` and `children` twice. When the schema already exposes them as relationships, which is the case for any schema read from the API, they were emitted once from the relationship list and once more from the hierarchy, and the second pair overrode the first. Regenerate with `infrahubctl protocols` to drop the duplicates.

infrahub_sdk/protocols_generator/constants.py

@@ -1,4 +1,5 @@
 TEMPLATE_FILE_NAME = "template.j2"
+HEADER_FILE_NAME = "header.j2"
 
 ATTRIBUTE_KIND_MAP = {
     "ID": "String",

infrahub_sdk/protocols_generator/generator.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 
 import jinja2
+from typing_extensions import assert_never
 
 from .. import protocols as sdk_protocols
 from ..schema import (
@@ -18,11 +19,12 @@
     RelationshipSchemaAPI,
     TemplateSchemaAPI,
 )
-from .constants import ATTRIBUTE_KIND_MAP, CORE_BASE_CLASS_TO_SYNCIFY, TEMPLATE_FILE_NAME
+from .constants import ATTRIBUTE_KIND_MAP, CORE_BASE_CLASS_TO_SYNCIFY, HEADER_FILE_NAME, TEMPLATE_FILE_NAME
+from .target import ProtocolTarget
 
 
-def load_template() -> str:
-    path = Path(__file__).parent / TEMPLATE_FILE_NAME
+def load_template(file_name: str = TEMPLATE_FILE_NAME) -> str:
+    path = Path(__file__).parent / file_name
     return path.read_text()
 
 
@@ -35,7 +37,10 @@ def move_to_end_of_list(lst: list, item: str) -> list:
 
 
 class CodeGenerator:
-    def __init__(self, schema: dict[str, MainSchemaTypesAll]) -> None:
+    def __init__(
+        self, schema: dict[str, MainSchemaTypesAll], target: ProtocolTarget = ProtocolTarget.USER_SCHEMA
+    ) -> None:
+        self.target: ProtocolTarget = target
         self.generics: dict[str, GenericSchemaAPI | GenericSchema] = {}
         self.nodes: dict[str, NodeSchemaAPI | NodeSchema] = {}
         self.profiles: dict[str, ProfileSchemaAPI] = {}
@@ -51,14 +56,29 @@ def __init__(self, schema: dict[str, MainSchemaTypesAll]) -> None:
             if isinstance(schema_type, TemplateSchemaAPI):
                 self.templates[name] = schema_type
 
-        self.base_protocols = [
-            e
-            for e in dir(sdk_protocols)
-            if not e.startswith("__")
-            and not e.endswith("__")
-            and e
-            not in {"TYPE_CHECKING", "CoreNode", "Optional", "Protocol", "Union", "annotations", "runtime_checkable"}
-        ]
+        match self.target:
+            case ProtocolTarget.USER_SCHEMA:
+                self.base_protocols = [
+                    e
+                    for e in dir(sdk_protocols)
+                    if not e.startswith("__")
+                    and not e.endswith("__")
+                    and e
+                    not in {
+                        "TYPE_CHECKING",
+                        "CoreNode",
+                        "Optional",
+                        "Protocol",
+                        "Union",
+                        "annotations",
+                        "runtime_checkable",
+                    }
+                ]
+            case ProtocolTarget.SDK_CORE:
+                # Nothing can be imported from the module being generated.
+                self.base_protocols = []
+            case _:
+                assert_never(self.target)
 
         self.sorted_generics = self._sort_and_filter_models(self.generics, filters=["CoreNode", *self.base_protocols])
         self.sorted_nodes = self._sort_and_filter_models(self.nodes, filters=["CoreNode", *self.base_protocols])
@@ -69,7 +89,29 @@ def __init__(self, schema: dict[str, MainSchemaTypesAll]) -> None:
             self.templates, filters=["CoreObjectTemplate", *self.base_protocols]
         )
 
+        # Which referenced names have a Sync counterpart to switch to when rendering sync output.
+        # The two positions resolve names differently for a user schema: a peer may be any core
+        # kind, while an inheritance list only ever switches the few base classes. Generating the
+        # core module makes both the same, since every class in it is local.
+        match self.target:
+            case ProtocolTarget.USER_SCHEMA:
+                self._inherited_sync_names = frozenset(CORE_BASE_CLASS_TO_SYNCIFY)
+                self._peer_sync_names = frozenset(
+                    name for name in self.base_protocols if f"{name}Sync" in self.base_protocols
+                )
+            case ProtocolTarget.SDK_CORE:
+                local_names = self._local_class_names() | {"CoreNode"}
+                self._inherited_sync_names = local_names
+                self._peer_sync_names = local_names
+            case _:
+                assert_never(self.target)
+
     def render(self, sync: bool = True) -> str:
+        """Render the protocols module.
+
+        ``sync`` selects which variant to render for ``USER_SCHEMA``. ``SDK_CORE`` renders both
+        variants into a single module, so it does not apply there.
+        """
         jinja2_env = jinja2.Environment(
             loader=jinja2.BaseLoader(),
             trim_blocks=True,
@@ -80,23 +122,45 @@ def render(self, sync: bool = True) -> str:
         jinja2_env.filters["render_relationship"] = self._jinja2_filter_render_relationship
         jinja2_env.filters["syncify"] = self._jinja2_filter_syncify
 
-        template = jinja2_env.from_string(load_template())
-        return template.render(
+        header = jinja2_env.from_string(load_template(HEADER_FILE_NAME))
+        body = jinja2_env.from_string(load_template(TEMPLATE_FILE_NAME))
+
+        match self.target:
+            case ProtocolTarget.USER_SCHEMA:
+                return header.render(sync=sync, base_protocols=self.base_protocols, core=False) + self._render_body(
+                    body, sync=sync, suffix=""
+                )
+            case ProtocolTarget.SDK_CORE:
+                return (
+                    header.render(sync=False, base_protocols=self.base_protocols, core=True)
+                    + self._render_body(body, sync=False, suffix="")
+                    + self._render_body(body, sync=True, suffix="Sync")
+                )
+            case _:
+                assert_never(self.target)
+
+    def _render_body(self, body: jinja2.Template, sync: bool, suffix: str) -> str:
+        return body.render(
             generics=self.sorted_generics,
             nodes=self.sorted_nodes,
             profiles=self.sorted_profiles,
             templates=self.sorted_templates,
-            base_protocols=self.base_protocols,
             core_node_name="CoreNodeSync" if sync else "CoreNode",
             sync=sync,
+            suffix=suffix,
         )
 
-    @staticmethod
-    def _jinja2_filter_syncify(value: str | list, sync: bool = False) -> str | list:
+    def _local_class_names(self) -> frozenset[str]:
+        return frozenset(
+            f"{model.namespace}{model.name}"
+            for model in (*self.sorted_generics, *self.sorted_nodes, *self.sorted_profiles, *self.sorted_templates)
+        )
+
+    def _jinja2_filter_syncify(self, value: str | list, sync: bool = False) -> str | list:
         """Filter to help with the convertion to sync.
 
         If a string is provided, append Sync to the end of the string
-        If a list is provided, search for CoreNode and replace it with CoreNodeSync
+        If a list is provided, append Sync to the items that have a Sync counterpart
         """
         if isinstance(value, list):
             # Order the list based on the CORE_BASE_CLASS_TO_SYNCIFY list to ensure the base classes are always last
@@ -110,7 +174,7 @@ def _jinja2_filter_syncify(value: str | list, sync: bool = False) -> str | list:
             return f"{value}Sync"
 
         if isinstance(value, list):
-            return [f"{item}Sync" if item in CORE_BASE_CLASS_TO_SYNCIFY else item for item in value]
+            return [f"{item}Sync" if item in self._inherited_sync_names else item for item in value]
 
         return value
 
@@ -136,10 +200,9 @@ def _jinja2_filter_render_relationship(self, value: RelationshipSchemaAPI, sync:
 
         if sync:
             type_ += "Sync"
-            # Core peer protocols expose a dedicated ``*Sync`` variant; reference it in sync
-            # output. Locally generated node/generic classes keep their name (they already
-            # inherit the sync base class), so only swap when a ``*Sync`` peer actually exists.
-            if f"{peer}Sync" in self.base_protocols:
+            # Peers with a dedicated ``*Sync`` variant are referenced by it in sync output. The
+            # rest keep their name, because they already are the sync class.
+            if peer in self._peer_sync_names:
                 peer = f"{peer}Sync"
 
         return f"{name}: {type_}[{peer}]"

infrahub_sdk/protocols_generator/header.j2

@@ -0,0 +1,92 @@
+{% if core %}
+# Generated by "invoke backend.generate", do not edit directly
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from .protocols_base import CoreNode, CoreNodeSync
+
+if TYPE_CHECKING:
+    from infrahub_sdk.node import (
+        RelationshipAttribute,
+        RelationshipAttributeSync,
+        RelationshipManager,
+        RelationshipManagerSync,
+    )
+    from .protocols_base import (
+        AnyAttribute,
+        AnyAttributeOptional,
+        String,
+        StringOptional,
+        Integer,
+        IntegerOptional,
+        Boolean,
+        BooleanOptional,
+        DateTime,
+        DateTimeOptional,
+        Dropdown,
+        DropdownOptional,
+        HashedPassword,
+        HashedPasswordOptional,
+        MacAddress,
+        MacAddressOptional,
+        IPHost,
+        IPHostOptional,
+        IPNetwork,
+        IPNetworkOptional,
+        IPAddress,
+        IPAddressOptional,
+        JSONAttribute,
+        JSONAttributeOptional,
+        ListAttribute,
+        ListAttributeOptional,
+        URL,
+        URLOptional,
+    )
+
+{% else %}
+#
+# Generated by "infrahubctl protocols"
+#
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Optional
+
+from infrahub_sdk.protocols import {{ "CoreNode" | syncify(sync) }}, {{ base_protocols | join(', ') }}
+
+if TYPE_CHECKING:
+    from infrahub_sdk.node import {{ "RelatedNode" | syncify(sync) }}, {{ "RelationshipAttribute" | syncify(sync) }}, {{ "RelationshipManager" | syncify(sync) }}
+    from infrahub_sdk.protocols_base import (
+        AnyAttribute,
+        AnyAttributeOptional,
+        String,
+        StringOptional,
+        Integer,
+        IntegerOptional,
+        Boolean,
+        BooleanOptional,
+        DateTime,
+        DateTimeOptional,
+        Dropdown,
+        DropdownOptional,
+        HashedPassword,
+        HashedPasswordOptional,
+        MacAddress,
+        MacAddressOptional,
+        IPHost,
+        IPHostOptional,
+        IPNetwork,
+        IPNetworkOptional,
+        IPAddress,
+        IPAddressOptional,
+        JSONAttribute,
+        JSONAttributeOptional,
+        ListAttribute,
+        ListAttributeOptional,
+        URL,
+        URLOptional,
+    )
+
+{% endif %}

infrahub_sdk/protocols_generator/target.py

@@ -0,0 +1,16 @@
+from enum import Enum
+
+
+class ProtocolTarget(str, Enum):
+    """Which protocols module is being generated.
+
+    ``USER_SCHEMA`` renders protocols for a user's own schema into a standalone module, one file
+    per async/sync variant, importing the core kinds it references from ``infrahub_sdk.protocols``.
+
+    ``SDK_CORE`` renders ``infrahub_sdk.protocols`` itself. Every kind is local, so nothing can be
+    imported from that module, both variants share a single file, and the sync classes carry a
+    ``Sync`` suffix to keep their names distinct.
+    """
+
+    USER_SCHEMA = "user-schema"
+    SDK_CORE = "sdk-core"