fastrag.query.chat.base#

The ChatStore contract: sessions, chats, and messages.

Serving a chatbot means remembering conversations: anonymous sessions, the chats inside them, and the messages inside those. ChatStore is the pluggable contract for that persistence — a session’s identity and expiry, a chat’s ownership and lifecycle, and a message’s role/content/generation outcome — covering exactly what a serving layer needs and nothing about how that layer is built (no FastAPI, no HTTP types, no SSE).

A persisted ChatMessage is a different thing from Message, the model-facing turn a ChatModel consumes: a ChatMessage carries an id, timestamps, sources, and a generation outcome that a Message has no use for. to_llm_history() is the conversion between them.

Module Attributes

`MESSAGE_ROLES`	The literal values above, as plain tuples — the single source of truth both this module's `Literal` aliases and `fastrag.query.chat.schema`'s Postgres `ENUM` types are built from, so the two can never drift apart.
`DEFAULT_LIMIT`	Default and maximum page size for `ChatStore.list_chats()` and `ChatStore.list_messages()`.

Functions

to_llm_history(messages)

Turn a page of persisted messages into model-facing chat history.

Classes

`Chat`(id, session_id, created_at, expires_at)	One conversation: an ordered, linear list of messages.
`ChatMessage`(id, chat_id, created_at, role[, ...])	One persisted turn, from `user`, `assistant`, or `system`.
`ChatStore`()	Pluggable abstract persistence for sessions, chats, and messages.
`Page`(items, has_more)	One page of a ULID-cursor listing.
`Session`(id, created_at, expires_at[, ...])	One anonymous browser identity, owning its chats.

fastrag.query.chat.base.MESSAGE_ROLES: tuple[str, ...] = ('user', 'assistant', 'system')#: The literal values above, as plain tuples — the single source of truth both this module’s Literal aliases and fastrag.query.chat.schema’s Postgres ENUM types are built from, so the two can never drift apart.

fastrag.query.chat.base.DEFAULT_LIMIT = 50#: Default and maximum page size for ChatStore.list_chats() and ChatStore.list_messages().

class fastrag.query.chat.base.Session(id, created_at, expires_at, ip_address=None, country=None)[source]#

Bases: object

One anonymous browser identity, owning its chats.

Parameters:

id (str)
created_at (datetime)
expires_at (datetime)
ip_address (str | None)
country (str | None)

id#

ULID, minted on creation. The bearer credential itself — a caller who holds this id can act as this session — so it is never logged.

Type:: str

created_at#

Derived from id, so ORDER BY id and ORDER BY created_at always agree.

Type:: datetime.datetime

expires_at#

When this session (and, by cascade, its chats and their messages) stops being usable.

Type:: datetime.datetime

ip_address#

The creating request’s peer address, if known.

Type:: str | None

country#

A two-letter country code, if known.

Type:: str | None

class fastrag.query.chat.base.Chat(id, session_id, created_at, expires_at)[source]#

Bases: object

One conversation: an ordered, linear list of messages.

Parameters:

id (str)
session_id (str)
created_at (datetime)
expires_at (datetime)

id#

ULID, minted on creation.

Type:: str

session_id#

The owning Session.

Type:: str

created_at#

Derived from id.

Type:: datetime.datetime

expires_at#

When this chat (and, by cascade, its messages) stops being usable. Independent of its session’s expires_at — a chat can go idle while its session stays active.

Type:: datetime.datetime

class fastrag.query.chat.base.ChatMessage(id, chat_id, created_at, role, content=None, sources=None, finish_reason=None, error_code=None, error_message=None)[source]#

Bases: object

One persisted turn, from user, assistant, or system.

A different thing from Message, the model-facing turn a ChatModel consumes — see to_llm_history() for the conversion.

Parameters:

id (str)
chat_id (str)
created_at (datetime)
role (Literal['user', 'assistant', 'system'])
content (str | None)
sources (list[dict[str, Any]] | None)
finish_reason (Literal['stop', 'length', 'error', 'interrupted'] | None)
error_code (Literal['MODEL_ERROR', 'TIMEOUT', 'RATE_LIMIT', 'CONTEXT_TOO_LONG', 'NETWORK_ERROR', 'SERVICE_UNAVAILABLE', 'INVALID_REQUEST', 'INVALID_RESPONSE', 'INTERNAL_ERROR'] | None)
error_message (str | None)

id#

ULID, minted on creation.

Type:: str

chat_id#

The owning Chat.

Type:: str

created_at#

Derived from id.

Type:: datetime.datetime

role#

Who sent the turn.

Type:: Literal[‘user’, ‘assistant’, ‘system’]

content#

The turn’s text. None only on an assistant turn whose finish_reason is "error" and nothing was generated before it failed.

Type:: str | None

sources#

The documents the answer was grounded in, serialized as plain JSON objects (a store never interprets their shape). None for a user/system turn.

Type:: list[dict[str, Any]] | None

finish_reason#

Why an assistant turn stopped. Required for assistant turns, absent otherwise.

Type:: Literal[‘stop’, ‘length’, ‘error’, ‘interrupted’] | None

error_code#

Set exactly when finish_reason == "error".

Type:: Literal[‘MODEL_ERROR’, ‘TIMEOUT’, ‘RATE_LIMIT’, ‘CONTEXT_TOO_LONG’, ‘NETWORK_ERROR’, ‘SERVICE_UNAVAILABLE’, ‘INVALID_REQUEST’, ‘INVALID_RESPONSE’, ‘INTERNAL_ERROR’] | None

error_message#

Set exactly when finish_reason == "error".

Type:: str | None

class fastrag.query.chat.base.Page(items, has_more)[source]#

Bases: Generic[T]

One page of a ULID-cursor listing.

Parameters:

items (list[T])
has_more (bool)

items#

This page’s rows, newest-first (highest id first).

Type:: list[fastrag.query.chat.base.T]

has_more#

Whether an older row exists beyond this page — pass this page’s last item’s id as the next call’s before to fetch it.

Type:: bool

fastrag.query.chat.base.to_llm_history(messages)[source]#

Turn a page of persisted messages into model-facing chat history.

Reorders messages into chronological (oldest-first) order — the order every ChatModel expects — and drops any turn with content is None (an assistant turn that errored before generating anything), which a chat model has nothing to replay. A list_messages page is newest-first, so this is normally the conversion a caller applies right before building a prompt.

Parameters:: messages (Sequence[ChatMessage])
Return type:: list[Message]

class fastrag.query.chat.base.ChatStore[source]#

Bases: AbstractAsyncContextManager[ChatStore], ABC

Pluggable abstract persistence for sessions, chats, and messages.

Storage only: no FastAPI, no HTTP types, no SSE, no QueryEngine — a serving layer built on top of a ChatStore owns all of that.

abstractmethod async create_session(*, ttl_days, ip_address=None, country=None)[source]#

Mint and persist a new session.

Parameters:

ttl_days (int) – Days until the session expires without activity.
ip_address (str | None) – The creating request’s peer address, if known.
country (str | None) – A two-letter country code, if known.

Return type:

Session

abstractmethod async get_session(session_id)[source]#

Fetch an active session by id.

Returns:: The session, or None if session_id is unknown or its expires_at has already passed.
Parameters:: session_id (str)
Return type:: Session | None

abstractmethod async touch_session(session_id, *, ttl_days, min_bump_days)[source]#

Bump an active session’s expiry, throttled.

The expiry is only written when the new value is at least min_bump_days later than the stored one, so an active conversation stays alive without a write on every request.

Parameters:

session_id (str) – The session to touch.
ttl_days (int) – Days from now the session’s new expiry would be, if bumped.
min_bump_days (int) – Minimum headroom, in days, the candidate expiry must gain over the stored one before it is written.

Returns:

The session as it now stands (bumped or not), or None if session_id is unknown or already expired.

Return type:

Session | None

abstractmethod async create_chat(*, session_id, ttl_days)[source]#

Start a new, empty chat owned by session_id.

Parameters:

session_id (str) – The owning session.
ttl_days (int) – Days until the chat expires without activity.

Return type:

Chat

abstractmethod async get_chat(chat_id, *, session_id)[source]#

Fetch a chat, scoped to its owning session.

Returns:

The chat, or None if chat_id does not exist or does not belong to session_id.

Parameters:

chat_id (str)
session_id (str)

Return type:

Chat | None

abstractmethod async list_chats(session_id, *, before=None, limit=50)[source]#

List session_id’s chats, newest-first.

Parameters:

session_id (str) – The owning session.
before (str | None) – When given, only chats older than this chat id are returned — pass a previous page’s last item’s id to continue paging.
limit (int) – Maximum chats to return.

Return type:

Page[Chat]

abstractmethod async delete_chat(chat_id, *, session_id)[source]#

Delete a chat and every message in it.

Parameters:

chat_id (str) – The chat to delete.
session_id (str) – The owning session — chat_id is only deleted if it belongs to this session.

Returns:

True if a chat was deleted, False if chat_id did not exist or did not belong to session_id.

Return type:

bool

abstractmethod async append_message(chat_id, *, role, content=None, sources=None, finish_reason=None, error_code=None, error_message=None)[source]#

Append a new turn to a chat.

Parameters:

chat_id (str) – The chat this turn belongs to.
role (Literal['user', 'assistant', 'system']) – Who sent the turn.
content (str | None) – The turn’s text. May be None only for an assistant turn whose finish_reason is "error".
sources (list[dict[str, Any]] | None) – The documents the answer was grounded in, as plain JSON objects. Only meaningful for an assistant turn.
finish_reason (Literal['stop', 'length', 'error', 'interrupted'] | None) – Required for an assistant turn, must be absent otherwise.
error_code (Literal['MODEL_ERROR', 'TIMEOUT', 'RATE_LIMIT', 'CONTEXT_TOO_LONG', 'NETWORK_ERROR', 'SERVICE_UNAVAILABLE', 'INVALID_REQUEST', 'INVALID_RESPONSE', 'INTERNAL_ERROR'] | None) – Set exactly when finish_reason == "error".
error_message (str | None) – Set exactly when finish_reason == "error".

Return type:

ChatMessage

abstractmethod async list_messages(chat_id, *, before=None, limit=50)[source]#

List a chat’s messages, newest-first.

Parameters:

chat_id (str) – The chat to list.
before (str | None) – When given, only messages older than this message id are returned — pass a previous page’s last item’s id to continue paging.
limit (int) – Maximum messages to return.

Return type:

Page[ChatMessage]

abstractmethod async update_last_assistant_message(chat_id, *, content, sources=None, finish_reason, error_code=None, error_message=None)[source]#

Overwrite a chat’s most recent assistant turn in place.

The storage primitive behind both the /retry flow (regenerate the last answer) and the /edit flow (regenerate after editing an earlier turn): both end by replacing the chat’s latest assistant message with a new outcome, in place, rather than deleting and re-appending.

Parameters:

chat_id (str) – The chat whose latest assistant message is updated.
content (str | None) – The turn’s new text. May be None only when finish_reason == "error".
sources (list[dict[str, Any]] | None) – The documents the new answer was grounded in.
finish_reason (Literal['stop', 'length', 'error', 'interrupted']) – Why generation stopped this time.
error_code (Literal['MODEL_ERROR', 'TIMEOUT', 'RATE_LIMIT', 'CONTEXT_TOO_LONG', 'NETWORK_ERROR', 'SERVICE_UNAVAILABLE', 'INVALID_REQUEST', 'INVALID_RESPONSE', 'INTERNAL_ERROR'] | None) – Set exactly when finish_reason == "error".
error_message (str | None) – Set exactly when finish_reason == "error".

Returns:

The updated message, or None if chat_id has no assistant message yet.

Return type:

ChatMessage | None