Add support for alternate format for relationship many

dgarros · dgarros · commit 49f2b74e2ed3 · 2025-04-09T09:19:55.000+02:00
diff --git a/.gitignore b/.gitignore
@@ -27,4 +27,5 @@ dist/*
 **/*.csv
 
 # Generated files
-generated/
+generated/
+sandbox/
diff --git a/infrahub_sdk/spec/object.py b/infrahub_sdk/spec/object.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from enum import Enum
 from typing import TYPE_CHECKING, Any
 
 from pydantic import BaseModel, Field
@@ -14,16 +15,44 @@
     from ..schema import MainSchemaTypesAPI, RelationshipSchema
 
 
+def validate_list_of_scalars(value: list[Any]) -> bool:
+    return all(isinstance(item, (str, int, float, bool)) for item in value)
+
+
+def validate_list_of_hfids(value: list[Any]) -> bool:
+    return all(isinstance(item, (str, list)) for item in value)
+
+
+def validate_list_of_data_dicts(value: list[Any]) -> bool:
+    return all(isinstance(item, dict) and "data" in item for item in value)
+
+
+def validate_list_of_objects(value: list[Any]) -> bool:
+    return all(isinstance(item, dict) for item in value)
+
+
+class RelationshipDataFormat(str, Enum):
+    UNKNOWN = "unknown"
+
+    ONE_REF = "one_ref"
+    ONE_OBJ = "one_obj"
+
+    MANY_OBJ_DICT_LIST = "many_obj_dict_list"
+    MANY_OBJ_LIST_DICT = "many_obj_list_dict"
+    MANY_REF = "many_ref_list"
+
+
 class RelationshipInfo(BaseModel):
     name: str
     rel_schema: RelationshipSchema
     peer_kind: str
     peer_rel: RelationshipSchema | None = None
-    is_reference: bool = True
     reason_relationship_not_valid: str | None = None
+    format: RelationshipDataFormat = RelationshipDataFormat.UNKNOWN
 
     @property
     def is_bidirectional(self) -> bool:
+        """Indicate if a relationship with the same identifier exists on the other side"""
         return bool(self.peer_rel)
 
     @property
@@ -36,6 +65,10 @@ def is_mandatory(self) -> bool:
     def is_valid(self) -> bool:
         return not self.reason_relationship_not_valid
 
+    @property
+    def is_reference(self) -> bool:
+        return self.format in [RelationshipDataFormat.ONE_REF, RelationshipDataFormat.MANY_REF]
+
 
 async def get_relationship_info(
     client: InfrahubClient, schema: MainSchemaTypesAPI, key: str, value: Any, branch: str | None = None
@@ -63,10 +96,38 @@ async def get_relationship_info(
     except ValueError:
         pass
 
-    # Check if the content of the relationship is a reference to existing objects
-    #  or if it contains the data to create/update related objects
-    if isinstance(value, dict) and "data" in value:
-        info.is_reference = False
+    if rel_schema.cardinality == "one" and isinstance(value, list):
+        # validate the list is composed of string
+        if validate_list_of_scalars(value):
+            info.format = RelationshipDataFormat.ONE_REF
+        else:
+            info.reason_relationship_not_valid = "Too many objects provided for a relationship of cardinality one"
+
+    elif rel_schema.cardinality == "one" and isinstance(value, dict) and "data" in value:
+        info.format = RelationshipDataFormat.ONE_OBJ
+
+    elif (
+        rel_schema.cardinality == "many"
+        and isinstance(value, dict)
+        and "data" in value
+        and validate_list_of_objects(value["data"])
+    ):
+        # Initial format, we need to support it for backward compatibility for menu
+        # it's helpful if there is only one type of object to manage
+        info.format = RelationshipDataFormat.MANY_OBJ_DICT_LIST
+
+    elif rel_schema.cardinality == "many" and isinstance(value, dict) and "data" not in value:
+        info.reason_relationship_not_valid = "Invalid structure for a relationship of cardinality many,"
+        " either provide a dict with data as a list or a list of objects"
+
+    elif rel_schema.cardinality == "many" and isinstance(value, list):
+        if validate_list_of_data_dicts(value):
+            info.format = RelationshipDataFormat.MANY_OBJ_LIST_DICT
+        elif validate_list_of_hfids(value):
+            info.format = RelationshipDataFormat.MANY_REF
+        else:
+            info.reason_relationship_not_valid = "Invalid structure for a relationship of cardinality many,"
+            " either provide a list of dict with data or a list of hfids"
 
     return info
 
@@ -100,7 +161,7 @@ async def validate_object(
         context = context or {}
 
         # First validate if all mandatory fields are present
-        for element in schema.mandatory_attribute_names + schema.mandatory_relationship_names:
+        for element in schema.mandatory_input_names:
             if not any([element in data.keys(), element in context.keys()]):
                 errors.append(ValidationError(identifier=element, message=f"{element} is mandatory"))
 
@@ -162,6 +223,7 @@ async def create_node(
         for key, value in data.items():
             if key in schema.attribute_names:
                 clean_data[key] = value
+                continue
 
             if key in schema.relationship_names:
                 rel_schema = schema.get_relationship(name=key)
@@ -181,22 +243,31 @@ async def create_node(
                 #  - if the relationship is not bidirectional, then we need to create the related object First
                 if rel_info.is_reference and isinstance(value, list):
                     clean_data[key] = value
-                elif rel_info.is_reference and rel_schema.cardinality == "one" and isinstance(value, str):
+                elif rel_info.format == RelationshipDataFormat.ONE_REF and isinstance(value, str):
                     clean_data[key] = [value]
                 elif not rel_info.is_reference and rel_info.is_bidirectional and rel_info.is_mandatory:
                     remaining_rels.append(key)
                 elif not rel_info.is_reference and not rel_info.is_mandatory:
-                    nodes = await cls.create_related_nodes(
-                        client=client,
-                        rel_info=rel_info,
-                        data=value["data"],
-                        branch=branch,
-                        default_schema_kind=default_schema_kind,
-                    )
-                    if rel_info.rel_schema.cardinality == "one":
+                    if rel_info.format == RelationshipDataFormat.ONE_OBJ:
+                        nodes = await cls.create_related_nodes(
+                            client=client,
+                            rel_info=rel_info,
+                            data=value,
+                            branch=branch,
+                            default_schema_kind=default_schema_kind,
+                        )
                         clean_data[key] = nodes[0]
+
                     else:
+                        nodes = await cls.create_related_nodes(
+                            client=client,
+                            rel_info=rel_info,
+                            data=value,
+                            branch=branch,
+                            default_schema_kind=default_schema_kind,
+                        )
                         clean_data[key] = nodes
+
                 else:
                     raise ValueError(f"Situation unaccounted for: {rel_info}")
 
@@ -223,16 +294,15 @@ async def create_node(
             if rel_schema.identifier is None:
                 raise ValueError("identifier must be defined")
 
-            rel_data = data[rel]["data"]
             context = {}
-
             if rel_info.peer_rel:
                 context[rel_info.peer_rel.name] = node.id
 
+            # TODO need to account for the different format here
             await cls.create_related_nodes(
                 client=client,
                 rel_info=rel_info,
-                data=rel_data,
+                data=data[rel],
                 context=context,
                 branch=branch,
                 default_schema_kind=default_schema_kind,
@@ -254,20 +324,20 @@ async def create_related_nodes(
 
         nodes: list[InfrahubNode] = []
 
-        if rel_info.rel_schema.cardinality == "one" and isinstance(data, dict):
+        if rel_info.format == RelationshipDataFormat.ONE_OBJ:
             node = await cls.create_node(
                 client=client,
                 schema=peer_schema,
-                data=data,
+                data=data["data"],
                 context=context,
                 branch=branch,
                 default_schema_kind=default_schema_kind,
             )
             return [node]
 
-        if rel_info.rel_schema.cardinality == "many" and isinstance(data, list):
+        if rel_info.format == RelationshipDataFormat.MANY_OBJ_DICT_LIST:
             context = context or {}
-            for idx, peer_data in enumerate(data):
+            for idx, peer_data in enumerate(data["data"]):
                 context["list_index"] = idx
                 if isinstance(peer_data, dict):
                     node = await cls.create_node(
@@ -281,6 +351,19 @@ async def create_related_nodes(
                     nodes.append(node)
             return nodes
 
+        if rel_info.format == RelationshipDataFormat.MANY_OBJ_LIST_DICT:
+            for item in data:
+                node = await cls.create_node(
+                    client=client,
+                    schema=peer_schema,
+                    data=item["data"],
+                    context=context,
+                    branch=branch,
+                    default_schema_kind=default_schema_kind,
+                )
+                nodes.append(node)
+            return nodes
+
         raise ValueError(
             f"Relationship {rel_info.rel_schema.name} doesn't have the right format {rel_info.rel_schema.cardinality} / {type(data)}"
         )
diff --git a/tests/unit/sdk/spec/test_object.py b/tests/unit/sdk/spec/test_object.py
@@ -1,9 +1,16 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 import pytest
-from pytest_httpx import HTTPXMock
 
-from infrahub_sdk.client import InfrahubClient
 from infrahub_sdk.exceptions import ValidationError
-from infrahub_sdk.spec.object import ObjectFile
+from infrahub_sdk.spec.object import ObjectFile, RelationshipDataFormat, get_relationship_info
+
+if TYPE_CHECKING:
+    from pytest_httpx import HTTPXMock
+
+    from infrahub_sdk.client import InfrahubClient
 
 
 @pytest.fixture
@@ -61,3 +68,57 @@ async def test_validate_object_bad_syntax02(
         await obj.validate_format(client=client)
 
     assert "notvalidattribute" in str(exc.value)
+
+
+get_relationship_info_testdata = [
+    pytest.param(
+        [
+            {"data": {"name": "Blue"}},
+            {"data": {"name": "Red"}},
+        ],
+        True,
+        RelationshipDataFormat.MANY_OBJ_LIST_DICT,
+        id="many_obj_list_dict",
+    ),
+    pytest.param(
+        {
+            "data": [
+                {"name": "Blue"},
+                {"name": "Red"},
+            ],
+        },
+        True,
+        RelationshipDataFormat.MANY_OBJ_DICT_LIST,
+        id="many_obj_dict_list",
+    ),
+    pytest.param(
+        ["blue", "red"],
+        True,
+        RelationshipDataFormat.MANY_REF,
+        id="many_ref",
+    ),
+    pytest.param(
+        [
+            {"name": "Blue"},
+            {"name": "Red"},
+        ],
+        False,
+        RelationshipDataFormat.UNKNOWN,
+        id="many_invalid_list_dict",
+    ),
+]
+
+
+@pytest.mark.parametrize("data,is_valid,format", get_relationship_info_testdata)
+async def test_get_relationship_info_tags(
+    client: InfrahubClient,
+    mock_schema_query_01: HTTPXMock,
+    data: dict | list,
+    is_valid: bool,
+    format: RelationshipDataFormat,
+):
+    location_schema = await client.schema.get(kind="BuiltinLocation")
+
+    rel_info = await get_relationship_info(client, location_schema, "tags", data)
+    assert rel_info.is_valid == is_valid
+    assert rel_info.format == format
