orxtra v0.13.0 /dispatch.src.orxtra.dispatch._pg_backend
Updated
On this page

PgDispatchBackend, the asyncpg-backed DispatchBackend over dispatch-owned tables (sources, subscriptions, subscription_actions, accumulator_buffer, dispatch_cursor), with Action (de)serialization.

#dispatch.src.orxtra.dispatch._pg_backend

#dispatch.src.orxtra.dispatch._pg_backend

#_serialize_action

python
def _serialize_action(action: ScriptAction | LogAction | WorkflowAction | EventAction | NotifyAction) -> tuple[str, str]

Decompose an Action into (action_type, action_config_json).

#_deserialize_action

python
def _deserialize_action(action_type: str, action_config: str) -> Action

Reconstruct an Action from DB columns.

#PgDispatchBackend

asyncpg-backed implementation of DispatchBackend.

#create_source

python
async def create_source(self, source: Source) -> UUID

#get_source

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

#get_source_by_slug

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

#list_sources

python
async def list_sources(self) -> list[Source]

#delete_source

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

#create_subscription

python
async def create_subscription(self, subscription: Subscription) -> UUID

#get_subscription

python
async def get_subscription(self, sub_id: UUID) -> Subscription | None

#list_subscriptions

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

#update_subscription

python
async def update_subscription(self, sub_id: UUID, *, enabled: bool) -> None

#delete_subscription

python
async def delete_subscription(self, sub_id: UUID) -> None

#create_action

python
async def create_action(self, action: SubscriptionAction) -> UUID

#list_actions

python
async def list_actions(self, sub_id: UUID) -> list[SubscriptionAction]

#delete_actions

python
async def delete_actions(self, sub_id: UUID) -> None

#buffer_event

python
async def buffer_event(self, entry: AccumulatorEntry) -> UUID

#claim_batch

python
async def claim_batch(self, action_id: UUID, limit: int=100) -> list[AccumulatorEntry]

#confirm_batch

python
async def confirm_batch(self, entry_ids: list[UUID]) -> None

#pending_count

python
async def pending_count(self, action_id: UUID) -> int

#get_cursor_position

python
async def get_cursor_position(self, cursor_name: str) -> UUID | None

Return the last_processed_event_id for cursor_name, or None.

#advance_cursor

python
async def advance_cursor(self, cursor_name: str, event_id: UUID) -> None

Upsert the cursor position for cursor_name.

#is_action_completed

python
async def is_action_completed(self, event_id: UUID, action_id: UUID) -> bool

Return True if a completion record exists for this event+action.

#record_completion

python
async def record_completion(self, event_id: UUID, action_id: UUID, result_status: str) -> None

Insert a completion record (idempotent via unique constraint).

#poll_events_since

python
async def poll_events_since(self, since_id: UUID | None, limit: int=100) -> list[dict[str, Any]]

Return events after since_id, ordered by id (UUIDv7 = time-ordered).

Each row is returned as a dict with keys: id, run_id, task_id, event_type, principal_id, data, idempotency_key, created_at.

Search