Merge pull request #5877 from opsmill/dga-20250226-webhook

Refactoring of Webhook to support `event_type` and `branch_scope`

Author: dgarros

backend/infrahub/computed_attribute/triggers.py

@@ -12,7 +12,7 @@
     trigger=EventTrigger(events={"infrahub.branch.created"}),
     actions=[
         ExecuteWorkflow(
-            name=COMPUTED_ATTRIBUTE_SETUP_PYTHON.name,
+            workflow=COMPUTED_ATTRIBUTE_SETUP_PYTHON,
             parameters={
                 "branch_name": "{{ event.resource['infrahub.branch.name'] }}",
                 "trigger_updates": False,
@@ -31,7 +31,7 @@
     trigger=EventTrigger(events={"infrahub.repository.update_commit"}),
     actions=[
         ExecuteWorkflow(
-            name=COMPUTED_ATTRIBUTE_SETUP_PYTHON.name,
+            workflow=COMPUTED_ATTRIBUTE_SETUP_PYTHON,
             parameters={
                 "branch_name": "{{ event.resource['infrahub.branch.name'] }}",
                 "commit": "{{ event.payload['commit'] }}",
@@ -50,7 +50,7 @@
     trigger=EventTrigger(events={"infrahub.branch.deleted"}),
     actions=[
         ExecuteWorkflow(
-            name=COMPUTED_ATTRIBUTE_REMOVE_PYTHON.name,
+            workflow=COMPUTED_ATTRIBUTE_REMOVE_PYTHON,
             parameters={
                 "branch_name": "{{ event.resource['infrahub.branch.name'] }}",
                 "context": {
@@ -67,7 +67,7 @@
     trigger=EventTrigger(events={"infrahub.schema.update"}),
     actions=[
         ExecuteWorkflow(
-            name=COMPUTED_ATTRIBUTE_SETUP.name,
+            workflow=COMPUTED_ATTRIBUTE_SETUP,
             parameters={
                 "branch_name": "{{ event.resource['infrahub.branch.name'] }}",
                 "context": {
@@ -77,7 +77,7 @@
             },
         ),
         ExecuteWorkflow(
-            name=COMPUTED_ATTRIBUTE_SETUP_PYTHON.name,
+            workflow=COMPUTED_ATTRIBUTE_SETUP_PYTHON,
             parameters={
                 "branch_name": "{{ event.resource['infrahub.branch.name'] }}",
                 "context": {

backend/infrahub/core/constants/__init__.py

@@ -43,6 +43,29 @@
 
 NULL_VALUE = "NULL"
 
+EVENT_NAMESPACE = "infrahub"
+
+
+class EventType(InfrahubStringEnum):
+    BRANCH_CREATED = f"{EVENT_NAMESPACE}.branch.created"
+    BRANCH_DELETED = f"{EVENT_NAMESPACE}.branch.deleted"
+    BRANCH_MERGED = f"{EVENT_NAMESPACE}.branch.merged"
+    BRANCH_REBASED = f"{EVENT_NAMESPACE}.branch.rebased"
+
+    SCHEMA_UPDATED = f"{EVENT_NAMESPACE}.schema.update"
+
+    NODE_CREATED = f"{EVENT_NAMESPACE}.node.created"
+    NODE_UPDATED = f"{EVENT_NAMESPACE}.node.updated"
+    NODE_DELETED = f"{EVENT_NAMESPACE}.node.deleted"
+
+    GROUP_MEMBER_ADDED = f"{EVENT_NAMESPACE}.group.member_added"
+    GROUP_MEMBER_REMOVED = f"{EVENT_NAMESPACE}.group.member_removed"
+
+    REPOSITORY_UPDATE_COMMIT = f"{EVENT_NAMESPACE}.repository.update_commit"
+
+    ARTIFACT_CREATED = f"{EVENT_NAMESPACE}.artifact.created"
+    ARTIFACT_UPDATED = f"{EVENT_NAMESPACE}.artifact.updated"
+
 
 class PermissionLevel(enum.Flag):
     READ = 1

backend/infrahub/core/protocols.py

@@ -197,6 +197,8 @@ class CoreValidator(CoreNode):
 
 class CoreWebhook(CoreNode):
     name: String
+    event_type: Enum
+    branch_scope: Dropdown
     description: StringOptional
     url: URL
     validate_certificates: BooleanOptional

backend/infrahub/core/schema/definitions/core/webhook.py

@@ -1,7 +1,4 @@
-from infrahub.core.constants import (
-    BranchSupportType,
-    InfrahubKind,
-)
+from infrahub.core.constants import AllowOverrideType, BranchSupportType, EventType, InfrahubKind
 
 core_webhook = {
     "name": "Webhook",
@@ -12,11 +9,48 @@
     "order_by": ["name__value"],
     "display_labels": ["name__value"],
     "include_in_menu": False,
+    "icon": "mdi:webhook",
     "branch": BranchSupportType.AGNOSTIC.value,
     "uniqueness_constraints": [["name__value"]],
     "attributes": [
         {"name": "name", "kind": "Text", "unique": True, "order_weight": 1000},
-        {"name": "description", "kind": "Text", "optional": True, "order_weight": 2000},
+        {
+            "name": "event_type",
+            "kind": "Text",
+            "enum": ["all"] + EventType.available_types(),
+            "default_value": "all",
+            "order_weight": 1500,
+            "description": "The event type that triggers the webhook",
+        },
+        {
+            "name": "branch_scope",
+            "kind": "Dropdown",
+            "choices": [
+                {
+                    "name": "all_branches",
+                    "label": "All Branches",
+                    "description": "All branches",
+                    "color": "#fef08a",
+                },
+                {
+                    "name": "default_branch",
+                    "label": "Default Branch",
+                    "description": "Only the default branch",
+                    "color": "#86efac",
+                },
+                {
+                    "name": "other_branches",
+                    "label": "Other Branches",
+                    "description": "All branches except the default branch",
+                    "color": "#e5e7eb",
+                },
+            ],
+            "default_value": "default_branch",
+            "optional": False,
+            "order_weight": 2000,
+            "allow_override": AllowOverrideType.NONE,
+        },
+        {"name": "description", "kind": "Text", "optional": True, "order_weight": 2500},
         {"name": "url", "kind": "URL", "order_weight": 3000},
         {
             "name": "validate_certificates",

backend/infrahub/git/integrator.py

@@ -1151,7 +1151,7 @@ async def execute_python_check(
 
     @task(name="python-transform-execute", task_run_name="Execute Python Transform", cache_policy=NONE)  # type: ignore[arg-type]
     async def execute_python_transform(
-        self, branch_name: str, commit: str, location: str, client: InfrahubClient, data: Optional[dict] = None
+        self, branch_name: str, commit: str, location: str, client: InfrahubClient, data: dict | None = None
     ) -> Any:
         """Execute A Python Transform stored in the repository."""
         log = get_run_logger()

backend/infrahub/menu/menu.py

@@ -261,32 +261,11 @@ def _extract_node_icon(model: MainSchemaTypes) -> str:
                 namespace="Builtin",
                 name="Webhooks",
                 label="Webhooks",
-                icon=_extract_node_icon(infrahub_schema.get(InfrahubKind.CUSTOMWEBHOOK)),
+                icon=_extract_node_icon(infrahub_schema.get(InfrahubKind.WEBHOOK)),
+                kind=InfrahubKind.WEBHOOK,
                 protected=True,
                 section=MenuSection.INTERNAL,
                 order_weight=3000,
-                children=[
-                    MenuItemDefinition(
-                        namespace="Builtin",
-                        name="WebhookStandard",
-                        label="Webhooks",
-                        kind=InfrahubKind.STANDARDWEBHOOK,
-                        icon=_extract_node_icon(infrahub_schema.get(InfrahubKind.STANDARDWEBHOOK)),
-                        protected=True,
-                        section=MenuSection.INTERNAL,
-                        order_weight=1000,
-                    ),
-                    MenuItemDefinition(
-                        namespace="Builtin",
-                        name="WebhookCustom",
-                        label="Custom Webhooks",
-                        kind=InfrahubKind.CUSTOMWEBHOOK,
-                        icon=_extract_node_icon(infrahub_schema.get(InfrahubKind.CUSTOMWEBHOOK)),
-                        protected=True,
-                        section=MenuSection.INTERNAL,
-                        order_weight=2000,
-                    ),
-                ],
             ),
             MenuItemDefinition(
                 namespace="Builtin",

backend/infrahub/message_bus/operations/event/worker.py

@@ -2,11 +2,8 @@
 
 from infrahub.message_bus import messages
 from infrahub.services import InfrahubServices
-from infrahub.workflows.catalogue import WEBHOOK_CONFIGURE
 
 
 @flow(name="event-worker-newprimary-api")
 async def new_primary_api(message: messages.EventWorkerNewPrimaryAPI, service: InfrahubServices) -> None:
     service.log.info("api_worker promoted to primary", worker_id=message.worker_id)
-
-    await service.workflow.submit_workflow(workflow=WEBHOOK_CONFIGURE)

backend/infrahub/trigger/catalogue.py

@@ -4,9 +4,10 @@
     TRIGGER_COMPUTED_ATTRIBUTE_PYTHON_SETUP_BRANCH,
     TRIGGER_COMPUTED_ATTRIBUTE_PYTHON_SETUP_COMMIT,
 )
+from infrahub.trigger.models import TriggerDefinition
 from infrahub.webhook.triggers import TRIGGER_WEBHOOK_SETUP_UPDATE
 
-triggers = [
+builtin_triggers: list[TriggerDefinition] = [
     TRIGGER_COMPUTED_ATTRIBUTE_ALL_SCHEMA,
     TRIGGER_COMPUTED_ATTRIBUTE_PYTHON_CLEAN_BRANCH,
     TRIGGER_COMPUTED_ATTRIBUTE_PYTHON_SETUP_BRANCH,

backend/infrahub/trigger/models.py

@@ -1,19 +1,26 @@
+from __future__ import annotations
+
 from datetime import timedelta
 from enum import Enum
-from typing import Any
-from uuid import UUID
+from typing import TYPE_CHECKING, Any
 
 from prefect.events.actions import RunDeployment
 from prefect.events.schemas.automations import EventTrigger as PrefectEventTrigger
 from prefect.events.schemas.automations import Posture
 from prefect.events.schemas.events import ResourceSpecification
 from pydantic import BaseModel, Field
 
+from infrahub.workflows.models import WorkflowDefinition  # noqa: TC001
+
 from .constants import NAME_SEPARATOR
 
+if TYPE_CHECKING:
+    from uuid import UUID
+
 
 class TriggerType(str, Enum):
     BUILTIN = "builtin"
+    WEBHOOK = "webhook"
     # OBJECT = "object"
     # COMPUTED_ATTR = "computed_attr"
 
@@ -35,19 +42,44 @@ def get_prefect(self) -> PrefectEventTrigger:
 
 
 class ExecuteWorkflow(BaseModel):
-    name: str
+    workflow: WorkflowDefinition
     parameters: dict[str, Any] = Field(default_factory=dict)
 
+    @property
+    def name(self) -> str:
+        return self.workflow.name
+
     def get_prefect(self, mapping: dict[str, UUID]) -> RunDeployment:
         deployment_id = mapping[self.name]
+        return self.get(deployment_id)
 
+    def get(self, id: UUID) -> RunDeployment:
         return RunDeployment(
             source="selected",
-            deployment_id=deployment_id,
+            deployment_id=id,
             parameters=self.parameters,
             job_variables={},
         )
 
+    def validate_parameters(self) -> None:
+        if not self.parameters:
+            return
+
+        workflow_params = self.workflow.get_parameters()
+        workflow_required_params = [p.name for p in workflow_params.values() if p.required]
+        trigger_params = list(self.parameters.keys())
+
+        missing_required_params = set(workflow_required_params) - set(trigger_params)
+        wrong_params = set(trigger_params) - set(workflow_params)
+
+        if missing_required_params:
+            raise ValueError(
+                f"Missing required parameters: {missing_required_params} for workflow {self.workflow.name}"
+            )
+
+        if wrong_params:
+            raise ValueError(f"Workflow {self.workflow.name} doesn't support parameters: {wrong_params}")
+
 
 class TriggerDefinition(BaseModel):
     name: str
@@ -64,6 +96,10 @@ def get_deployment_names(self) -> list[str]:
     def generate_name(self) -> str:
         return f"{self.type.value}{NAME_SEPARATOR}{self.name}"
 
+    def validate_actions(self) -> None:
+        for action in self.actions:
+            action.validate_parameters()
+
 
 class BuiltinTriggerDefinition(TriggerDefinition):
     type: TriggerType = TriggerType.BUILTIN

backend/infrahub/trigger/tasks.py

@@ -1,22 +1,27 @@
 from typing import TYPE_CHECKING
 
-from infrahub_sdk.utils import compare_lists
 from prefect import get_run_logger, task
 from prefect.automations import AutomationCore
 from prefect.cache_policies import NONE
 from prefect.client.orchestration import PrefectClient
 from prefect.client.schemas.filters import DeploymentFilter, DeploymentFilterName
 
-from .catalogue import triggers
-from .constants import DEPRECATED_STATIC_TRIGGER_NAMES
+from infrahub.trigger.models import TriggerDefinition
+
+# from .catalogue import triggers
 from .models import TriggerType
 
 if TYPE_CHECKING:
     from uuid import UUID
 
 
-@task(name="trigger-setup", task_run_name="Setup triggers in task-manager", cache_policy=NONE)  # type: ignore[arg-type]
-async def setup_triggers(client: PrefectClient) -> None:
+@task(name="trigger-setup", task_run_name="Setup triggers of type {trigger_type.value}", cache_policy=NONE)  # type: ignore[arg-type]
+async def setup_triggers(
+    client: PrefectClient,
+    triggers: list[TriggerDefinition],
+    trigger_type: TriggerType = TriggerType.BUILTIN,
+    deprecated_triggers: list[str] | None = None,
+) -> None:
     log = get_run_logger()
 
     # -------------------------------------------------------------
@@ -32,15 +37,15 @@ async def setup_triggers(client: PrefectClient) -> None:
     deployments_mapping: dict[str, UUID] = {name: item.id for name, item in deployments.items()}
     existing_automations = {item.name: item for item in await client.read_automations()}
 
-    builtin_automations = [
-        item.name for item in await client.read_automations() if item.name.startswith(TriggerType.BUILTIN.value)
+    trigger_automations = [
+        item.name for item in await client.read_automations() if item.name.startswith(trigger_type.value)
     ]
     trigger_names = [trigger.generate_name() for trigger in triggers]
 
-    _, to_delete, _ = compare_lists(list1=builtin_automations, list2=trigger_names)
+    to_delete = set(trigger_automations) - set(trigger_names)
 
     # -------------------------------------------------------------
-    # Create or Update all builtin triggers
+    # Create or Update all triggers
     # -------------------------------------------------------------
     for trigger in triggers:
         automation = AutomationCore(
@@ -61,7 +66,7 @@ async def setup_triggers(client: PrefectClient) -> None:
             log.info(f"{trigger.name} Created")
 
     # -------------------------------------------------------------
-    # Delete Builtin Triggers that shouldn't be there
+    # Delete Triggers that shouldn't be there
     # -------------------------------------------------------------
     for item_to_delete in to_delete:
         existing_automation = existing_automations.get(item_to_delete)
@@ -75,11 +80,12 @@ async def setup_triggers(client: PrefectClient) -> None:
     # -------------------------------------------------------------
     # Delete Deprecated triggers
     # -------------------------------------------------------------
-    for trigger_name in DEPRECATED_STATIC_TRIGGER_NAMES:
-        existing_automation = existing_automations.get(trigger_name)
+    if deprecated_triggers:
+        for trigger_name in deprecated_triggers:
+            existing_automation = existing_automations.get(trigger_name)
 
-        if not existing_automation:
-            continue
+            if not existing_automation:
+                continue
 
-        await client.delete_automation(automation_id=existing_automation.id)
-        log.info(f"{trigger_name} Deleted")
+            await client.delete_automation(automation_id=existing_automation.id)
+            log.info(f"{trigger_name} Deleted")