Convert BranchDelete task and event to Prefect

Author: dgarros

backend/infrahub/core/branch/tasks.py

@@ -16,12 +16,13 @@
 from infrahub.core.validators.models.validate_migration import SchemaValidateMigrationData
 from infrahub.core.validators.tasks import schema_validate_migrations
 from infrahub.dependencies.registry import get_component_registry
+from infrahub.events.branch_action import BranchDeleteEvent
 from infrahub.exceptions import ValidationError
 from infrahub.log import get_log_data
 from infrahub.message_bus import Meta, messages
 from infrahub.services import services
 from infrahub.worker import WORKER_IDENTITY
-from infrahub.workflows.catalogue import IPAM_RECONCILIATION
+from infrahub.workflows.catalogue import BRANCH_CANCEL_PROPOSED_CHANGES, IPAM_RECONCILIATION
 from infrahub.workflows.utils import add_branch_tag
 
 
@@ -199,3 +200,18 @@ async def merge_branch(branch: str, conflict_resolution: dict[str, bool] | None
         meta=Meta(initiator_id=WORKER_IDENTITY, request_id=request_id),
     )
     await service.send(message=message)
+
+
+@flow(name="branch-delete")
+async def delete_branch(branch: str) -> None:
+    service = services.service
+
+    await add_branch_tag(branch_name=branch)
+
+    obj = await Branch.get_by_name(db=service.database, name=str(branch))
+    event = BranchDeleteEvent(branch=branch, branch_id=obj.get_id(), sync_with_git=obj.sync_with_git)
+    await obj.delete(db=service.database)
+
+    await service.workflow.submit_workflow(workflow=BRANCH_CANCEL_PROPOSED_CHANGES, parameters={"branch_name": branch})
+
+    await service.event.send(event=event)

backend/infrahub/core/node/standard.py

@@ -41,6 +41,11 @@ class StandardNode(BaseModel):
     def get_type(cls) -> str:
         return cls.__name__
 
+    def get_id(self) -> str:
+        if not self.id:
+            raise ValueError("id isn't defined yet")
+        return self.id
+
     @staticmethod
     def guess_field_type(field: FieldInfo) -> Any:
         """Return the type of a Pydantic model field.

backend/infrahub/events/branch_action.py

@@ -0,0 +1,35 @@
+from pydantic import Field
+
+from infrahub.message_bus import InfrahubMessage
+from infrahub.message_bus.messages.event_branch_delete import EventBranchDelete
+from infrahub.message_bus.messages.refresh_registry_branches import RefreshRegistryBranches
+
+from .models import InfrahubBranchEvent
+
+
+class BranchDeleteEvent(InfrahubBranchEvent):
+    """Event generated when a branch has been deleted"""
+
+    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")
+
+    def get_name(self) -> str:
+        return f"{self.get_event_namespace()}.branch.deleted"
+
+    def get_resource(self) -> dict[str, str]:
+        return {
+            "prefect.resource.id": f"infrahub.branch.{self.branch}",
+            "infrahub.branch.id": self.branch_id,
+        }
+
+    def get_messages(self) -> list[InfrahubMessage]:
+        events = [
+            EventBranchDelete(
+                branch=self.branch,
+                branch_id=self.branch_id,
+                sync_with_git=self.sync_with_git,
+                meta=self.get_message_meta(),
+            ),
+            RefreshRegistryBranches(),
+        ]
+        return events

backend/infrahub/events/models.py

@@ -27,7 +27,7 @@ def get_name(self) -> str:
     def get_resource(self) -> dict[str, str]:
         raise NotImplementedError
 
-    def get_message(self) -> InfrahubMessage:
+    def get_messages(self) -> list[InfrahubMessage]:
         raise NotImplementedError
 
     def get_related(self) -> list[dict[str, str]]:

backend/infrahub/events/node_action.py

@@ -3,6 +3,7 @@
 from pydantic import Field
 
 from infrahub.core.constants import MutationAction
+from infrahub.message_bus import InfrahubMessage
 from infrahub.message_bus.messages.event_node_mutated import EventNodeMutated
 
 from .models import InfrahubBranchEvent
@@ -29,12 +30,14 @@ def get_resource(self) -> dict[str, str]:
     def get_payload(self) -> dict[str, Any]:
         return self.data
 
-    def get_message(self) -> EventNodeMutated:
-        return EventNodeMutated(
-            branch=self.branch,
-            kind=self.kind,
-            node_id=self.node_id,
-            action=self.action.value,
-            data=self.data,
-            meta=self.get_message_meta(),
-        )
+    def get_messages(self) -> list[InfrahubMessage]:
+        return [
+            EventNodeMutated(
+                branch=self.branch,
+                kind=self.kind,
+                node_id=self.node_id,
+                action=self.action.value,
+                data=self.data,
+                meta=self.get_message_meta(),
+            )
+        ]

backend/infrahub/graphql/mutations/branch.py

@@ -26,7 +26,7 @@
 from infrahub.log import get_log_data, get_logger
 from infrahub.message_bus import Meta, messages
 from infrahub.worker import WORKER_IDENTITY
-from infrahub.workflows.catalogue import BRANCH_MERGE, BRANCH_REBASE
+from infrahub.workflows.catalogue import BRANCH_DELETE, BRANCH_MERGE, BRANCH_REBASE
 
 from ..types import BranchType
 
@@ -137,22 +137,10 @@ class Arguments:
     async def mutate(cls, root: dict, info: GraphQLResolveInfo, data: BranchNameInput) -> Self:
         context: GraphqlContext = info.context
 
-        async with UserTask.from_graphql_context(title=f"Delete branch: {data['name']}", context=context):
-            obj = await Branch.get_by_name(db=context.db, name=str(data.name))
-            await obj.delete(db=context.db)
-
-            if context.service:
-                log_data = get_log_data()
-                request_id = log_data.get("request_id", "")
-                message = messages.EventBranchDelete(
-                    branch=obj.name,
-                    branch_id=str(obj.id),
-                    sync_with_git=obj.sync_with_git,
-                    meta=Meta(request_id=request_id),
-                )
-                await context.service.send(message=message)
-
-            return cls(ok=True)
+        obj = await Branch.get_by_name(db=context.db, name=str(data.name))
+        assert context.service
+        await context.service.workflow.execute_workflow(workflow=BRANCH_DELETE, parameters={"branch": obj.name})
+        return cls(ok=True)
 
 
 class BranchUpdate(Mutation):

backend/infrahub/message_bus/messages/__init__.py

@@ -33,14 +33,12 @@
 from .request_generatordefinition_check import RequestGeneratorDefinitionCheck
 from .request_generatordefinition_run import RequestGeneratorDefinitionRun
 from .request_graphqlquerygroup_update import RequestGraphQLQueryGroupUpdate
-from .request_proposed_change_cancel import RequestProposedChangeCancel
 from .request_proposedchange_pipeline import RequestProposedChangePipeline
 from .request_repository_checks import RequestRepositoryChecks
 from .request_repository_userchecks import RequestRepositoryUserChecks
 from .schema_migration_path import SchemaMigrationPath, SchemaMigrationPathResponse
 from .schema_validator_path import SchemaValidatorPath, SchemaValidatorPathResponse
 from .send_echo_request import SendEchoRequest, SendEchoRequestResponse
-from .trigger_proposed_change_cancel import TriggerProposedChangeCancel
 from .trigger_webhook_actions import TriggerWebhookActions
 
 MESSAGE_MAP: dict[str, type[InfrahubMessage]] = {
@@ -73,7 +71,6 @@
     "request.generator_definition.check": RequestGeneratorDefinitionCheck,
     "request.generator_definition.run": RequestGeneratorDefinitionRun,
     "request.graphql_query_group.update": RequestGraphQLQueryGroupUpdate,
-    "request.proposed_change.cancel": RequestProposedChangeCancel,
     "request.proposed_change.data_integrity": RequestProposedChangeDataIntegrity,
     "request.proposed_change.pipeline": RequestProposedChangePipeline,
     "request.proposed_change.refresh_artifacts": RequestProposedChangeRefreshArtifacts,
@@ -84,7 +81,6 @@
     "request.repository.checks": RequestRepositoryChecks,
     "request.repository.user_checks": RequestRepositoryUserChecks,
     "send.echo.request": SendEchoRequest,
-    "trigger.proposed_change.cancel": TriggerProposedChangeCancel,
     "trigger.webhook.actions": TriggerWebhookActions,
 }