diff --git a/backend/infrahub/computed_attribute/tasks.py b/backend/infrahub/computed_attribute/tasks.py index 687f0cadecc..f77ca28e3a5 100644 --- a/backend/infrahub/computed_attribute/tasks.py +++ b/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 - } - } - } - } - } - } -} -""" diff --git a/backend/infrahub/core/query_group/__init__.py b/backend/infrahub/core/query_group/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/backend/infrahub/core/query_group/subscribers.py b/backend/infrahub/core/query_group/subscribers.py new file mode 100644 index 00000000000..bc05d979cec --- /dev/null +++ b/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"] + ] diff --git a/backend/infrahub/core/regeneration/impact.py b/backend/infrahub/core/regeneration/impact.py index 412c1cf4490..85f1e9ef74e 100644 --- a/backend/infrahub/core/regeneration/impact.py +++ b/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]