fastrag.ingestion.cache#

The pluggable step cache.

The CacheStore contract, its PostgreSQL and SQLite implementations, and the identity helpers every caching step builds cache ids with.

Caching is explicit and per-step (docs/specification/API.md: “Cache is explicit in each step … it is better to be explicit about it”): a step is handed a CacheStore or cache=False, never a hidden global.

class fastrag.ingestion.cache.PostgresCache(*, url, schema='fastrag_cache', pool_size=5, read_batch_size=128, read_batch_delay=0.001)[source]#

Bases: CacheStore

Cache pipeline sources and documents on a PostgreSQL server.

async with PostgresCache(url=DATABASE_URL) as cache:
    sources = SitemapCrawler(url=SITEMAP_URL, cache=cache)
Parameters:

  • 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 cache’s tables live in. Created if absent. Defaults to a dedicated schema so the cache never collides with application tables.

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

  • read_batch_size (int) – Maximum buckets per bulk read (default 128). Concurrent single reads are coalesced up to this limit.

  • read_batch_delay (float) – Seconds to collect concurrent reads (default 0.001). Zero yields once to the event loop; size 1 disables grouping.

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 bounded batches, one SQL statement per batch.

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 bounded batches, one SQL statement per batch.

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

class fastrag.ingestion.cache.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

fastrag.ingestion.cache.document_cache_id(step_name, document_id, config_hash)[source]#

Return the cache id for one document under one step configuration.

Parameters:

  • step_name (str) – The transforming step’s class name, e.g. "CharacterTextSplitter".

  • document_id (str) – The upstream Document’s id — cached independently per document to increase the cache hit rate.

  • config_hash (str) – The step’s config_hash.

Return type:

str

fastrag.ingestion.cache.source_cache_id(step_name, target, config_hash)[source]#

Return the cache id for one fetch target under one step configuration.

Parameters:

  • step_name (str) – The fetching step’s class name, e.g. "WebCrawler".

  • target (str) – What was fetched, e.g. a start URL — cached independently per target to increase the cache hit rate.

  • config_hash (str) – The step’s config_hash.

Return type:

str

Modules

base

The pluggable cache contract every step-side cache backend implements.

ids

Cache identity helpers, shared by every step that caches.

postgres

PostgreSQL-backed CacheStore.

schema

The ORM models behind PostgresCache.

sqlite

SQLite-backed CacheStore.