Skip to content

Store

ConversationStore

Bases: ABC

Persistent storage for rooms, events, bindings, and participants.

Implement this ABC to plug in any storage backend (SQL, Redis, etc.). The library ships with InMemoryStore for development and testing.

create_room abstractmethod async 

create_room(room)

Persist a new room.

get_room abstractmethod async 

get_room(room_id)

Get a room by ID, or None if it doesn't exist.

update_room abstractmethod async 

update_room(room)

Update an existing room.

delete_room abstractmethod async 

delete_room(room_id)

Delete a room. Returns True if the room existed.

list_rooms abstractmethod async 

list_rooms(offset=0, limit=50)

List rooms with pagination.

find_rooms abstractmethod async 

find_rooms(organization_id=None, status=None, metadata_filter=None, *, limit=100, offset=0)

Find rooms matching the given filters.

find_latest_room abstractmethod async 

find_latest_room(participant_id, channel_type=None, status=None)

Find the most recent room for a participant.

find_room_id_by_channel abstractmethod async 

find_room_id_by_channel(channel_id, status=None)

Find a room ID that has a binding for the given channel_id.

add_event abstractmethod async 

add_event(event)

Store a new event.

get_event abstractmethod async 

get_event(event_id)

Get an event by ID.

update_event abstractmethod async 

update_event(event)

Update an existing event (e.g., mark as edited or deleted).

list_events abstractmethod async 

list_events(room_id, offset=0, limit=50, visibility_filter=None)

List events in a room with pagination and optional visibility filter.

check_idempotency abstractmethod async 

check_idempotency(room_id, key)

Check if an idempotency key has been seen. Returns True if duplicate.

get_event_count abstractmethod async 

get_event_count(room_id)

Return the total number of events in a room.

add_event_auto_index async 

add_event_auto_index(room_id, event)

Atomically assign the next index and store the event.

The default implementation reads the count and writes in two steps. Backends should override this with an atomic implementation (e.g. a single SQL transaction) to prevent race conditions on the index.

add_binding abstractmethod async 

add_binding(binding)

Attach a channel binding to a room.

get_binding abstractmethod async 

get_binding(room_id, channel_id)

Get a channel binding, or None if not attached.

update_binding abstractmethod async 

update_binding(binding)

Update an existing channel binding.

remove_binding abstractmethod async 

remove_binding(room_id, channel_id)

Detach a channel from a room. Returns True if it was attached.

list_bindings abstractmethod async 

list_bindings(room_id)

List all channel bindings for a room.

add_participant abstractmethod async 

add_participant(participant)

Add a participant to a room.

get_participant abstractmethod async 

get_participant(room_id, participant_id)

Get a participant by ID within a room.

update_participant abstractmethod async 

update_participant(participant)

Update a participant.

list_participants abstractmethod async 

list_participants(room_id)

List all participants in a room.

create_identity abstractmethod async 

create_identity(identity)

Create a new identity record.

get_identity abstractmethod async 

get_identity(identity_id)

Get an identity by ID.

resolve_identity abstractmethod async 

resolve_identity(channel_type, address)

Look up an identity by channel type and address.

link_address(identity_id, channel_type, address)

Link a channel address to an identity.

add_task abstractmethod async 

add_task(task)

Store a new task.

get_task abstractmethod async 

get_task(task_id)

Get a task by ID.

list_tasks abstractmethod async 

list_tasks(room_id, status=None)

List tasks for a room, optionally filtered by status.

update_task abstractmethod async 

update_task(task)

Update a task.

add_observation abstractmethod async 

add_observation(observation)

Store a new observation.

list_observations abstractmethod async 

list_observations(room_id)

List all observations for a room.

mark_read abstractmethod async 

mark_read(room_id, channel_id, event_id)

Mark an event as read for a channel.

mark_all_read abstractmethod async 

mark_all_read(room_id, channel_id)

Mark all events as read for a channel.

get_unread_count abstractmethod async 

get_unread_count(room_id, channel_id)

Return the number of unread events for a channel in a room.

InMemoryStore 

InMemoryStore()

Bases: ConversationStore

Dict-based in-memory store for development and testing.

add_event_auto_index async 

add_event_auto_index(room_id, event)

Atomically assign index = len(room_events) and append.

PostgresStore 

PostgresStore(dsn=None, pool=None)

Bases: ConversationStore

PostgreSQL-backed conversation store using asyncpg.

init async 

init(min_size=2, max_size=10)

Create the connection pool (if needed) and ensure schema exists.

close async 

close()

Release the connection pool if we own it.

add_event_auto_index async 

add_event_auto_index(room_id, event)

Atomically assign the next index and store the event in one transaction.

Uses SELECT ... FOR UPDATE on the rooms table to serialise concurrent index assignments for the same room, preventing duplicate indices under READ COMMITTED isolation.