orxtra v0.13.0 /services.src.orxtra.services._events
Updated
On this page

Event ingestion and streaming: fire_event() writes to the trace store attributed to a caller principal, fire_blocking() is its sync wrapper, event_stream() yields parsed events from an EventBus.

#services.src.orxtra.services._events

#services.src.orxtra.services._events

#fire_event

python
async def fire_event(pool: asyncpg.Pool, caller_principal: Principal, *, run_id: UUID | None, event_name: str, payload: dict[str, Any] | None=None, idempotency_key: str | None=None) -> tuple[UUID, bool]

Fire an event into the trace store, attributed to caller_principal.

Every event has an actor: the capability path injects the authenticated caller's principal; direct callers (the webhook receiver, the dispatch worker's re-fire) pass the source or system principal explicitly.

Returns (event_id, inserted). When idempotency_key is provided and the key already exists, inserted is False and the existing event's id is returned.

#fire_blocking

python
def fire_blocking(pool: asyncpg.Pool, caller_principal: Principal, *, run_id: UUID | None, event_name: str, payload: dict[str, Any] | None=None, idempotency_key: str | None=None) -> tuple[UUID, bool]

Synchronous wrapper around fire_event for non-async contexts.

#event_stream

python
async def event_stream(bus: EventBus, *, channel: str=EVENTS_CHANNEL, run_id: UUID | None=None, event_types: list[str] | None=None) -> AsyncIterator[dict[str, Any]]

Async generator that yields parsed events from an EventBus subscription.

Subscribes to the given channel and yields each event as a parsed dict. Optional filters narrow the stream to a specific run_id or event types.

Search