Event System and Triggers

Infrahub uses a sophisticated event system built on Prefect to enable real-time automation and integrations. Events are emitted when significant actions occur and can trigger workflows, webhooks, and other automations.

Event Architecture

Events flow through Infrahub in this pattern:

Action occurs (node created, branch merged, etc.)
Event emitted to Prefect event system
Triggers evaluate event against conditions
Automations execute matching workflows/webhooks

Event Types

Node Events

Emitted when nodes are created, updated, or deleted:

# backend/infrahub/events/node_action.py:16
class NodeMutatedEvent(InfrahubEvent):
    event_name: ClassVar[str] = "infrahub.node.created"  # or updated/deleted
    
    kind: str  # Node kind (e.g., "NetworkDevice")
    node_id: str  # Node UUID
    action: MutationAction  # created, updated, deleted
    changelog: NodeChangelog  # Changed attributes/relationships
    fields: list[str]  # Modified fields

Event names:

infrahub.node.created
infrahub.node.updated
infrahub.node.deleted

Branch Events

Emitted for branch lifecycle operations:

# backend/infrahub/events/branch_action.py
class BranchCreatedEvent(InfrahubEvent):
    event_name: ClassVar[str] = "infrahub.branch.created"
    
    branch_name: str
    branch_id: str
    sync_with_git: bool

class BranchMergedEvent(InfrahubEvent):
    event_name: ClassVar[str] = "infrahub.branch.merged"
    
    branch_name: str
    branch_id: str
    proposed_change_id: str | None

class BranchDeletedEvent(InfrahubEvent):
    event_name: ClassVar[str] = "infrahub.branch.deleted"
    
    branch_name: str
    branch_id: str
    sync_with_git: bool

Repository Events

Emitted when repositories change:

infrahub.repository.created
infrahub.repository.updated
infrahub.repository.deleted

Artifact Events

infrahub.artifact.created
infrahub.artifact.updated
infrahub.artifact.generated

Generator Events

infrahub.generator.executed
infrahub.generator.completed

Schema Events

infrahub.schema.updated
infrahub.schema.validated

Proposed Change Events

infrahub.proposed_change.created
infrahub.proposed_change.updated
infrahub.proposed_change.merged
infrahub.proposed_change.closed

Validator Events

infrahub.validator.created
infrahub.validator.completed

Event Structure

Base Event Model

# backend/infrahub/events/models.py:148
class InfrahubEvent(BaseModel):
    meta: EventMeta  # Metadata about event
    event_name: ClassVar[str]  # Event type identifier
    
    def get_resource(self) -> dict[str, str]:
        # Primary resource this event concerns
        ...
    
    def get_related(self) -> list[dict[str, str]]:
        # Related resources (branch, account, etc.)
        ...
    
    def get_payload(self) -> dict[str, Any]:
        # Event data
        ...

Event Metadata

# backend/infrahub/events/models.py:29
class EventMeta(BaseModel):
    branch: Branch | None  # Branch where event occurred
    request_id: str  # Correlation ID
    account_id: str | None  # User who triggered event
    initiator_id: str  # Worker identity
    context: InfrahubContext  # Full context
    level: int  # Event depth (for nested events)
    has_children: bool  # Event spawns child events
    id: UUID  # Unique event ID
    parent: UUID | None  # Parent event ID
    ancestors: list[ParentEvent]  # Chain of parent events

Event Resources

Events include resource metadata for filtering:

def get_related(self) -> list[dict[str, str]]:
    related = [
        {"prefect.resource.id": __version__, "prefect.resource.role": "infrahub.version"},
        {"prefect.resource.id": self.get_id(), "prefect.resource.role": "infrahub.event"},
        {"prefect.resource.id": f"infrahub.account.{account_id}", "prefect.resource.role": "infrahub.account"},
        {"prefect.resource.id": f"infrahub.branch.{branch_id}", "prefect.resource.role": "infrahub.branch"},
    ]
    return related

Creating Events

Emit Event from Code

from infrahub.events.node_action import NodeMutatedEvent
from infrahub.events.models import EventMeta
from infrahub.core.constants import MutationAction

# Create event
event = NodeMutatedEvent(
    meta=EventMeta.from_context(context),
    kind="NetworkDevice",
    node_id=node.id,
    action=MutationAction.CREATED,
    changelog=node_changelog,
    fields=["name", "ip_address"],
)

# Emit to event service
event_service = await get_event_service()
await event_service.emit(event=event)

Event Metadata Options

# Create from context
meta = EventMeta.from_context(context)

# Create from parent event
meta = EventMeta.from_parent(parent_event)

# Create with dummy context (testing)
meta = EventMeta.with_dummy_context(branch)

Event Triggers

Triggers define conditions that cause automations to execute.

Trigger Types

# backend/infrahub/trigger/models.py
class TriggerType(StrEnum):
    ACTION = "action"  # Action-based triggers
    WEBHOOK = "webhook"  # Webhook triggers

Event Trigger Definition

# backend/infrahub/trigger/models.py
class EventTrigger(BaseModel):
    events: set[str]  # Event names to match
    match: dict[str, Any] | None = None  # Resource matching
    match_related: dict[str, Any] | None = None  # Related resource matching
    
    def get_prefect(self) -> EventTrigger:
        # Convert to Prefect EventTrigger
        ...

Example Trigger Definitions

Webhook Configuration Trigger

# backend/infrahub/webhook/triggers.py:5
TRIGGER_WEBHOOK_SETUP_UPDATE = BuiltinTriggerDefinition(
    name="webhook-configure-one",
    trigger=EventTrigger(
        events={"infrahub.node.created", "infrahub.node.updated"},
        match={
            "infrahub.node.kind": [InfrahubKind.CUSTOMWEBHOOK, InfrahubKind.STANDARDWEBHOOK],
        },
    ),
    actions=[
        ExecuteWorkflow(
            workflow=WEBHOOK_CONFIGURE_ONE,
            parameters={
                "webhook_name": "{{ event.payload['data']['changelog']['display_label'] }}",
                "event_data": {
                    "__prefect_kind": "json",
                    "value": {"__prefect_kind": "jinja", "template": "{{ event.payload['data'] | tojson }}"},
                },
            },
        ),
    ],
)

Branch Filter Trigger

trigger = EventTrigger(
    events={"infrahub.node.created"},
    match={"infrahub.node.kind": "NetworkDevice"},
    match_related={
        "prefect.resource.role": "infrahub.branch",
        "infrahub.resource.label": "main",  # Only main branch
    },
)

Exclude Branch Trigger

trigger = EventTrigger(
    events={"infrahub.node.updated"},
    match={"infrahub.node.kind": "IPAddress"},
    match_related={
        "prefect.resource.role": "infrahub.branch",
        "infrahub.resource.label": "!main",  # All except main
    },
)

Trigger Actions

Execute Workflow Action

# backend/infrahub/trigger/models.py
class ExecuteWorkflow(BaseModel):
    workflow: WorkflowDefinition  # Workflow to execute
    parameters: dict[str, Any]  # Parameters to pass
    
    def get(self, deployment_id: UUID) -> RunDeployment:
        # Create Prefect RunDeployment action
        return RunDeployment(
            deployment_id=deployment_id,
            parameters=self.parameters,
        )

Parameter Templating

Actions support Jinja2 templating for parameters:

ExecuteWorkflow(
    workflow=MY_WORKFLOW,
    parameters={
        # Direct template
        "branch_name": "{{ event.resource['infrahub.branch.name'] }}",
        
        # JSON template
        "event_data": {
            "__prefect_kind": "json",
            "value": {
                "__prefect_kind": "jinja",
                "template": "{{ event.payload['data'] | tojson }}"
            },
        },
        
        # Static value
        "timeout": 300,
    },
)

Built-in Triggers

Infrahub includes several built-in triggers:

Repository Triggers

Repository Add: Configure new repositories
Repository Update: Reconfigure on updates
Repository Delete: Clean up automations

Generator Triggers

Group Member Added: Run generator when target added
Group Member Removed: Update when target removed
Generator Definition Updated: Reconfigure generator

Webhook Triggers

Webhook Created: Configure automation
Webhook Updated: Update automation
Webhook Deleted: Remove automation

Custom Triggers

Define Trigger

from infrahub.trigger.models import BuiltinTriggerDefinition, EventTrigger, ExecuteWorkflow
from infrahub.workflows.catalogue import MY_WORKFLOW

CUSTOM_TRIGGER = BuiltinTriggerDefinition(
    name="custom-device-validation",
    trigger=EventTrigger(
        events={"infrahub.node.created", "infrahub.node.updated"},
        match={"infrahub.node.kind": "NetworkDevice"},
    ),
    actions=[
        ExecuteWorkflow(
            workflow=MY_WORKFLOW,
            parameters={
                "device_id": "{{ event.payload['data']['node_id'] }}",
                "branch": "{{ event.resource['infrahub.branch.name'] }}",
            },
        ),
    ],
)

Register Trigger

# backend/infrahub/trigger/builtin.py
from infrahub.trigger.models import BuiltinTriggerDefinition

BUILTIN_TRIGGERS: list[BuiltinTriggerDefinition] = [
    # ... existing triggers
    CUSTOM_TRIGGER,
]

Setup Triggers

# Triggers are set up automatically on startup
# Or manually via workflow:

from infrahub.trigger.tasks import trigger_configure_all

await trigger_configure_all()

Event Service

Emit Events

from infrahub.workers.dependencies import get_event_service

event_service = await get_event_service()

# Emit single event
await event_service.emit(event=my_event)

# Emit multiple events
await event_service.emit_many(events=[event1, event2, event3])

Query Events

Events are stored in Prefect and can be queried:

from prefect.client.orchestration import get_client as get_prefect_client

async with get_prefect_client(sync_client=False) as client:
    events = await client.read_events(
        occurred_before="2024-03-02T12:00:00Z",
        occurred_after="2024-03-01T00:00:00Z",
        event_name="infrahub.node.created",
    )

Event Context Propagation

Events maintain context across nested operations:

# Parent event
parent_event = NodeMutatedEvent(
    meta=EventMeta.from_context(context),
    ...
)

# Child event inherits context
child_event = ArtifactGeneratedEvent(
    meta=EventMeta.from_parent(parent_event),
    ...
)

# Child event will have:
# - parent.meta.has_children = True
# - child.meta.parent = parent.meta.id
# - child.meta.ancestors = [parent] + parent.ancestors
# - child.meta.level = parent.meta.level + 1

Event Filtering

Events can be filtered using resource metadata:

By Node Kind

match={"infrahub.node.kind": "NetworkDevice"}

By Multiple Kinds

match={"infrahub.node.kind": ["NetworkDevice", "NetworkInterface"]}

By Branch

match_related={
    "prefect.resource.role": "infrahub.branch",
    "infrahub.resource.label": "main",
}

By Account

match_related={
    "prefect.resource.role": "infrahub.account",
    "infrahub.resource.id": "user-uuid",
}

By Attribute Change

match_related={
    "prefect.resource.role": "infrahub.node.attribute_update",
    "infrahub.attribute.name": "ip_address",
}

Real-World Examples

Automatic IPAM Validation

IPAM_VALIDATION_TRIGGER = BuiltinTriggerDefinition(
    name="ipam-auto-validate",
    trigger=EventTrigger(
        events={"infrahub.node.created", "infrahub.node.updated"},
        match={"infrahub.node.kind": ["IPAddress", "IPPrefix"]},
    ),
    actions=[
        ExecuteWorkflow(
            workflow=IPAM_VALIDATION_WORKFLOW,
            parameters={
                "node_id": "{{ event.payload['data']['node_id'] }}",
                "node_kind": "{{ event.payload['data']['kind'] }}",
                "branch": "{{ event.resource['infrahub.branch.name'] }}",
            },
        ),
    ],
)

Schema Migration on Update

SCHEMA_MIGRATION_TRIGGER = BuiltinTriggerDefinition(
    name="schema-auto-migrate",
    trigger=EventTrigger(
        events={"infrahub.schema.updated"},
        match_related={
            "prefect.resource.role": "infrahub.branch",
            "infrahub.resource.label": "main",
        },
    ),
    actions=[
        ExecuteWorkflow(
            workflow=SCHEMA_APPLY_MIGRATION,
            parameters={
                "branch": "{{ event.resource['infrahub.branch.name'] }}",
            },
        ),
    ],
)

Audit Trail

AUDIT_TRIGGER = BuiltinTriggerDefinition(
    name="audit-trail",
    trigger=EventTrigger(
        events={"infrahub.node.updated", "infrahub.node.deleted"},
        match={"infrahub.node.kind": ["NetworkDevice", "IPAddress"]},
    ),
    actions=[
        ExecuteWorkflow(
            workflow=AUDIT_LOG_WORKFLOW,
            parameters={
                "event_type": "{{ event.event }}",
                "node_id": "{{ event.payload['data']['node_id'] }}",
                "account_id": "{{ event.resource['infrahub.account.id'] }}",
                "timestamp": "{{ event.occurred }}",
                "changelog": {
                    "__prefect_kind": "json",
                    "value": {"__prefect_kind": "jinja", "template": "{{ event.payload['data']['changelog'] | tojson }}"},
                },
            },
        ),
    ],
)

Best Practices

Event Design

Include all relevant context in events
Use descriptive event names
Keep event payloads focused
Include correlation IDs for tracing

Trigger Design

Make triggers specific to reduce unnecessary executions
Use match filters to target exact resources
Avoid overly broad event patterns
Test triggers in non-production branches first

Performance

Minimize event payload size
Use async workflows for long operations
Batch process when possible
Monitor trigger execution times

Debugging

Use event IDs for correlation
Check Prefect UI for event history
Review automation execution logs
Test event emission in isolation

Troubleshooting

Events Not Triggering

Verify event name matches trigger pattern
Check match/match_related filters
Ensure trigger is registered and active
Review Prefect automation status

Duplicate Triggers

Check for multiple trigger definitions
Verify trigger names are unique
Review automation list in Prefect UI

Performance Issues

Profile event emission overhead
Optimize event payload size
Review trigger execution frequency
Consider event batching

Get Started

Core Concepts

Schema & Data Modeling

Data Management

Version Control & Branching

Transformations & Artifacts

Integration & Automation

Deployment & Operations

​Event Architecture

​Event Types

​Node Events

​Branch Events

​Repository Events

​Artifact Events

​Generator Events

​Schema Events

​Proposed Change Events

​Validator Events

​Event Structure

​Base Event Model

​Event Metadata

​Event Resources

​Creating Events

​Emit Event from Code

​Event Metadata Options

​Event Triggers

​Trigger Types

​Event Trigger Definition

​Example Trigger Definitions

​Webhook Configuration Trigger

​Branch Filter Trigger

​Exclude Branch Trigger

​Trigger Actions

​Execute Workflow Action

​Parameter Templating

​Built-in Triggers

​Repository Triggers

​Generator Triggers

​Webhook Triggers

​Custom Triggers

​Define Trigger

​Register Trigger

​Setup Triggers

​Event Service

​Emit Events

​Query Events

​Event Context Propagation

​Event Filtering

​By Node Kind

​By Multiple Kinds

​By Branch

​By Account

​By Attribute Change

​Real-World Examples

​Automatic IPAM Validation

​Schema Migration on Update

​Audit Trail

​Best Practices

​Event Design

​Trigger Design

​Performance

​Debugging

​Troubleshooting

​Events Not Triggering

​Duplicate Triggers

​Performance Issues

Build docs developers (and LLMs) love