fastrag.query.chat.postgres#

PostgreSQL-backed ChatStore.

The schema is self-managed: __aenter__ creates the target Postgres schema, its three tables, and its three ENUM types if they are absent — the same pattern PostgresCache established for the ingestion cache. Nothing here touches Alembic or any shared declarative Base.

Every driver failure is translated into a ChatStoreError (or its transient subclass), including a failure to establish the connection at all (bad password, unreachable host, DNS) — closed in PostgresChatStore.__aenter__/_connection.

Classes

PostgresChatStore(*, database_url[, schema, ...])

Persist sessions, chats, and messages on a PostgreSQL server.
class fastrag.query.chat.postgres.PostgresChatStore(*, database_url, schema='fastrag_chat', pool_size=5)[source]#

Bases: ChatStore

Persist sessions, chats, and messages on a PostgreSQL server.

async with PostgresChatStore(database_url=DATABASE_URL) as chat_store:
    session = await chat_store.create_session(ttl_days=30)
Parameters:

  • database_url (str) – postgresql://user:password@host:port/database. Accepted as a plain postgresql:// URL; the asyncpg driver is selected internally.

  • schema (str) – The Postgres schema this store’s tables live in. Created if absent. Defaults to a dedicated schema so the store never collides with application tables.

  • pool_size (int) – Connections held by this store’s engine.

property engine: AsyncEngine#

The store’s underlying async engine.

The serving layer (fastrag.serve) uses this engine — not one it owns — for two Postgres-specific things that need a raw engine rather than a ChatStore method: the per-chat advisory lock that serializes generations, and the /healthz connection check. Requires the store to have been entered (__aenter__), else it raises fastrag.exceptions.ChatStoreError. Not part of the ChatStore contract — a non-Postgres store may not offer one (in which case those two serving features are unavailable).

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

See create_session().

Parameters:

  • ttl_days (int)

  • ip_address (str | None)

  • country (str | None)

Return type:

Session

async get_session(session_id)[source]#

See get_session().

Parameters:

session_id (str)

Return type:

Session | None

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

See touch_session().

Parameters:

  • session_id (str)

  • ttl_days (int)

  • min_bump_days (int)

Return type:

Session | None

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

See create_chat().

Parameters:

  • session_id (str)

  • ttl_days (int)

Return type:

Chat

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

See get_chat().

Parameters:

  • chat_id (str)

  • session_id (str)

Return type:

Chat | None

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

See list_chats().

Parameters:

  • session_id (str)

  • before (str | None)

  • limit (int)

Return type:

Page[Chat]

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

See delete_chat().

Parameters:

  • chat_id (str)

  • session_id (str)

Return type:

bool

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

See append_message().

Parameters:

  • chat_id (str)

  • 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)

Return type:

ChatMessage

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

See list_messages().

Parameters:

  • chat_id (str)

  • before (str | None)

  • limit (int)

Return type:

Page[ChatMessage]

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

See update_last_assistant_message().

Parameters:

  • chat_id (str)

  • content (str | None)

  • sources (list[dict[str, Any]] | None)

  • finish_reason (Literal['stop', 'length', 'error', 'interrupted'])

  • 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)

Return type:

ChatMessage | None