Spaces:
Sleeping
Sleeping
File size: 1,584 Bytes
d725335 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | 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
|