refactor(core): share the query-group subscriber lookup

backend/infrahub/computed_attribute/tasks.py

@@ -8,7 +8,8 @@
 from prefect.logging import get_run_logger
 
 from infrahub import lock
-from infrahub.core.constants import ComputedAttributeKind, InfrahubKind, MutationAction
+from infrahub.core.constants import ComputedAttributeKind, MutationAction
+from infrahub.core.query_group.subscribers import fetch_subscriber_refs
 from infrahub.core.recompute.bulk_write import AttributeValueWrite
 from infrahub.core.recompute.dispatch import build_bulk_recompute_dispatcher
 from infrahub.core.registry import registry
@@ -667,17 +668,8 @@ async def query_transform_targets(
     schema_branch = registry.schema.get_schema_branch(name=branch_name)
     client = get_client()
     client.request_context = context.to_request_context()
-    targets = await client.execute_graphql(
-        query=GATHER_GRAPHQL_QUERY_SUBSCRIBERS, variables={"members": [object_id]}, branch_name=branch_name
-    )
-
-    subscribers: list[PythonTransformTarget] = []
-
-    for group in targets[InfrahubKind.GRAPHQLQUERYGROUP]["edges"]:
-        for subscriber in group["node"]["subscribers"]["edges"]:
-            subscribers.append(
-                PythonTransformTarget(object_id=subscriber["node"]["id"], kind=subscriber["node"]["__typename"])
-            )
+    refs = await fetch_subscriber_refs(client=client, node_ids=[object_id], branch=branch_name)
+    subscribers = [PythonTransformTarget(object_id=ref.id, kind=ref.kind) for ref in refs]
 
     nodes_with_computed_attributes = schema_branch.computed_attributes.get_python_attributes_per_node()
 
@@ -706,23 +698,3 @@ async def query_transform_targets(
                 # Must be a creation tag: in-flow tag updates drop tags added mid-run.
                 tags=[WorkflowTag.BRANCH.render(identifier=branch_name)],
             )
-
-
-GATHER_GRAPHQL_QUERY_SUBSCRIBERS = """
-query GatherGraphQLQuerySubscribers($members: [ID!]) {
-  CoreGraphQLQueryGroup(members__ids: $members) {
-    edges {
-      node {
-        subscribers {
-          edges {
-            node {
-              id
-              __typename
-            }
-          }
-        }
-      }
-    }
-  }
-}
-"""

backend/infrahub/core/query_group/subscribers.py

@@ -0,0 +1,60 @@
+"""Resolution of the nodes subscribed to a GraphQL query group.
+
+Kept free of any dependency beyond the SDK client so that consumers in unrelated packages
+can share it without importing each other.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from infrahub.core.constants import InfrahubKind
+
+if TYPE_CHECKING:
+    from infrahub_sdk.client import InfrahubClient
+
+GATHER_GRAPHQL_QUERY_SUBSCRIBERS = """
+query GatherGraphQLQuerySubscribers($members: [ID!]) {
+  CoreGraphQLQueryGroup(members__ids: $members) {
+    edges {
+      node {
+        subscribers {
+          edges {
+            node {
+              id
+              __typename
+            }
+          }
+        }
+      }
+    }
+  }
+}
+"""
+
+
+@dataclass(frozen=True, slots=True)
+class SubscriberRef:
+    """A node subscribed to a query group, as the gather query reports it."""
+
+    id: str
+    kind: str
+
+
+async def fetch_subscriber_refs(*, client: InfrahubClient, node_ids: list[str], branch: str) -> list[SubscriberRef]:
+    """Every node subscribed to a query group that has any of ``node_ids`` as a member.
+
+    The same subscriber is reported once per matching group, so callers that cannot accept
+    duplicates must deduplicate.
+    """
+    result = await client.execute_graphql(
+        query=GATHER_GRAPHQL_QUERY_SUBSCRIBERS,
+        branch_name=branch,
+        variables={"members": node_ids},
+    )
+    return [
+        SubscriberRef(id=subscriber["node"]["id"], kind=subscriber["node"]["__typename"])
+        for group in result[InfrahubKind.GRAPHQLQUERYGROUP]["edges"]
+        for subscriber in group["node"]["subscribers"]["edges"]
+    ]

backend/infrahub/core/regeneration/impact.py

@@ -3,7 +3,7 @@
 from typing import TYPE_CHECKING, assert_never
 
 from infrahub.core import registry
-from infrahub.core.constants import InfrahubKind
+from infrahub.core.query_group.subscribers import fetch_subscriber_refs
 from infrahub.graphql.analyzer import InfrahubGraphQLQueryAnalyzer
 from infrahub.graphql.execution import cached_parse
 from infrahub.graphql.initialization import prepare_graphql_params
@@ -21,26 +21,6 @@
     from infrahub_sdk.diff import NodeDiff
 
 
-GATHER_GRAPHQL_QUERY_SUBSCRIBERS = """
-query GatherGraphQLQuerySubscribers($members: [ID!]) {
-  CoreGraphQLQueryGroup(members__ids: $members) {
-    edges {
-      node {
-        subscribers {
-          edges {
-            node {
-              id
-              __typename
-            }
-          }
-        }
-      }
-    }
-  }
-}
-"""
-
-
 async def get_field_level_impacted_subscribers(
     query_payload: str,
     diff_summary: list[NodeDiff],
@@ -107,15 +87,5 @@ async def get_field_level_impacted_subscribers(
 async def _get_subscribers_for_nodes(
     node_ids: list[str], branch: str, client: InfrahubClient
 ) -> list[ProposedChangeSubscriber]:
-    result = await client.execute_graphql(
-        query=GATHER_GRAPHQL_QUERY_SUBSCRIBERS,
-        branch_name=branch,
-        variables={"members": node_ids},
-    )
-    subscribers = []
-    for group in result[InfrahubKind.GRAPHQLQUERYGROUP]["edges"]:
-        for subscriber in group["node"]["subscribers"]["edges"]:
-            subscribers.append(
-                ProposedChangeSubscriber(subscriber_id=subscriber["node"]["id"], kind=subscriber["node"]["__typename"])
-            )
-    return subscribers
+    refs = await fetch_subscriber_refs(client=client, node_ids=node_ids, branch=branch)
+    return [ProposedChangeSubscriber(subscriber_id=ref.id, kind=ref.kind) for ref in refs]