IFC-1886: Paginated branch graphql query

backend/infrahub/graphql/queries/__init__.py

@@ -1,5 +1,5 @@
 from .account import AccountPermissions, AccountToken
-from .branch import BranchQueryList
+from .branch import BranchQueryList, InfrahubBranchQueryList
 from .internal import InfrahubInfo
 from .ipam import (
     DeprecatedIPAddressGetNextAvailable,
@@ -20,6 +20,7 @@
     "BranchQueryList",
     "DeprecatedIPAddressGetNextAvailable",
     "DeprecatedIPPrefixGetNextAvailable",
+    "InfrahubBranchQueryList",
     "InfrahubIPAddressGetNextAvailable",
     "InfrahubIPPrefixGetNextAvailable",
     "InfrahubInfo",

backend/infrahub/graphql/queries/branch.py

@@ -2,10 +2,10 @@
 
 from typing import TYPE_CHECKING, Any
 
-from graphene import ID, Field, List, NonNull, String
+from graphene import ID, Field, Int, List, NonNull, String
 
 from infrahub.graphql.field_extractor import extract_graphql_fields
-from infrahub.graphql.types import BranchType
+from infrahub.graphql.types import BranchType, InfrahubBranchType
 
 if TYPE_CHECKING:
     from graphql import GraphQLResolveInfo
@@ -28,3 +28,36 @@ async def branch_resolver(
     resolver=branch_resolver,
     required=True,
 )
+
+
+async def infrahub_branch_resolver(
+    root: dict,  # noqa: ARG001
+    info: GraphQLResolveInfo,
+    **kwargs: Any,
+) -> dict[str, Any]:
+    page = kwargs.pop("page", 1)
+    limit = kwargs.pop("limit", 100)
+    offset = (page - 1) * limit
+    fields = {name: str(field.type) for name, field in BranchType._meta.fields.items()}
+
+    branches = await BranchType.get_list(
+        fields=fields, graphql_context=info.context, limit=limit, offset=offset, **kwargs
+    )
+
+    return {
+        "current_page": page,
+        "count_per_page": limit,
+        "branches": branches,
+    }
+
+
+InfrahubBranchQueryList = Field(
+    InfrahubBranchType,
+    ids=List(ID),
+    name=String(),
+    page=Int(default_value=1),
+    limit=Int(default_value=100),
+    description="Retrieve paginated information about active branches.",
+    resolver=infrahub_branch_resolver,
+    required=True,
+)

backend/infrahub/graphql/schema.py

@@ -39,6 +39,7 @@
     BranchQueryList,
     DeprecatedIPAddressGetNextAvailable,
     DeprecatedIPPrefixGetNextAvailable,
+    InfrahubBranchQueryList,
     InfrahubInfo,
     InfrahubIPAddressGetNextAvailable,
     InfrahubIPPrefixGetNextAvailable,
@@ -65,6 +66,7 @@ class InfrahubBaseQuery(ObjectType):
 
     Relationship = Relationship
 
+    InfrahubBranch = InfrahubBranchQueryList
     InfrahubInfo = InfrahubInfo
     InfrahubStatus = InfrahubStatus
 

backend/infrahub/graphql/types/__init__.py

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

backend/infrahub/graphql/types/branch.py

@@ -2,7 +2,7 @@
 
 from typing import TYPE_CHECKING, Any
 
-from graphene import Boolean, Field, String
+from graphene import Boolean, Field, Int, List, NonNull, String
 
 from infrahub.core.branch import Branch
 from infrahub.core.constants import GLOBAL_BRANCH_NAME
@@ -44,3 +44,9 @@ async def get_list(
                 return []
 
             return [await obj.to_graphql(fields=fields) for obj in objs if obj.name != GLOBAL_BRANCH_NAME]
+
+
+class InfrahubBranchType(InfrahubObjectType):
+    current_page = Int()
+    count_per_page = Int()
+    branches = List(of_type=NonNull(BranchType))

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

@@ -148,3 +148,81 @@ async def test_branch_query(
         assert id_response.data
         assert id_response.data["Branch"][0]["name"] == "branch3"
         assert len(id_response.data["Branch"]) == 1
+
+    async def test_paginated_branch_query(
+        self,
+        db: InfrahubDatabase,
+        default_branch: Branch,
+        register_core_models_schema,
+        session_admin,
+        client,
+        service,
+    ):
+        for i in range(10):
+            create_branch_query = """
+            mutation {
+                BranchCreate(data: { name: "%s", description: "%s" }) {
+                    ok
+                    object {
+                        id
+                        name
+                    }
+                }
+            }
+            """ % (
+                f"sample-branch-{i}",
+                f"sample description {i}",
+            )
+
+            gql_params = await prepare_graphql_params(
+                db=db,
+                include_subscription=False,
+                branch=default_branch,
+                account_session=session_admin,
+                service=service,
+            )
+            branch_result = await graphql(
+                schema=gql_params.schema,
+                source=create_branch_query,
+                context_value=gql_params.context,
+                root_value=None,
+                variable_values={},
+            )
+            assert branch_result.errors is None
+            assert branch_result.data
+
+        query = """
+        query {
+            InfrahubBranch(page: 2, limit: 5) {
+                branches {
+                    id
+                    name
+                    description
+                    origin_branch
+                    branched_from
+                    created_at
+                    is_default
+                    is_isolated
+                    has_schema_changes
+                    sync_with_git
+                }
+                current_page
+                count_per_page
+            }
+        }
+        """
+        gql_params = await prepare_graphql_params(
+            db=db, include_subscription=False, branch=default_branch, service=service
+        )
+        all_branches = await graphql(
+            schema=gql_params.schema,
+            source=query,
+            context_value=gql_params.context,
+            root_value=None,
+            variable_values={},
+        )
+        assert all_branches.errors is None
+        assert all_branches.data
+        assert len(all_branches.data["InfrahubBranch"]["branches"]) == 5
+        assert all_branches.data["InfrahubBranch"]["current_page"] == 2
+        assert all_branches.data["InfrahubBranch"]["count_per_page"] == 5