feat: add update_from_dict to update node fields from a dict

changelog/272.added.md

@@ -0,0 +1 @@
+Added a new `update_from_dict()` method on `InfrahubNode` and `InfrahubNodeSync` to update several attributes and relationships of an existing node at once from a dict, mirroring the data format accepted when creating a node. Attribute values accept scalars, dicts with `value` and properties, or `from_pool` allocations; relationships accept IDs, dicts, node objects, or lists of those for cardinality-many. Passing `None` clears a cardinality-one relationship and `[]` clears a cardinality-many relationship. Unknown field names raise a `ValueError` and leave the node unmodified.

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

@@ -1191,6 +1191,31 @@ The metadata is populated only when the parent query was executed with
 
 - NodeMetadata | None: The node metadata if fetched, otherwise ``None``.
 
+#### `update_from_dict`
+
+```python
+update_from_dict(self, data: dict[str, Any]) -> None
+```
+
+Update attributes and relationships of this node from a dict.
+
+Values accept the same formats as node creation: for attributes, a scalar value,
+a dict with a ``value`` key and optional properties, or a dict with a
+``from_pool`` key; for relationships, an ID, a dict describing the peer, a node
+object, or a list of those for cardinality-many. Pass ``None`` to clear a
+cardinality-one relationship and ``[]`` (or ``None``) to clear a
+cardinality-many relationship.
+
+**Args:**
+
+- `data`: Mapping of attribute and relationship names to their
+new values.
+
+**Raises:**
+
+- `ValueError`: If a key does not match any attribute or relationship of the
+schema. The node is left unmodified in that case.
+
 #### `get_kind`
 
 ```python

infrahub_sdk/node/node.py

@@ -294,6 +294,42 @@ def _get_request_context(self, request_context: RequestContext | None = None) ->
     def _init_relationships(self, data: dict | None = None) -> None:
         pass
 
+    def update_from_dict(self, data: dict[str, Any]) -> None:
+        """Update attributes and relationships of this node from a dict.
+
+        Values accept the same formats as node creation: for attributes, a scalar value,
+        a dict with a ``value`` key and optional properties, or a dict with a
+        ``from_pool`` key; for relationships, an ID, a dict describing the peer, a node
+        object, or a list of those for cardinality-many. Pass ``None`` to clear a
+        cardinality-one relationship and ``[]`` (or ``None``) to clear a
+        cardinality-many relationship.
+
+        Args:
+            data (dict[str, Any]): Mapping of attribute and relationship names to their
+                new values.
+
+        Raises:
+            ValueError: If a key does not match any attribute or relationship of the
+                schema. The node is left unmodified in that case.
+
+        """
+        unknown_fields = [key for key in data if key not in self._attributes and key not in self._relationships]
+        if unknown_fields:
+            raise ValueError(
+                f"Unable to update {self._schema.kind}, unknown field(s): {', '.join(sorted(unknown_fields))}"
+            )
+
+        for name, value in data.items():
+            if name in self._attributes:
+                attribute = Attribute(name=name, schema=self._schema.get_attribute(name=name), data=value)
+                attribute.value_has_been_mutated = True
+                self._attribute_data[name] = attribute
+            else:
+                self._update_relationship_from_dict(name=name, data=value)
+
+    def _update_relationship_from_dict(self, name: str, data: Any) -> None:
+        raise NotImplementedError
+
     def __repr__(self) -> str:
         if self.display_label:
             return self.display_label
@@ -980,6 +1016,23 @@ def __setattr__(self, name: str, value: Any) -> None:
 
         super().__setattr__(name, value)
 
+    def _update_relationship_from_dict(self, name: str, data: Any) -> None:
+        rel_schema = self._schema.get_relationship(name=name)
+        if rel_schema.cardinality == RelationshipCardinality.ONE:
+            setattr(self, name, data)
+            return
+
+        manager = RelationshipManager(
+            name=name,
+            client=self._client,
+            node=self,
+            branch=self._branch,
+            schema=rel_schema,
+            data=data if data is not None else [],
+        )
+        manager._has_update = True
+        self._relationship_cardinality_many_data[name] = manager
+
     async def generate(self, nodes: list[str] | None = None) -> None:
         """Trigger artifact generation for this artifact definition.
 
@@ -2171,6 +2224,23 @@ def __setattr__(self, name: str, value: Any) -> None:
 
         super().__setattr__(name, value)
 
+    def _update_relationship_from_dict(self, name: str, data: Any) -> None:
+        rel_schema = self._schema.get_relationship(name=name)
+        if rel_schema.cardinality == RelationshipCardinality.ONE:
+            setattr(self, name, data)
+            return
+
+        manager = RelationshipManagerSync(
+            name=name,
+            client=self._client,
+            node=self,
+            branch=self._branch,
+            schema=rel_schema,
+            data=data if data is not None else [],
+        )
+        manager._has_update = True
+        self._relationship_cardinality_many_data[name] = manager
+
     def generate(self, nodes: list[str] | None = None) -> None:
         """Trigger artifact generation for this artifact definition.
 

infrahub_sdk/protocols_base.py

@@ -211,6 +211,8 @@ def get_raw_graphql_data(self) -> dict | None: ...
 
     def get_node_metadata(self) -> NodeMetadata | None: ...
 
+    def update_from_dict(self, data: dict[str, Any]) -> None: ...
+
 
 class CoreNode(CoreNodeBase):
     async def save(

tests/unit/sdk/test_node.py

@@ -2044,6 +2044,140 @@ async def test_update_input_data_empty_relationship(
     assert location._generate_input_data()["data"] == expected_data
 
 
+@pytest.mark.parametrize("client_type", client_types)
+async def test_update_from_dict_attributes(
+    client: InfrahubClient,
+    location_schema: NodeSchemaAPI,
+    location_data01: dict[str, Any],
+    client_type: str,
+) -> None:
+    if client_type == "standard":
+        location = InfrahubNode(client=client, schema=location_schema, data=location_data01)
+    else:
+        location = InfrahubNodeSync(client=client, schema=location_schema, data=location_data01)
+
+    location.update_from_dict(data={"name": "JFK1", "description": "new description"})
+
+    assert location.name.value == "JFK1"
+    assert location.description.value == "new description"
+    assert location._generate_input_data(exclude_unmodified=True)["data"] == {
+        "data": {
+            "id": "llllllll-llll-llll-llll-llllllllllll",
+            "name": {"value": "JFK1"},
+            "description": {"value": "new description"},
+        },
+    }
+
+
+@pytest.mark.parametrize("client_type", client_types)
+async def test_update_from_dict_attribute_with_properties(
+    client: InfrahubClient,
+    location_schema: NodeSchemaAPI,
+    location_data01: dict[str, Any],
+    client_type: str,
+) -> None:
+    if client_type == "standard":
+        location = InfrahubNode(client=client, schema=location_schema, data=location_data01)
+    else:
+        location = InfrahubNodeSync(client=client, schema=location_schema, data=location_data01)
+
+    location.update_from_dict(data={"name": {"value": "JFK1", "is_protected": True}})
+
+    assert location.name.value == "JFK1"
+    assert location.name.is_protected is True
+    assert location._generate_input_data(exclude_unmodified=True)["data"] == {
+        "data": {
+            "id": "llllllll-llll-llll-llll-llllllllllll",
+            "name": {"value": "JFK1", "is_protected": True},
+        },
+    }
+
+
+@pytest.mark.parametrize("client_type", client_types)
+async def test_update_from_dict_relationships(
+    client: InfrahubClient,
+    location_schema: NodeSchemaAPI,
+    location_data01: dict[str, Any],
+    client_type: str,
+) -> None:
+    if client_type == "standard":
+        location = InfrahubNode(client=client, schema=location_schema, data=location_data01)
+    else:
+        location = InfrahubNodeSync(client=client, schema=location_schema, data=location_data01)
+
+    location.update_from_dict(
+        data={
+            "primary_tag": "gggggggg-gggg-gggg-gggg-gggggggggggg",
+            "tags": [
+                "gggggggg-gggg-gggg-gggg-gggggggggggg",
+                {"id": "rrrrrrrr-rrrr-rrrr-rrrr-rrrrrrrrrrrr"},
+            ],
+        }
+    )
+
+    assert location.primary_tag.id == "gggggggg-gggg-gggg-gggg-gggggggggggg"
+    assert [peer.id for peer in location.tags.peers] == [
+        "gggggggg-gggg-gggg-gggg-gggggggggggg",
+        "rrrrrrrr-rrrr-rrrr-rrrr-rrrrrrrrrrrr",
+    ]
+    assert location._generate_input_data()["data"] == {
+        "data": {
+            "id": "llllllll-llll-llll-llll-llllllllllll",
+            "name": {"value": "DFW"},
+            "primary_tag": {"id": "gggggggg-gggg-gggg-gggg-gggggggggggg"},
+            "tags": [
+                {"id": "gggggggg-gggg-gggg-gggg-gggggggggggg"},
+                {"id": "rrrrrrrr-rrrr-rrrr-rrrr-rrrrrrrrrrrr"},
+            ],
+            "type": {"value": "SITE"},
+        },
+    }
+
+
+@pytest.mark.parametrize("client_type", client_types)
+async def test_update_from_dict_clear_relationships(
+    client: InfrahubClient,
+    location_schema: NodeSchemaAPI,
+    location_data01: dict[str, Any],
+    client_type: str,
+) -> None:
+    if client_type == "standard":
+        location = InfrahubNode(client=client, schema=location_schema, data=location_data01)
+    else:
+        location = InfrahubNodeSync(client=client, schema=location_schema, data=location_data01)
+
+    location.update_from_dict(data={"primary_tag": None, "tags": []})
+
+    assert location._generate_input_data()["data"] == {
+        "data": {
+            "id": "llllllll-llll-llll-llll-llllllllllll",
+            "name": {"value": "DFW"},
+            "primary_tag": None,
+            "tags": [],
+            "type": {"value": "SITE"},
+        },
+    }
+
+
+@pytest.mark.parametrize("client_type", client_types)
+async def test_update_from_dict_unknown_field_raises(
+    client: InfrahubClient,
+    location_schema: NodeSchemaAPI,
+    location_data01: dict[str, Any],
+    client_type: str,
+) -> None:
+    if client_type == "standard":
+        location = InfrahubNode(client=client, schema=location_schema, data=location_data01)
+    else:
+        location = InfrahubNodeSync(client=client, schema=location_schema, data=location_data01)
+
+    with pytest.raises(ValueError, match="unknown field"):
+        location.update_from_dict(data={"name": "JFK1", "nonexistent": "value"})
+
+    # The node must be left untouched when validation fails
+    assert location.name.value == "DFW"
+
+
 @pytest.mark.parametrize("client_type", client_types)
 async def test_node_get_relationship_from_store(
     client: InfrahubClient,