change response format

Author: solababs

backend/infrahub/graphql/queries/branch.py

@@ -5,7 +5,7 @@
 from graphene import ID, Field, Int, List, NonNull, String
 
 from infrahub.graphql.field_extractor import extract_graphql_fields
-from infrahub.graphql.types import BranchType, InfrahubBranchType
+from infrahub.graphql.types import BranchType, InfrahubBranch, InfrahubBranchType
 
 if TYPE_CHECKING:
     from graphql import GraphQLResolveInfo
@@ -39,12 +39,10 @@ async def infrahub_branch_resolver(
     fields = extract_graphql_fields(info)
     result: dict[str, Any] = {}
     if "edges" in fields:
-        result["edges"] = [
-            {"node": branch}
-            for branch in await BranchType.get_list(
-                graphql_context=info.context, fields=fields.get("edges", {}).get("node", {}), limit=limit, offset=offset
-            )
-        ]
+        branches = await InfrahubBranch.get_list(
+            graphql_context=info.context, fields=fields.get("edges", {}).get("node", {}), limit=limit, offset=offset
+        )
+        result["edges"] = [{"node": branch} for branch in branches]
     if "count" in fields:
         result["count"] = await InfrahubBranchType.get_list_count(graphql_context=info.context)
     return result

backend/infrahub/graphql/types/__init__.py

@@ -21,7 +21,7 @@
     StrAttributeType,
     TextAttributeType,
 )
-from .branch import BranchType, InfrahubBranchType
+from .branch import BranchType, InfrahubBranch, InfrahubBranchType
 from .interface import InfrahubInterface
 from .node import InfrahubObject
 from .permission import PaginatedObjectPermission
@@ -41,6 +41,7 @@
     "DropdownType",
     "IPHostType",
     "IPNetworkType",
+    "InfrahubBranch",
     "InfrahubBranchType",
     "InfrahubInterface",
     "InfrahubObject",

backend/infrahub/graphql/types/branch.py

@@ -31,6 +31,10 @@ class Meta:
         name = "Branch"
         model = Branch
 
+    @staticmethod
+    async def _map_fields_to_graphql(objs: list[Branch], fields: dict) -> list[dict[str, Any]]:
+        return [await obj.to_graphql(fields=fields) for obj in objs if obj.name != GLOBAL_BRANCH_NAME]
+
     @classmethod
     async def get_list(
         cls,
@@ -44,11 +48,60 @@ async def get_list(
             if not objs:
                 return []
 
-            return [await obj.to_graphql(fields=fields) for obj in objs if obj.name != GLOBAL_BRANCH_NAME]
+            return await cls._map_fields_to_graphql(objs=objs, fields=fields)
+
+
+class RequiredStringValueField(InfrahubObjectType):
+    value = String(required=True)
+
+
+class NonRequiredStringValueField(InfrahubObjectType):
+    value = String(required=False)
+
+
+class NonRequiredBooleanValueField(InfrahubObjectType):
+    value = Boolean(required=False)
+
+
+class InfrahubBranch(BranchType):
+    id = String(required=True)
+    created_at = String(required=False)
+
+    name = Field(RequiredStringValueField, required=True)
+    description = Field(NonRequiredStringValueField, required=False)
+    origin_branch = Field(NonRequiredStringValueField, required=False)
+    branched_from = Field(NonRequiredStringValueField, required=False)
+    sync_with_git = Field(NonRequiredBooleanValueField, required=False)
+    is_default = Field(NonRequiredBooleanValueField, required=False)
+    is_isolated = Field(
+        NonRequiredBooleanValueField, required=False, deprecation_reason="non isolated mode is not supported anymore"
+    )
+    has_schema_changes = Field(NonRequiredBooleanValueField, required=False)
+
+    class Meta:
+        description = "InfrahubBranch"
+        name = "InfrahubBranch"
+
+    @staticmethod
+    async def _map_fields_to_graphql(objs: list[Branch], fields: dict) -> list[dict[str, Any]]:
+        field_keys = fields.keys()
+        result: list[dict[str, Any]] = []
+        for obj in objs:
+            if obj.name == GLOBAL_BRANCH_NAME:
+                continue
+            data = {}
+            for field in field_keys:
+                value = getattr(obj, field, None)
+                if isinstance(fields.get(field), dict):
+                    data[field] = {"value": value}
+                else:
+                    data[field] = value
+            result.append(data)
+        return result
 
 
 class InfrahubBranchEdge(InfrahubObjectType):
-    node = Field(BranchType, required=True)
+    node = Field(InfrahubBranch, required=True)
 
 
 class InfrahubBranchType(InfrahubObjectType):

backend/tests/unit/graphql/queries/test_branch.py

@@ -189,8 +189,12 @@ async def test_paginated_branch_query(
                     count
                     edges {
                         node {
-                            name
-                            description
+                            name {
+                                value
+                            }
+                            description {
+                                value
+                            }
                         }
                     }
                 }
@@ -210,22 +214,22 @@ async def test_paginated_branch_query(
 
         expected_branches = [
             {
-                "description": "Default Branch",
-                "name": "main",
+                "description": {"value": "Default Branch"},
+                "name": {"value": "main"},
             },
             {
-                "description": "my description",
-                "name": "branch3",
+                "description": {"value": "my description"},
+                "name": {"value": "branch3"},
             },
             *[
                 {
-                    "description": f"sample description {i}",
-                    "name": f"sample-branch-{i}",
+                    "description": {"value": f"sample description {i}"},
+                    "name": {"value": f"sample-branch-{i}"},
                 }
                 for i in range(10)
             ],
         ]
         all_branches_data_only = [branch.get("node") for branch in all_branches.data["InfrahubBranch"]["edges"]]
-        assert all_branches_data_only.sort(key=operator.itemgetter("name")) == expected_branches.sort(
-            key=operator.itemgetter("name")
+        assert all_branches_data_only.sort(key=lambda x: x["name"]["value"]) == expected_branches.sort(
+            key=lambda x: x["name"]["value"]
         )