feat(automations): add event trigger match and inputs

This commit is contained in:
CREDO23 2026-05-29 17:48:48 +02:00
parent 3ba18c7750
commit 4ba637ea44
4 changed files with 98 additions and 0 deletions

View file

@ -0,0 +1,17 @@
"""Build run inputs from a published event."""
from __future__ import annotations
from typing import Any
from app.event_bus import Event
def event_runtime_inputs(event: Event) -> dict[str, Any]:
"""Flatten the event payload and stamp event metadata as run inputs."""
return {
**event.payload,
"event_type": event.event_type,
"event_id": event.event_id,
"occurred_at": event.occurred_at.isoformat(),
}

View file

@ -0,0 +1,16 @@
"""Pure predicate: does an event trigger fire for a given event?"""
from __future__ import annotations
from typing import Any
from app.event_bus import Event
from .filter import matches
def trigger_matches_event(params: dict[str, Any], event: Event) -> bool:
"""True when an event trigger configured with ``params`` should fire for ``event``."""
if params.get("event_type") != event.event_type:
return False
return matches(params.get("filter") or {}, event.payload)