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

InMemoryDispatchBackend, a dict-based DispatchBackend for tests: sources, subscriptions, actions, accumulator buffering with claim/confirm semantics, cursor tracking, and completion records.

#dispatch.src.orxtra.dispatch._memory_backend

#dispatch.src.orxtra.dispatch._memory_backend

#InMemoryDispatchBackend

Dict-based in-memory 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

#advance_cursor

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

#is_action_completed

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

#record_completion

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

#poll_events_since

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

#inject_event

python
def inject_event(self, event: dict[str, Any]) -> None

Test helper: add an event to the in-memory store.

Search