IHS-156 / #497 : support from_pool attributes on Python SDK queries

AGENTS.md

@@ -7,7 +7,7 @@ Infrahub Python SDK - async/sync client for Infrahub infrastructure management.
 ```bash
 uv sync --all-groups --all-extras   # Install all deps
 uv run invoke format                # Format code
-uv run invoke lint                  # All linters (code + yamllint + documentation)
+uv run invoke lint                  # Full pipeline: ruff, yamllint, ty, mypy, markdownlint, vale
 uv run invoke lint-code             # All linters for Python code
 uv run pytest tests/unit/           # Unit tests
 uv run pytest tests/integration/    # Integration tests

changelog/497.fixed.md

@@ -0,0 +1 @@
+Fixed Python SDK query generation regarding from_pool generated attribute value

docs/docs/python-sdk/sdk_ref/infrahub_sdk/node/attribute.mdx

@@ -24,3 +24,16 @@ value(self) -> Any
 ```python
 value(self, value: Any) -> None
 ```
+
+#### `is_from_pool_attribute`
+
+```python
+is_from_pool_attribute(self) -> bool
+```
+
+Check whether this attribute's value is sourced from a resource pool.
+
+**Returns:**
+
+- True if the attribute value is a resource pool node or was explicitly
+- allocated from a pool.

infrahub_sdk/node/attribute.py

@@ -2,7 +2,7 @@
 
 import ipaddress
 from collections.abc import Callable
-from typing import TYPE_CHECKING, Any, get_args
+from typing import TYPE_CHECKING, Any, NamedTuple, get_args
 
 from ..protocols_base import CoreNodeBase
 from ..uuidt import UUIDT
@@ -25,8 +25,12 @@ def __init__(self, name: str, schema: AttributeSchemaAPI, data: Any | dict) -> N
         """
         self.name = name
         self._schema = schema
+        self._from_pool: dict[str, Any] | None = None
 
-        if not isinstance(data, dict) or "value" not in data:
+        if isinstance(data, dict) and "from_pool" in data:
+            self._from_pool = data.pop("from_pool")
+            data.setdefault("value", None)
+        elif not isinstance(data, dict) or "value" not in data:
             data = {"value": data}
 
         self._properties_flag = PROPERTIES_FLAG
@@ -76,38 +80,57 @@ def value(self, value: Any) -> None:
         self._value = value
         self.value_has_been_mutated = True
 
-    def _generate_input_data(self) -> dict | None:
-        data: dict[str, Any] = {}
-        variables: dict[str, Any] = {}
+    def _initialize_graphql_payload(self) -> _GraphQLPayloadAttribute:
+        """Resolve the attribute value into a GraphQL mutation payload object."""
 
-        if self.value is None:
-            if self._schema.optional and self.value_has_been_mutated:
-                data["value"] = None
-            return data
-
-        if isinstance(self.value, str):
-            if SAFE_VALUE.match(self.value):
-                data["value"] = self.value
-            else:
-                var_name = f"value_{UUIDT.new().hex}"
-                variables[var_name] = self.value
-                data["value"] = f"${var_name}"
-        elif isinstance(self.value, get_args(IP_TYPES)):
-            data["value"] = self.value.with_prefixlen
-        elif isinstance(self.value, CoreNodeBase) and self.value.is_resource_pool():
-            data["from_pool"] = {"id": self.value.id}
-        else:
-            data["value"] = self.value
-
-        for prop_name in self._properties_flag:
-            if getattr(self, prop_name) is not None:
-                data[prop_name] = getattr(self, prop_name)
-
-        for prop_name in self._properties_object:
-            if getattr(self, prop_name) is not None:
-                data[prop_name] = getattr(self, prop_name)._generate_input_data()
+        # Pool-based allocation (dict data or resource-pool node)
+        if self._from_pool is not None:
+            return _GraphQLPayloadAttribute(
+                payload_dict={"from_pool": self._from_pool}, variables={}, need_additional_properties=True
+            )
+        if isinstance(self.value, CoreNodeBase) and self.value.is_resource_pool():
+            return _GraphQLPayloadAttribute(
+                payload_dict={"from_pool": {"id": self.value.id}}, variables={}, need_additional_properties=True
+            )
 
-        return {"data": data, "variables": variables}
+        # Null value
+        if self.value is None:
+            data = {"value": None} if (self._schema.optional and self.value_has_been_mutated) else {}
+            return _GraphQLPayloadAttribute(payload_dict=data, variables={}, need_additional_properties=False)
+
+        # Unsafe strings need a variable binding to avoid injection
+        if isinstance(self.value, str) and not SAFE_VALUE.match(self.value):
+            var_name = f"value_{UUIDT.new().hex}"
+            return _GraphQLPayloadAttribute(
+                payload_dict={"value": f"${var_name}"},
+                variables={var_name: self.value},
+                need_additional_properties=True,
+            )
+
+        # Safe strings, IP types, and everything else
+        value = self.value.with_prefixlen if isinstance(self.value, get_args(IP_TYPES)) else self.value
+        return _GraphQLPayloadAttribute(payload_dict={"value": value}, variables={}, need_additional_properties=True)
+
+    def _generate_input_data(self) -> _GraphQLPayloadAttribute:
+        """Build the input payload for a GraphQL mutation on this attribute.
+
+        Returns a ResolvedValue object, which contains all the data required.
+        """
+        graphql_payload = self._initialize_graphql_payload()
+
+        properties_flag: dict[str, Any] = {
+            property_name: getattr(self, property_name)
+            for property_name in self._properties_flag
+            if getattr(self, property_name) is not None
+        }
+        properties_object: dict[str, dict] = {
+            property_name: getattr(self, property_name)._generate_input_data()
+            for property_name in self._properties_object
+            if getattr(self, property_name) is not None
+        }
+        graphql_payload.add_properties(properties_flag, properties_object)
+
+        return graphql_payload
 
     def _generate_query_data(self, property: bool = False, include_metadata: bool = False) -> dict | None:
         data: dict[str, Any] = {"value": None}
@@ -128,7 +151,42 @@ def _generate_query_data(self, property: bool = False, include_metadata: bool =
         return data
 
     def _generate_mutation_query(self) -> dict[str, Any]:
-        if isinstance(self.value, CoreNodeBase) and self.value.is_resource_pool():
+        if self.is_from_pool_attribute():
             # If it points to a pool, ask for the value of the pool allocated resource
             return {self.name: {"value": None}}
         return {}
+
+    def is_from_pool_attribute(self) -> bool:
+        """Check whether this attribute's value is sourced from a resource pool.
+
+        Returns:
+            True if the attribute value is a resource pool node or was explicitly
+            allocated from a pool.
+        """
+        return (isinstance(self.value, CoreNodeBase) and self.value.is_resource_pool()) or self._from_pool is not None
+
+
+class _GraphQLPayloadAttribute(NamedTuple):
+    """Result of resolving an attribute value for a GraphQL mutation.
+
+    Attributes:
+        payload_dict: Key/value entries to include in the mutation payload
+            (e.g. ``{"value": ...}`` or ``{"from_pool": ...}``).
+        variables: GraphQL variable bindings for unsafe string values.
+        need_additional_properties: When ``True``, the payload needs to append property flags/objects
+    """
+
+    payload_dict: dict[str, Any]
+    variables: dict[str, Any]
+    need_additional_properties: bool
+
+    def to_dict(self) -> dict[str, Any]:
+        return {"data": self.payload_dict, "variables": self.variables}
+
+    def add_properties(self, properties_flag: dict[str, Any], properties_object: dict[str, dict]) -> None:
+        if self.need_additional_properties:
+            for prop_name, prop in properties_flag.items():
+                self.payload_dict[prop_name] = prop
+
+            for prop_name, prop in properties_object.items():
+                self.payload_dict[prop_name] = prop

infrahub_sdk/node/node.py

@@ -216,7 +216,7 @@ def is_resource_pool(self) -> bool:
     def get_raw_graphql_data(self) -> dict | None:
         return self._data
 
-    def _generate_input_data(  # noqa: C901, PLR0915
+    def _generate_input_data(  # noqa: C901
         self,
         exclude_unmodified: bool = False,
         exclude_hfid: bool = False,
@@ -228,27 +228,18 @@ def _generate_input_data(  # noqa: C901, PLR0915
             dict[str, Dict]: Representation of an input data in dict format
         """
 
-        data = {}
-        variables = {}
+        data: dict[str, Any] = {}
+        variables: dict[str, Any] = {}
 
         for item_name in self._attributes:
             attr: Attribute = getattr(self, item_name)
             if attr._schema.read_only:
                 continue
-            attr_data = attr._generate_input_data()
-
-            # NOTE, this code has been inherited when we splitted attributes and relationships
-            # into 2 loops, most likely it's possible to simply it
-            if attr_data and isinstance(attr_data, dict):
-                if variable_values := attr_data.get("data"):
-                    data[item_name] = variable_values
-                else:
-                    data[item_name] = attr_data
-                if variable_names := attr_data.get("variables"):
-                    variables.update(variable_names)
-
-            elif attr_data and isinstance(attr_data, list):
-                data[item_name] = attr_data
+            graphql_payload = attr._generate_input_data()
+            if graphql_payload.payload_dict:
+                data[item_name] = graphql_payload.payload_dict
+            if graphql_payload.variables:
+                variables.update(graphql_payload.variables)
 
         for item_name in self._relationships:
             allocate_from_pool = False
@@ -1011,11 +1002,7 @@ async def _process_mutation_result(
 
         for attr_name in self._attributes:
             attr = getattr(self, attr_name)
-            if (
-                attr_name not in object_response
-                or not isinstance(attr.value, InfrahubNodeBase)
-                or not attr.value.is_resource_pool()
-            ):
+            if attr_name not in object_response or not attr.is_from_pool_attribute():
                 continue
 
             # Process allocated resource from a pool and update attribute
@@ -1819,11 +1806,7 @@ def _process_mutation_result(
 
         for attr_name in self._attributes:
             attr = getattr(self, attr_name)
-            if (
-                attr_name not in object_response
-                or not isinstance(attr.value, InfrahubNodeBase)
-                or not attr.value.is_resource_pool()
-            ):
+            if attr_name not in object_response or not attr.is_from_pool_attribute():
                 continue
 
             # Process allocated resource from a pool and update attribute

tests/AGENTS.md

@@ -17,6 +17,12 @@ uv run pytest tests/unit/test_client.py      # Single file
 ```text
 tests/
 ├── unit/           # Fast, mocked, no external deps
+│   ├── ctl/        # CLI command tests
+│   └── sdk/        # SDK tests
+│       ├── pool/   # Resource pool allocation tests
+│       ├── spec/   # Object spec tests
+│       ├── checks/ # InfrahubCheck tests
+│       └── ...     # Core SDK tests (client, node, schema, etc.)
 ├── integration/    # Real Infrahub via testcontainers
 ├── fixtures/       # Test data (JSON, YAML)
 └── helpers/        # Test utilities