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

Subscription and event-source CRUD: subscribe()/unsubscribe() manage principal-owned dispatch subscriptions and their action chains; create_source() and friends register and manage event sources.

#services.src.orxtra.services._dispatch

#services.src.orxtra.services._dispatch

#_resolve_action_from_dict

python
def _resolve_action_from_dict(action_data: dict[str, Any]) -> Action

Resolve a plain dict to a typed Action instance.

Detects the action type from dict keys and validates via pydantic.

#subscribe

python
async def subscribe(backend: DispatchBackend, caller_principal: Principal, filter_pred: FilterPredicate, actions: list[dict[str, Any]], *, storage: str='persistent') -> UUID

Create a subscription with actions, owned by the calling principal.

Thin wrapper: builds a Subscription from the filter predicate, attributing ownership to caller_principal (the authenticated actor derived at the dispatch choke point), persists it via the backend, then creates SubscriptionActions for each action dict in order. Action dicts are resolved to typed Action instances before storage.

A subscription is operational state owned by its principal: the FK CASCADEs, so deleting the owner deletes the subscription.

#unsubscribe

python
async def unsubscribe(backend: DispatchBackend, subscription_id: UUID) -> None

Disable and delete a subscription and its actions.

#list_subscriptions

python
async def list_subscriptions(backend: DispatchBackend, *, enabled_only: bool=True, principal_id: UUID | None=None) -> list[Subscription]

List subscriptions, optionally filtering to enabled-only and by owner.

#create_source

python
async def create_source(pool: asyncpg.Pool | None, backend: DispatchBackend, principal_storage: PrincipalStorage, caller_principal: Principal, slug: str, name: str, *, credential_id: UUID | None=None, config: dict[str, Any] | None=None) -> UUID

Create a new event source, minting the source's identity at birth.

Flow, mirroring the runs vertical:

  1. If a credential_id is supplied, validate it exists against auth's

credentials table via pool -- an unknown id is a hard error. This closes the honor-system gap where a source could reference a non-existent credential. Reading auth's tables is legal: the single-writer rule bars writes to another module's tables, not reads.

  1. Generate the source id client-side so the source's principal can exist

before the row that FKs into it.

  1. Mint the source principal (kind=source, external_ref=source id,

display_name=slug).

  1. Persist the source attributed to the caller (created_by).

#get_source

python
async def get_source(backend: DispatchBackend, source_id: UUID) -> Source | None

Get a source by ID, or None if not found.

#get_source_by_slug

python
async def get_source_by_slug(backend: DispatchBackend, slug: str) -> Source | None

Get a source by slug, or None if not found.

#list_sources

python
async def list_sources(backend: DispatchBackend) -> list[Source]

List all registered sources.

#delete_source

python
async def delete_source(backend: DispatchBackend, source_id: UUID) -> None

Delete a source by ID.

Search