fastrag.ingestion.cache.sqlite#

SQLite-backed CacheStore.

One file and no server, for a caller who does not want to run Postgres. Built on the standard-library sqlite3 module, so it adds no dependency.

The layout mirrors fastrag.ingestion.cache.schema: a bucket per cache id, a snapshot per item of a bucket’s output, and a record per logical id carrying the garbage-collection clock. __aenter__ creates the tables if they are absent.

sqlite3 is blocking, so every operation runs as one transaction on a dedicated worker thread that owns the connection. Operations from one instance are therefore serialized; separate processes may share the file, since the database runs in WAL mode and waits out a writer’s lock.

Classes

SQLiteCache(*, path[, timeout])

Cache pipeline sources and documents in a local SQLite file.
class fastrag.ingestion.cache.sqlite.SQLiteCache(*, path, timeout=30.0)[source]#

Bases: CacheStore

Cache pipeline sources and documents in a local SQLite file.

async with SQLiteCache(path=".fastrag/cache.sqlite") as cache:
    sources = SitemapCrawler(url=SITEMAP_URL, cache=cache)
Parameters:

  • path (str | os.PathLike[str]) – The database file. Created, along with missing parent directories, on __aenter__. ":memory:" keeps the cache in memory for the lifetime of the async with block.

  • timeout (float) – Seconds to wait for another process’s write lock before an operation fails with TransientCacheError.

async get_source_entries(source_cache_id)[source]#

See get_source_entries().

Parameters:

source_cache_id (str)

Return type:

list[Source] | None

async get_source_entries_many(source_cache_ids)[source]#

Fetch source buckets in one transaction.

Parameters:

source_cache_ids (Sequence[str])

Return type:

dict[str, list[Source] | None]

async get_source_headers_many(cache_ids)[source]#

Retrieve only identities, validators and routing metadata for indexing.

Parameters:

cache_ids (Sequence[str])

Return type:

dict[str, list[Source] | None]

async get_bucket_sources(source_cache_id, source_ids)[source]#

See get_bucket_sources().

Parameters:
Return type:

list[Source]

async put_source_entries(source_cache_id, sources)[source]#

See put_source_entries().

Parameters:
Return type:

None

async get_document_entries(document_cache_id)[source]#

See get_document_entries().

Parameters:

document_cache_id (str)

Return type:

list[Document] | None

async get_document_entries_many(document_cache_ids)[source]#

Fetch document buckets in one transaction.

Parameters:

document_cache_ids (Sequence[str])

Return type:

dict[str, list[Document] | None]

async get_document_headers_many(cache_ids)[source]#

Retrieve snapshot identifiers and routing metadata without large payloads.

Parameters:

cache_ids (Sequence[str])

Return type:

dict[str, list[Document] | None]

async put_document_entries(document_cache_id, documents)[source]#

See put_document_entries().

Parameters:
Return type:

None

async clear_sources()[source]#

See clear_sources().

Return type:

int

async clear_documents()[source]#

See clear_documents().

Return type:

int

async clear()[source]#

See clear().

Return type:

int

async delete_sources(source_ids)[source]#

See delete_sources().

Parameters:

source_ids (Sequence[str])

Return type:

int

async delete_documents(source_ids)[source]#

See delete_documents().

Parameters:

source_ids (Sequence[str])

Return type:

int

async collect_garbage(max_age)[source]#

See collect_garbage().

Parameters:

max_age (timedelta)

Return type:

int