tiny_trigger/actions.py

from __future__ import annotations

from .models import ActionEvent


def dispatch_events(
    events: list[ActionEvent],
    *,
    enable_webhooks: bool = False,
    webhook_url: str | None = None,
    timeout: float = 5.0,
) -> list[ActionEvent]:
    """Mark simulated actions and optionally POST webhook actions."""
    dispatched: list[ActionEvent] = []

    for event in events:
        if event.type != "webhook":
            dispatched.append(event.model_copy(update={"status": "simulated"}))
            continue

        target_url = event.url or webhook_url
        if not enable_webhooks:
            dispatched.append(event.model_copy(update={"status": "simulated_webhook"}))
            continue
        if not target_url:
            dispatched.append(event.model_copy(update={"status": "webhook_missing_url"}))
            continue

        try:
            import requests

            response = requests.post(target_url, json=event.payload, timeout=timeout)
            dispatched.append(
                event.model_copy(
                    update={
                        "status": "webhook_posted",
                        "response_status": response.status_code,
                    }
                )
            )
        except Exception as exc:  # pragma: no cover - network failure path
            dispatched.append(
                event.model_copy(
                    update={
                        "status": "webhook_failed",
                        "error": str(exc),
                    }
                )
            )

    return dispatched