Add option to have parent events and send secondary events

Author: ogenstad

backend/infrahub/core/changelog/models.py

@@ -9,9 +9,11 @@
 
 if TYPE_CHECKING:
     from infrahub.core.attribute import BaseAttribute
+    from infrahub.core.branch import Branch
     from infrahub.core.manager import RelationshipSchema
     from infrahub.core.query.relationship import RelationshipPeerData
     from infrahub.core.relationship.model import Relationship
+    from infrahub.database import InfrahubDatabase
 
 
 class PropertyChangelog(BaseModel):
@@ -221,6 +223,11 @@ class NodeChangelog(BaseModel):
     def parent(self) -> ChangelogNodeParent | None:
         return self._parent
 
+    @property
+    def updated_fields(self) -> list[str]:
+        """Return a list of update fields i.e. attributes and relationships"""
+        return list(self.relationships.keys()) + list(self.attributes.keys())
+
     @property
     def root_node_id(self) -> str:
         """Return the top level node_id"""
@@ -404,3 +411,50 @@ def changelog(self) -> RelationshipCardinalityOneChangelog | RelationshipCardina
                 return self.cardinality_one_relationship
             case RelationshipCardinality.MANY:
                 return self.cardinality_many_relationship
+
+
+class RelationshipChangelogGetter:
+    def __init__(self, db: InfrahubDatabase, branch: Branch) -> None:
+        self._db = db
+        self._branch = branch
+
+    async def get_changelogs(self, primary_changelog: NodeChangelog) -> list[NodeChangelog]:
+        """Return secondary changelogs based on this update
+
+        These will typically include updates to relationships on other nodes.
+        """
+        schema_branch = self._db.schema.get_schema_branch(name=self._branch.name)
+        node_schema = schema_branch.get(name=primary_changelog.node_kind)
+        secondaries: list[NodeChangelog] = []
+
+        for relationship in primary_changelog.relationships.values():
+            rel_schema = node_schema.get_relationship(name=relationship.name)
+            if isinstance(relationship, RelationshipCardinalityOneChangelog):
+                # For now this code only looks at the scenario when a cardinality=one relationship
+                # is added to a node and it has a cardinality=many relationship coming back from
+                # another node, it will be expanded to include all variations.
+                if relationship.peer_status == DiffAction.ADDED:
+                    peer_schema = schema_branch.get(name=str(relationship.peer_kind))
+                    peer_relation = peer_schema.get_relationship_by_identifier(
+                        id=str(rel_schema.identifier), raise_on_error=False
+                    )
+                    if peer_relation:
+                        node_changelog = NodeChangelog(
+                            node_id=str(relationship.peer_id),
+                            node_kind=str(relationship.peer_kind),
+                            display_label="n/a",
+                        )
+                        if peer_relation.cardinality == RelationshipCardinality.MANY:
+                            node_changelog.relationships[peer_relation.name] = RelationshipCardinalityManyChangelog(
+                                name=peer_relation.name,
+                                peers=[
+                                    RelationshipPeerChangelog(
+                                        peer_id=primary_changelog.node_id,
+                                        peer_kind=primary_changelog.node_kind,
+                                        peer_status=DiffAction.ADDED,
+                                    )
+                                ],
+                            )
+                            secondaries.append(node_changelog)
+
+        return secondaries

backend/infrahub/core/constants/__init__.py

@@ -289,18 +289,6 @@ class AttributeDBNodeType(InfrahubStringEnum):
     IPNETWORK = "ipnetwork"
 
 
-class EventLevel(InfrahubStringEnum):
-    ZERO = "zero"
-    ONE = "one"
-
-    def to_int(self) -> int:
-        match self:
-            case EventLevel.ZERO:
-                return 0
-            case EventLevel.ONE:
-                return 1
-
-
 RESTRICTED_NAMESPACES: list[str] = [
     "Account",
     "Branch",

backend/infrahub/events/models.py

@@ -7,7 +7,6 @@
 
 from infrahub import __version__
 from infrahub.core.branch import Branch  # noqa: TC001
-from infrahub.core.constants import EventLevel
 from infrahub.message_bus import InfrahubMessage, Meta
 from infrahub.worker import WORKER_IDENTITY
 
@@ -27,10 +26,30 @@ class EventMeta(BaseModel):
         default=WORKER_IDENTITY, description="The worker identity of the initial sender of this message"
     )
     context: list[dict] = Field(default_factory=list)
-    level: EventLevel = Field(default=EventLevel.ZERO)
+    level: int = Field(default=0)
+    has_children: bool = Field(
+        default=False, description="Indicates if this event might potentially have child events under it."
+    )
+
+    id: UUID = Field(
+        default_factory=uuid4,
+        description="UUID of the event",
+    )
+
+    parent: UUID | None = Field(default=None, description="The UUID of the parent event if applicable")
+
+    def get_id(self) -> str:
+        return str(self.id)
 
     def get_related(self) -> list[dict[str, str]]:
-        related: list[dict[str, str]] = []
+        related: list[dict[str, str]] = [
+            {"prefect.resource.id": __version__, "prefect.resource.role": "infrahub.version"},
+            {
+                "prefect.resource.id": self.get_id(),
+                "prefect.resource.role": "infrahub.event",
+                "infrahub.event.has_children": str(self.has_children).lower(),
+            },
+        ]
         if self.account_id:
             related.append(
                 {
@@ -50,25 +69,43 @@ def get_related(self) -> list[dict[str, str]]:
                 }
             )
 
-        related.append({"prefect.resource.id": __version__, "prefect.resource.role": "infrahub.version"})
+        if self.parent:
+            related.append(
+                {
+                    "prefect.resource.id": self.get_id(),
+                    "prefect.resource.role": "infrahub.child_event",
+                    "infrahub.event_parent.id": str(self.parent),
+                }
+            )
 
         return related
 
     @classmethod
     def default(cls) -> EventMeta:
         return cls()
 
+    @classmethod
+    def from_parent(cls, parent: InfrahubEvent) -> EventMeta:
+        """Create the metadata from an existing event
+
+        Note that this action will modify the existing event to indicate that children might be attached to the event
+        """
+        parent.meta.has_children = True
+        return cls(
+            parent=parent.meta.id,
+            branch=parent.meta.branch,
+            request_id=parent.meta.request_id,
+            initiator_id=parent.meta.initiator_id,
+            account_id=parent.meta.account_id,
+            level=parent.meta.level + 1,
+        )
+
 
 class InfrahubEvent(BaseModel):
     meta: EventMeta = Field(default_factory=EventMeta.default)
 
-    id: UUID = Field(
-        default_factory=uuid4,
-        description="UUID of the event",
-    )
-
     def get_id(self) -> str:
-        return str(self.id)
+        return self.meta.get_id()
 
     def get_event_namespace(self) -> str:
         return EVENT_NAMESPACE

backend/infrahub/graphql/mutations/main.py

@@ -9,6 +9,7 @@
 
 from infrahub import config, lock
 from infrahub.core import registry
+from infrahub.core.changelog.models import RelationshipChangelogGetter
 from infrahub.core.constants import InfrahubKind, MutationAction
 from infrahub.core.constraint.node.runner import NodeConstraintRunner
 from infrahub.core.manager import NodeManager
@@ -100,21 +101,39 @@ async def mutate(cls, root: dict, info: GraphQLResolveInfo, data: InputObjectTyp
             if graphql_context.account_session:
                 account_id = graphql_context.account_session.account_id
 
-            event = NodeMutatedEvent(
+            meta = EventMeta(
+                account_id=account_id,
+                initiator_id=WORKER_IDENTITY,
+                request_id=request_id,
+                branch=graphql_context.branch,
+            )
+            main_event = NodeMutatedEvent(
                 kind=obj._schema.kind,
                 node_id=obj.id,
                 data=obj.node_changelog,
                 action=action,
                 fields=_get_data_fields(data),
-                meta=EventMeta(
-                    account_id=account_id,
-                    initiator_id=WORKER_IDENTITY,
-                    request_id=request_id,
-                    branch=graphql_context.branch,
-                ),
+                meta=meta,
             )
-
-            graphql_context.background.add_task(graphql_context.active_service.event.send, event)
+            relationship_changelogs = RelationshipChangelogGetter(db=graphql_context.db, branch=graphql_context.branch)
+            node_changelogs = await relationship_changelogs.get_changelogs(primary_changelog=obj.node_changelog)
+
+            events = [main_event]
+
+            for node_changelog in node_changelogs:
+                meta = EventMeta.from_parent(parent=main_event)
+                event = NodeMutatedEvent(
+                    kind=node_changelog.node_kind,
+                    node_id=node_changelog.node_id,
+                    data=node_changelog,
+                    action=MutationAction.UPDATED,
+                    fields=node_changelog.updated_fields,
+                    meta=meta,
+                )
+                events.append(event)
+
+            for event in events:
+                graphql_context.background.add_task(graphql_context.active_service.event.send, event)
 
         return mutation
 

backend/infrahub/graphql/types/event.py

@@ -2,7 +2,7 @@
 
 from typing import TYPE_CHECKING, Any
 
-from graphene import Field, Int, Interface, List, NonNull, ObjectType, String
+from graphene import Boolean, Field, Int, Interface, List, NonNull, ObjectType, String
 from graphene.types.generic import GenericScalar
 
 from .common import RelatedNode
@@ -28,6 +28,8 @@ class EventNodeInterface(Interface):
     occurred_at = String(required=True)
     level = Int(required=True)
     primary_node = Field(RelatedNode, required=False)
+    has_children = Boolean(required=True)
+    parent_id = String(required=False)
 
     @classmethod
     def resolve_type(

backend/infrahub/task_manager/event.py

@@ -7,7 +7,6 @@
 from prefect.events.schemas.events import ResourceSpecification
 from pydantic import BaseModel, Field, TypeAdapter
 
-from infrahub.core.constants import EventLevel
 from infrahub.log import get_logger
 from infrahub.utils import get_nested_dict
 
@@ -33,10 +32,20 @@ def get_level(self) -> int:
             level = resource.get("infrahub.event.level")
             if level is None:
                 continue
-            return EventLevel(level).to_int()
+            try:
+                return int(level)
+            except ValueError:
+                return 0
 
         return 0
 
+    def get_parent(self) -> str | None:
+        for resource in self.related:
+            if resource.get("prefect.resource.role") != "infrahub.child_event":
+                continue
+            return resource.get("infrahub.event_parent.id")
+        return None
+
     def get_primary_node(self) -> dict[str, str] | None:
         node_id = self.resource.get("infrahub.node.id")
         node_kind = self.resource.get("infrahub.node.kind")
@@ -52,6 +61,15 @@ def get_account_id(self) -> str | None:
             return resource.get("infrahub.resource.id")
         return None
 
+    def has_children(self) -> bool:
+        for resource in self.related:
+            if resource.get("prefect.resource.role") != "infrahub.event":
+                continue
+            if resource.get("infrahub.event.has_children") == "true":
+                return True
+            return False
+        return False
+
     def _return_node_mutation(self) -> dict[str, Any]:
         attributes = []
 
@@ -89,9 +107,11 @@ def to_graphql(self) -> dict[str, Any]:
             "branch": self.get_branch(),
             "account_id": self.get_account_id(),
             "occurred_at": self.occurred.to_iso8601_string(),
+            "has_children": self.has_children(),
             "payload": self.payload,
             "level": self.get_level(),
             "primary_node": self.get_primary_node(),
+            "parent_id": self.get_parent(),
         }
         response.update(self._return_event_specifics())
         return response

backend/tests/helpers/events.py

@@ -12,7 +12,7 @@
 async def send_events(client: PrefectClient, events: list[InfrahubEvent]) -> list[Event]:
     events_data = [
         Event(
-            id=event.id,
+            id=event.meta.id,
             event=event.get_name(),
             payload=event.get_payload(),
             related=[RelatedResource(item) for item in event.get_related()],

backend/tests/unit/core/test_changelog.py

@@ -4,6 +4,7 @@
     PropertyChangelog,
     RelationshipCardinalityManyChangelog,
     RelationshipCardinalityOneChangelog,
+    RelationshipChangelogGetter,
     RelationshipPeerChangelog,
 )
 from infrahub.core.constants import DiffAction
@@ -132,6 +133,26 @@ async def test_node_changelog_creation(db: InfrahubDatabase, default_branch, ani
     )
     assert not dog1.node_changelog.parent
 
+    relationship_changelogs = RelationshipChangelogGetter(db=db, branch=default_branch)
+    secondary_changelogs = await relationship_changelogs.get_changelogs(primary_changelog=dog1.node_changelog)
+    assert len(secondary_changelogs) == 1
+
+    second = secondary_changelogs[0]
+    assert second.node_id == person1.id
+    assert second.node_kind == person1.get_kind()
+    assert list(second.relationships.keys()) == ["animals"]
+    assert second.relationships["animals"] == RelationshipCardinalityManyChangelog(
+        name="animals",
+        peers=[
+            RelationshipPeerChangelog(
+                peer_id=dog1.id,
+                peer_kind=dog1.get_kind(),
+                peer_status=DiffAction.ADDED,
+                properties={},
+            )
+        ],
+    )
+
 
 async def test_node_changelog_update_with_cardinality_one_relationship(
     db: InfrahubDatabase, default_branch, animal_person_schema

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

@@ -206,7 +206,7 @@ async def events_data(
 
 @pytest.fixture
 async def event_ids_inscope(events_data: dict[str, InfrahubEvent]) -> list[str]:
-    return [str(event.id) for event in events_data.values()]
+    return [str(event.meta.id) for event in events_data.values()]
 
 
 def filter_outofscope_events(result_data: dict, in_scope_ids: list[str]):

backend/tests/unit/task_manager/test_event.py

@@ -26,7 +26,7 @@
 
 def filter_outofscope_events(events: dict, in_scope_ids: list[str]):
     """
-    Because we can't garantee that Prefect is empty at the start of the test easily
+    Because we can't guarantee that Prefect is empty at the start of the test easily
     we need to exclude all events not created by this test suite.
     """
     filtered_events = [event for event in events["edges"] if event["node"]["id"] in in_scope_ids]
@@ -58,13 +58,13 @@ async def events_data(prefect_client: PrefectClient, branch1_id, branch2_id) ->
         "branch2_rebased": BranchRebasedEvent(branch_name="branch2", branch_id=branch2_id),
     }
 
-    await send_events(client=prefect_client, events=items.values())
+    await send_events(client=prefect_client, events=list(items.values()))
     return items
 
 
 @pytest.fixture(scope="module")
 async def event_ids_inscope(events_data: dict[str, InfrahubEvent]) -> list[str]:
-    return [str(event.id) for event in events_data.values()]
+    return [str(event.meta.id) for event in events_data.values()]
 
 
 async def test_query_no_filters(event_ids_inscope):