Merge pull request #5956 from opsmill/dga-20250306-event-payload

Cleanup events payload

Author: dgarros

backend/infrahub/computed_attribute/tasks.py

@@ -3,7 +3,6 @@
 from datetime import timedelta
 from typing import TYPE_CHECKING, Any
 
-import ujson
 from infrahub_sdk.protocols import (
     CoreNode,  # noqa: TC002
     CoreTransformPython,
@@ -237,14 +236,12 @@ async def process_jinja2(
     computed_attribute_kind: str,
     context: InfrahubContext,  # noqa: ARG001
     service: InfrahubServices,
-    updated_fields: str | None = None,
+    updated_fields: list[str] | None = None,
 ) -> None:
     log = get_run_logger()
 
     await add_tags(branches=[branch_name])
-    updates: list[str] = []
-    if isinstance(updated_fields, str):
-        updates = ujson.loads(updated_fields)
+    updates: list[str] = updated_fields or []
 
     target_branch_schema = (
         branch_name if branch_name in registry.get_altered_schema_branches() else registry.default_branch
@@ -384,7 +381,13 @@ async def computed_attribute_setup(
                             "object_id": "{{ event.resource['infrahub.node.id'] }}",
                             "computed_attribute_name": computed_attribute.attribute.name,
                             "computed_attribute_kind": computed_attribute.kind,
-                            "updated_fields": "{{ event.payload['fields'] | tojson }}",
+                            "updated_fields": {
+                                "__prefect_kind": "json",
+                                "value": {
+                                    "__prefect_kind": "jinja",
+                                    "template": "{{ event.payload['data']['fields'] | tojson }}",
+                                },
+                            },
                             "context": {
                                 "__prefect_kind": "json",
                                 "value": {
@@ -457,7 +460,13 @@ async def computed_attribute_setup(
                                 "object_id": "{{ event.resource['infrahub.node.id'] }}",
                                 "computed_attribute_name": computed_attribute.attribute.name,
                                 "computed_attribute_kind": computed_attribute.kind,
-                                "updated_fields": "{{ event.payload['fields'] | tojson }}",
+                                "updated_fields": {
+                                    "__prefect_kind": "json",
+                                    "value": {
+                                        "__prefect_kind": "jinja",
+                                        "template": "{{ event.payload['data']['fields'] | tojson }}",
+                                    },
+                                },
                                 "context": {
                                     "__prefect_kind": "json",
                                     "value": {

backend/infrahub/computed_attribute/triggers.py

@@ -1,3 +1,6 @@
+from infrahub.events.branch_action import BranchCreatedEvent, BranchDeletedEvent
+from infrahub.events.repository_action import CommitUpdatedEvent
+from infrahub.events.schema_action import SchemaUpdatedEvent
 from infrahub.trigger.models import BuiltinTriggerDefinition, EventTrigger, ExecuteWorkflow
 from infrahub.workflows.catalogue import (
     COMPUTED_ATTRIBUTE_REMOVE_PYTHON,
@@ -9,7 +12,7 @@
     name="computed-attribute-python-setup-on-branch-creation",
     previous_names={"Trigger-schema-update-event"},
     description="Trigger actions on branch create event",
-    trigger=EventTrigger(events={"infrahub.branch.created"}),
+    trigger=EventTrigger(events={BranchCreatedEvent.event_name}),
     actions=[
         ExecuteWorkflow(
             workflow=COMPUTED_ATTRIBUTE_SETUP_PYTHON,
@@ -28,7 +31,7 @@
 TRIGGER_COMPUTED_ATTRIBUTE_PYTHON_SETUP_COMMIT = BuiltinTriggerDefinition(
     name="computed-attribute-python-setup-on-commit",
     description="Trigger actions on branch create event",
-    trigger=EventTrigger(events={"infrahub.repository.update_commit"}),
+    trigger=EventTrigger(events={CommitUpdatedEvent.event_name}),
     actions=[
         ExecuteWorkflow(
             workflow=COMPUTED_ATTRIBUTE_SETUP_PYTHON,
@@ -47,7 +50,7 @@
 TRIGGER_COMPUTED_ATTRIBUTE_PYTHON_CLEAN_BRANCH = BuiltinTriggerDefinition(
     name="computed-attribute-python-cleanup-on-branch-deletion",
     description="Trigger actions on branch delete event",
-    trigger=EventTrigger(events={"infrahub.branch.deleted"}),
+    trigger=EventTrigger(events={BranchDeletedEvent.event_name}),
     actions=[
         ExecuteWorkflow(
             workflow=COMPUTED_ATTRIBUTE_REMOVE_PYTHON,
@@ -64,7 +67,7 @@
 
 TRIGGER_COMPUTED_ATTRIBUTE_ALL_SCHEMA = BuiltinTriggerDefinition(
     name="computed-attribute-all-setup-on-schema-update",
-    trigger=EventTrigger(events={"infrahub.schema.update"}),
+    trigger=EventTrigger(events={SchemaUpdatedEvent.event_name}),
     actions=[
         ExecuteWorkflow(
             workflow=COMPUTED_ATTRIBUTE_SETUP,

backend/infrahub/context.py

@@ -1,8 +1,11 @@
+from typing import Any
+
 from pydantic import BaseModel, Field
 from typing_extensions import Self
 
 from infrahub.auth import AccountSession
 from infrahub.core.branch import Branch
+from infrahub.core.constants import GLOBAL_BRANCH_NAME
 
 
 class ParentEvent(BaseModel):
@@ -21,6 +24,10 @@ class BranchContext(BaseModel):
     name: str
     id: str | None = None
 
+    @property
+    def is_global(self) -> bool:
+        return self.name == GLOBAL_BRANCH_NAME
+
 
 class InfrahubContext(BaseModel):
     branch: BranchContext
@@ -37,3 +44,6 @@ def set_event(self, name: str, id: str) -> None:
             self.event.id = id
         else:
             self.event = EventContext(name=name, id=id)
+
+    def to_event(self) -> dict[str, Any]:
+        return self.model_dump(mode="json")

backend/infrahub/core/branch/tasks.py

@@ -29,7 +29,7 @@
 from infrahub.dependencies.registry import get_component_registry
 from infrahub.events.branch_action import BranchCreatedEvent, BranchDeletedEvent, BranchMergedEvent, BranchRebasedEvent
 from infrahub.events.models import EventMeta, InfrahubEvent
-from infrahub.events.node_action import NodeMutatedEvent
+from infrahub.events.node_action import get_node_event
 from infrahub.exceptions import BranchNotFoundError, MergeFailedError, ValidationError
 from infrahub.graphql.mutations.models import BranchCreateModel  # noqa: TC001
 from infrahub.log import get_log_data
@@ -46,7 +46,7 @@
 
 
 @flow(name="branch-rebase", flow_run_name="Rebase branch {branch}")
-async def rebase_branch(branch: str, context: InfrahubContext, service: InfrahubServices) -> None:
+async def rebase_branch(branch: str, context: InfrahubContext, service: InfrahubServices) -> None:  # noqa: PLR0915
     async with service.database.start_session() as db:
         log = get_run_logger()
         await add_tags(branches=[branch])
@@ -175,11 +175,11 @@ async def rebase_branch(branch: str, context: InfrahubContext, service: Infrahub
         diff=default_branch_diff, branch=obj, db=db, migration_tracker=MigrationTracker(migrations=migrations)
     )
     for action, node_changelog in changelog_collector.collect_changelogs():
-        mutate_event = NodeMutatedEvent(
+        node_event_class = get_node_event(MutationAction.from_diff_action(diff_action=action))
+        mutate_event = node_event_class(
             kind=node_changelog.node_kind,
             node_id=node_changelog.node_id,
-            data=node_changelog,
-            action=MutationAction.from_diff_action(diff_action=action),
+            changelog=node_changelog,
             fields=node_changelog.updated_fields,
             meta=EventMeta.from_parent(parent=rebase_event, branch=obj),
         )
@@ -282,11 +282,11 @@ async def merge_branch(branch: str, context: InfrahubContext, service: InfrahubS
 
         for action, node_changelog in node_events:
             meta = EventMeta.from_parent(parent=merge_event, branch=default_branch)
-            mutate_event = NodeMutatedEvent(
+            node_event_class = get_node_event(MutationAction.from_diff_action(diff_action=action))
+            mutate_event = node_event_class(
                 kind=node_changelog.node_kind,
                 node_id=node_changelog.node_id,
-                data=node_changelog,
-                action=MutationAction.from_diff_action(diff_action=action),
+                changelog=node_changelog,
                 fields=node_changelog.updated_fields,
                 meta=meta,
             )

backend/infrahub/core/constants/__init__.py

@@ -52,7 +52,7 @@ class EventType(InfrahubStringEnum):
     BRANCH_MERGED = f"{EVENT_NAMESPACE}.branch.merged"
     BRANCH_REBASED = f"{EVENT_NAMESPACE}.branch.rebased"
 
-    SCHEMA_UPDATED = f"{EVENT_NAMESPACE}.schema.update"
+    SCHEMA_UPDATED = f"{EVENT_NAMESPACE}.schema.updated"
 
     NODE_CREATED = f"{EVENT_NAMESPACE}.node.created"
     NODE_UPDATED = f"{EVENT_NAMESPACE}.node.updated"

backend/infrahub/events/__init__.py

@@ -1,4 +1,27 @@
+from .artifact_action import ArtifactCreatedEvent, ArtifactUpdatedEvent
+from .branch_action import BranchCreatedEvent, BranchDeletedEvent, BranchMergedEvent, BranchRebasedEvent
+from .group_action import GroupMemberAddedEvent, GroupMemberRemovedEvent
 from .models import EventMeta, InfrahubEvent
-from .node_action import NodeMutatedEvent
+from .node_action import NodeCreatedEvent, NodeDeletedEvent, NodeUpdatedEvent
+from .repository_action import CommitUpdatedEvent
+from .validator_action import ValidatorFailedEvent, ValidatorPassedEvent, ValidatorStartedEvent
 
-__all__ = ["EventMeta", "InfrahubEvent", "NodeMutatedEvent"]
+__all__ = [
+    "ArtifactCreatedEvent",
+    "ArtifactUpdatedEvent",
+    "BranchCreatedEvent",
+    "BranchDeletedEvent",
+    "BranchMergedEvent",
+    "BranchRebasedEvent",
+    "CommitUpdatedEvent",
+    "EventMeta",
+    "GroupMemberAddedEvent",
+    "GroupMemberRemovedEvent",
+    "InfrahubEvent",
+    "NodeCreatedEvent",
+    "NodeDeletedEvent",
+    "NodeUpdatedEvent",
+    "ValidatorFailedEvent",
+    "ValidatorPassedEvent",
+    "ValidatorStartedEvent",
+]

backend/infrahub/events/artifact_action.py

@@ -1,8 +1,6 @@
-from typing import Any
+from typing import ClassVar
 
-from pydantic import Field, computed_field
-
-from infrahub.message_bus import InfrahubMessage
+from pydantic import Field
 
 from .constants import EVENT_NAMESPACE
 from .models import InfrahubEvent
@@ -19,7 +17,6 @@ class ArtifactEvent(InfrahubEvent):
     checksum_previous: str | None = Field(default=None, description="The previous checksum of the artifact")
     storage_id: str = Field(..., description="The current storage id of the artifact")
     storage_id_previous: str | None = Field(default=None, description="The previous storage id of the artifact")
-    created: bool = Field(..., description="Indicates if the artifact was created with this event or already existed")
 
     def get_related(self) -> list[dict[str, str]]:
         related = super().get_related()
@@ -52,25 +49,10 @@ def get_resource(self) -> dict[str, str]:
             "infrahub.branch.name": self.meta.context.branch.name,
         }
 
-    def get_payload(self) -> dict[str, Any]:
-        return {
-            "node_id": self.node_id,
-            "artifact_definition_id": self.artifact_definition_id,
-            "target_id": self.target_id,
-            "target_kind": self.target_kind,
-            "checksum": self.checksum,
-            "checksum_previous": self.checksum_previous,
-            "storage_id": self.storage_id,
-            "storage_id_previous": self.storage_id_previous,
-        }
 
-    def get_messages(self) -> list[InfrahubMessage]:
-        return []
+class ArtifactCreatedEvent(ArtifactEvent):
+    event_name: ClassVar[str] = f"{EVENT_NAMESPACE}.artifact.created"
+
 
-    @computed_field
-    def event_name(self) -> str:
-        match self.created:
-            case True:
-                return f"{EVENT_NAMESPACE}.artifact.created"
-            case False:
-                return f"{EVENT_NAMESPACE}.artifact.updated"
+class ArtifactUpdatedEvent(ArtifactEvent):
+    event_name: ClassVar[str] = f"{EVENT_NAMESPACE}.artifact.updated"

backend/infrahub/events/branch_action.py

@@ -1,4 +1,6 @@
-from pydantic import Field, computed_field
+from typing import ClassVar
+
+from pydantic import Field
 
 from infrahub.message_bus import InfrahubMessage
 from infrahub.message_bus.messages.refresh_registry_branches import RefreshRegistryBranches
@@ -11,6 +13,8 @@
 class BranchDeletedEvent(InfrahubEvent):
     """Event generated when a branch has been deleted"""
 
+    event_name: ClassVar[str] = f"{EVENT_NAMESPACE}.branch.deleted"
+
     branch_name: str = Field(..., description="The name of the branch")
     branch_id: str = Field(..., description="The ID of the mutated node")
     sync_with_git: bool = Field(..., description="Indicates if the branch was extended to Git")
@@ -34,14 +38,12 @@ def get_messages(self) -> list[InfrahubMessage]:
         ]
         return events
 
-    @computed_field
-    def event_name(self) -> str:
-        return f"{EVENT_NAMESPACE}.branch.deleted"
-
 
 class BranchCreatedEvent(InfrahubEvent):
     """Event generated when a branch has been created"""
 
+    event_name: ClassVar[str] = f"{EVENT_NAMESPACE}.branch.created"
+
     branch_name: str = Field(..., description="The name of the branch")
     branch_id: str = Field(..., description="The ID of the branch")
     sync_with_git: bool = Field(..., description="Indicates if the branch was extended to Git")
@@ -65,14 +67,12 @@ def get_messages(self) -> list[InfrahubMessage]:
         ]
         return events
 
-    @computed_field
-    def event_name(self) -> str:
-        return f"{EVENT_NAMESPACE}.branch.created"
-
 
 class BranchMergedEvent(InfrahubEvent):
     """Event generated when a branch has been merged"""
 
+    event_name: ClassVar[str] = f"{EVENT_NAMESPACE}.branch.merged"
+
     branch_name: str = Field(..., description="The name of the branch")
     branch_id: str = Field(..., description="The ID of the branch")
 
@@ -83,17 +83,12 @@ def get_resource(self) -> dict[str, str]:
             "infrahub.branch.name": self.branch_name,
         }
 
-    def get_messages(self) -> list[InfrahubMessage]:
-        return []
-
-    @computed_field
-    def event_name(self) -> str:
-        return f"{EVENT_NAMESPACE}.branch.merged"
-
 
 class BranchRebasedEvent(InfrahubEvent):
     """Event generated when a branch has been rebased"""
 
+    event_name: ClassVar[str] = f"{EVENT_NAMESPACE}.branch.rebased"
+
     branch_id: str = Field(..., description="The ID of the mutated node")
     branch_name: str = Field(..., description="The name of the branch")
 
@@ -113,7 +108,3 @@ def get_messages(self) -> list[InfrahubMessage]:
             RefreshRegistryRebasedBranch(branch=self.branch_name),
         ]
         return events
-
-    @computed_field
-    def event_name(self) -> str:
-        return f"{EVENT_NAMESPACE}.branch.rebased"

backend/infrahub/events/group_action.py

@@ -1,9 +1,8 @@
-from typing import Any
+from typing import ClassVar
 
-from pydantic import Field, computed_field
+from pydantic import Field
 
 from infrahub.core.constants import MutationAction
-from infrahub.message_bus import InfrahubMessage
 
 from .constants import EVENT_NAMESPACE
 from .models import EventNode, InfrahubEvent
@@ -78,10 +77,6 @@ def get_related(self) -> list[dict[str, str]]:
 
         return related
 
-    @computed_field
-    def event_name(self) -> str:
-        return f"{EVENT_NAMESPACE}.group.{self.action.value}"
-
     def get_resource(self) -> dict[str, str]:
         return {
             "prefect.resource.id": f"infrahub.node.{self.node_id}",
@@ -91,27 +86,12 @@ def get_resource(self) -> dict[str, str]:
             "infrahub.node.root_id": self.node_id,
         }
 
-    def get_payload(self) -> dict[str, Any]:
-        return {
-            "ancestors": [ancestor.model_dump() for ancestor in self.ancestors],
-            "members": [member.model_dump() for member in self.members],
-        }
-
-    def get_messages(self) -> list[InfrahubMessage]:
-        return []
-
 
 class GroupMemberAddedEvent(GroupMutatedEvent):
     action: MutationAction = MutationAction.CREATED
-
-    @computed_field
-    def event_name(self) -> str:
-        return f"{EVENT_NAMESPACE}.group.member_added"
+    event_name: ClassVar[str] = f"{EVENT_NAMESPACE}.group.member_added"
 
 
 class GroupMemberRemovedEvent(GroupMutatedEvent):
     action: MutationAction = MutationAction.DELETED
-
-    @computed_field
-    def event_name(self) -> str:
-        return f"{EVENT_NAMESPACE}.group.member_removed"
+    event_name: ClassVar[str] = f"{EVENT_NAMESPACE}.group.member_removed"