Backends

DatabaseBackend Protocol

chatbot_plugin_sdk.backends.base.DatabaseBackend

Bases: Protocol

Storage backend abstraction.

Two implementations are provided:

• :class:AsyncPgBackend — native asyncpg, binds to the creating event loop. Use with FastAPI or any long-running async application.

• :class:SyncPgBackend — psycopg2 wrapped in run_in_executor, not bound to any event loop. Use when callers use asyncio.run() from a ThreadPoolExecutor.

The processors always await backend methods, so the async signature is mandatory for both implementations.

setup async

setup(
    dense_dim: int | None, sparse_dim: int | None = None
) -> None

Idempotent. Creates schema + tables if missing; validates dimension if existing.

Parameters:

Name	Type	Description	Default
dense_dim	int | None	VECTOR(N) dimension. None → default VECTOR(768) placeholder.	required
sparse_dim	int | None	SPARSEVEC(N) vocabulary size (e.g. 30522 for BERT SPLADE). None → column stored as JSONB (sparse provider not configured).	None

Called by :class:IngestProcessor on first use.

validate async

validate(
    dense_dim: int | None, sparse_dim: int | None = None
) -> None

Read-only validation: tables must already exist. Raises if missing.

Parameters:

Name	Type	Description	Default
dense_dim	int | None	Expected VECTOR dimension from the dense provider.	required
sparse_dim	int | None	Expected SPARSEVEC dimension from the sparse provider.	None

Called by :class:RetrieveProcessor on first use.

upsert async

upsert(
    article_id: UUID,
    metadata: dict,
    chunks: list[str],
    dense_vectors: list[list[float]] | None,
    sparse_vectors: list[dict[str, float]] | None,
    articles_column_values: dict[str, Any] | None = None,
) -> None

Insert-or-replace article + chunks inside a single transaction.

url must be present in articles_column_values — it is used by :class:IngestProcessor to derive a deterministic article_id via uuid.uuid5(NAMESPACE_URL, url).

Parameters:

Name	Type	Description	Default
metadata	dict	Opaque JSONB blob — the SDK never interprets its keys.	required
articles_column_values	dict[str, Any] | None	SQL column values for the articles table (url, title, source, public_article_id, topic_id). Keys must be in the known article column set. url is required.	None

On success: commits. On any error: rolls back, raises :exc:DatabaseError.

search_dense async

search_dense(
    query_vec: list[float],
    top_k: int,
    filters: dict[str, Any] | None = None,
) -> list[SearchRow]

Cosine similarity search on the dense_vector column.

Parameters:

Name	Type	Description	Default
filters	dict[str, Any] | None	Column-level filters on the articles table, e.g. {"topic_id": "uuid"}. Keys must be known article columns.	None

search_sparse async

search_sparse(
    query_vec: dict[str, float],
    top_k: int,
    filters: dict[str, Any] | None = None,
) -> list[SearchRow]

Maximum inner product search on the sparse_vector column (<#> operator).

distance in returned rows is the negative inner product (lower = more similar), matching the same convention as search_dense so :func:_rrf_merge can treat both lists uniformly by rank.

close async

close() -> None

Dispose connection pool. Call on application shutdown.

SearchRow

chatbot_plugin_sdk.backends.base.SearchRow dataclass

SearchRow(
    chunk_id: str,
    article_id: str,
    chunk_index: int,
    content: str,
    distance: float,
    article_metadata: dict[str, Any],
)

Single search result row returned by DatabaseBackend.search_dense().

AsyncPgBackend

chatbot_plugin_sdk.backends.async_pg.AsyncPgBackend

AsyncPgBackend(config: DatabaseConfig)

Native asyncpg backend. Not safe to share across multiple asyncio.run() calls in different threads — use :class:SyncPgBackend for ThreadPoolExecutor scenarios.

SyncPgBackend

chatbot_plugin_sdk.backends.sync_pg.SyncPgBackend

SyncPgBackend(config: DatabaseConfig)

Thread-safe, event-loop-independent database backend using psycopg2.

All sync SQLAlchemy operations run in asyncio.get_running_loop().run_in_executor so they don't block the calling coroutine's event loop, yet the engine itself carries no asyncio state and can be shared across threads.

Thread safety

• The SQLAlchemy connection pool uses internal locks; sharing _engine across threads is safe.
• Each DB operation opens its own session + connection from the pool.
• Sessions are never shared between calls.

Multi-process note

• Do NOT share this backend across fork() boundaries. Create a new instance in each child process.