Refactor InfrahubNode to avoid dynamic class creation

Author: ogenstad

infrahub_sdk/node/node.py

@@ -4,10 +4,7 @@
 from typing import TYPE_CHECKING, Any
 
 from ..constants import InfrahubClientMode
-from ..exceptions import (
-    FeatureNotSupportedError,
-    NodeNotFoundError,
-)
+from ..exceptions import FeatureNotSupportedError, NodeNotFoundError, SchemaNotFoundError
 from ..graphql import Mutation, Query
 from ..schema import GenericSchemaAPI, RelationshipCardinality, RelationshipKind
 from ..utils import compare_lists, generate_short_id, get_flat_value
@@ -30,48 +27,6 @@
     from ..types import Order
 
 
-def generate_relationship_property(node: InfrahubNode | InfrahubNodeSync, name: str) -> property:
-    """Generates a property that stores values under a private non-public name.
-
-    Args:
-        node (Union[InfrahubNode, InfrahubNodeSync]): The node instance.
-        name (str): The name of the relationship property.
-
-    Returns:
-        A property object for managing the relationship.
-
-    """
-    internal_name = "_" + name.lower()
-    external_name = name
-
-    def prop_getter(self: InfrahubNodeBase) -> Any:
-        return getattr(self, internal_name)
-
-    def prop_setter(self: InfrahubNodeBase, value: Any) -> None:
-        if isinstance(value, RelatedNodeBase) or value is None:
-            setattr(self, internal_name, value)
-        else:
-            schema = [rel for rel in self._schema.relationships if rel.name == external_name][0]
-            if isinstance(node, InfrahubNode):
-                setattr(
-                    self,
-                    internal_name,
-                    RelatedNode(
-                        name=external_name, branch=node._branch, client=node._client, schema=schema, data=value
-                    ),
-                )
-            else:
-                setattr(
-                    self,
-                    internal_name,
-                    RelatedNodeSync(
-                        name=external_name, branch=node._branch, client=node._client, schema=schema, data=value
-                    ),
-                )
-
-    return property(prop_getter, prop_setter)
-
-
 class InfrahubNodeBase:
     """Base class for InfrahubNode and InfrahubNodeSync"""
 
@@ -86,6 +41,9 @@ def __init__(self, schema: MainSchemaTypesAPI, branch: str, data: dict | None =
         self._data = data
         self._branch = branch
         self._existing: bool = True
+        self._attribute_data: dict[str, Attribute] = {}
+        self._relationship_cardinality_many_data: dict[str, RelationshipManager | RelationshipManagerSync] = {}
+        self._relationship_cardinality_one_data: dict[str, RelatedNode | RelatedNodeSync] = {}
 
         # Generate a unique ID only to be used inside the SDK
         # The format if this ID is purposely different from the ID used by the API
@@ -180,12 +138,30 @@ def hfid_str(self) -> str | None:
     def _init_attributes(self, data: dict | None = None) -> None:
         for attr_schema in self._schema.attributes:
             attr_data = data.get(attr_schema.name, None) if isinstance(data, dict) else None
-            setattr(
-                self,
-                attr_schema.name,
-                Attribute(name=attr_schema.name, schema=attr_schema, data=attr_data),
+            self._attribute_data[attr_schema.name] = Attribute(
+                name=attr_schema.name, schema=attr_schema, data=attr_data
             )
 
+    def __getattr__(
+        self, name: str
+    ) -> Attribute | RelationshipManager | RelationshipManagerSync | RelatedNode | RelatedNodeSync:
+        if "_attribute_data" in self.__dict__ and name in self._attribute_data:
+            return self._attribute_data[name]
+        if "_relationship_cardinality_many_data" in self.__dict__ and name in self._relationship_cardinality_many_data:
+            return self._relationship_cardinality_many_data[name]
+        if "_relationship_cardinality_one_data" in self.__dict__ and name in self._relationship_cardinality_one_data:
+            return self._relationship_cardinality_one_data[name]
+
+        raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        """Set values for attributes that exist or revert to normal behaviour"""
+        if "_attribute_data" in self.__dict__ and name in self._attribute_data:
+            self._attribute_data[name].value = value
+            return
+
+        super().__setattr__(name, value)
+
     def _get_request_context(self, request_context: RequestContext | None = None) -> dict[str, Any] | None:
         if request_context:
             return request_context.model_dump(exclude_none=True)
@@ -506,7 +482,6 @@ def __init__(
             data: Optional data to initialize the node.
         """
         self._client = client
-        self.__class__ = type(f"{schema.kind}InfrahubNode", (self.__class__,), {})
 
         if isinstance(data, dict) and isinstance(data.get("node"), dict):
             data = data.get("node")
@@ -535,27 +510,36 @@ def _init_relationships(self, data: dict | None = None) -> None:
             rel_data = data.get(rel_schema.name, None) if isinstance(data, dict) else None
 
             if rel_schema.cardinality == "one":
-                setattr(self, f"_{rel_schema.name}", None)
-                setattr(
-                    self.__class__,
-                    rel_schema.name,
-                    generate_relationship_property(name=rel_schema.name, node=self),
+                self._relationship_cardinality_one_data[rel_schema.name] = RelatedNode(
+                    name=rel_schema.name, branch=self._branch, client=self._client, schema=rel_schema, data=rel_data
                 )
-                setattr(self, rel_schema.name, rel_data)
             else:
-                setattr(
-                    self,
-                    rel_schema.name,
-                    RelationshipManager(
-                        name=rel_schema.name,
-                        client=self._client,
-                        node=self,
-                        branch=self._branch,
-                        schema=rel_schema,
-                        data=rel_data,
-                    ),
+                self._relationship_cardinality_many_data[rel_schema.name] = RelationshipManager(
+                    name=rel_schema.name,
+                    client=self._client,
+                    node=self,
+                    branch=self._branch,
+                    schema=rel_schema,
+                    data=rel_data,
                 )
 
+    def __setattr__(self, name: str, value: Any) -> None:
+        """Set values for relationship names that exist or revert to normal behaviour"""
+        if "_relationship_cardinality_one_data" in self.__dict__ and name in self._relationship_cardinality_one_data:
+            rel_schemas = [rel_schema for rel_schema in self._schema.relationships if rel_schema.name == name]
+            if not rel_schemas:
+                raise SchemaNotFoundError(
+                    identifier=self._schema.kind,
+                    message=f"Unable to find relationship schema for '{name}' on {self._schema.kind}",
+                )
+            rel_schema = rel_schemas[0]
+            self._relationship_cardinality_one_data[name] = RelatedNode(
+                name=rel_schema.name, branch=self._branch, client=self._client, schema=rel_schema, data=value
+            )
+            return
+
+        super().__setattr__(name, value)
+
     async def generate(self, nodes: list[str] | None = None) -> None:
         self._validate_artifact_definition_support(ARTIFACT_DEFINITION_GENERATE_FEATURE_NOT_SUPPORTED_MESSAGE)
 
@@ -1036,7 +1020,6 @@ def __init__(
             branch (Optional[str]): The branch where the node resides.
             data (Optional[dict]): Optional data to initialize the node.
         """
-        self.__class__ = type(f"{schema.kind}InfrahubNodeSync", (self.__class__,), {})
         self._client = client
 
         if isinstance(data, dict) and isinstance(data.get("node"), dict):
@@ -1066,26 +1049,36 @@ def _init_relationships(self, data: dict | None = None) -> None:
             rel_data = data.get(rel_schema.name, None) if isinstance(data, dict) else None
 
             if rel_schema.cardinality == "one":
-                setattr(self, f"_{rel_schema.name}", None)
-                setattr(
-                    self.__class__,
-                    rel_schema.name,
-                    generate_relationship_property(name=rel_schema.name, node=self),
+                self._relationship_cardinality_one_data[rel_schema.name] = RelatedNodeSync(
+                    name=rel_schema.name, branch=self._branch, client=self._client, schema=rel_schema, data=rel_data
                 )
-                setattr(self, rel_schema.name, rel_data)
+
             else:
-                setattr(
-                    self,
-                    rel_schema.name,
-                    RelationshipManagerSync(
-                        name=rel_schema.name,
-                        client=self._client,
-                        node=self,
-                        branch=self._branch,
-                        schema=rel_schema,
-                        data=rel_data,
-                    ),
+                self._relationship_cardinality_many_data[rel_schema.name] = RelationshipManagerSync(
+                    name=rel_schema.name,
+                    client=self._client,
+                    node=self,
+                    branch=self._branch,
+                    schema=rel_schema,
+                    data=rel_data,
+                )
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        """Set values for relationship names that exist or revert to normal behaviour"""
+        if "_relationship_cardinality_one_data" in self.__dict__ and name in self._relationship_cardinality_one_data:
+            rel_schemas = [rel_schema for rel_schema in self._schema.relationships if rel_schema.name == name]
+            if not rel_schemas:
+                raise SchemaNotFoundError(
+                    identifier=self._schema.kind,
+                    message=f"Unable to find relationship schema for '{name}' on {self._schema.kind}",
                 )
+            rel_schema = rel_schemas[0]
+            self._relationship_cardinality_one_data[name] = RelatedNodeSync(
+                name=rel_schema.name, branch=self._branch, client=self._client, schema=rel_schema, data=value
+            )
+            return
+
+        super().__setattr__(name, value)
 
     def generate(self, nodes: list[str] | None = None) -> None:
         self._validate_artifact_definition_support(ARTIFACT_DEFINITION_GENERATE_FEATURE_NOT_SUPPORTED_MESSAGE)

infrahub_sdk/query_groups.py

@@ -101,7 +101,7 @@ async def get_group(self, store_peers: bool = False) -> InfrahubNode | None:
         if not store_peers:
             return group
 
-        self.previous_members = group.members.peers  # type: ignore[attr-defined]
+        self.previous_members = group.members.peers  # type: ignore[assignment, union-attr]
         return group
 
     async def delete_unused(self) -> None: