fastrag.ingestion.cache.base

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

Following docs/specification/caching.md: a step is handed a CacheStore or nothing at all, so caching stays explicit and per-step rather than a hidden global.

Classes

CacheStore()	Pluggable abstract cache backend with distinct lifecycle support.
SourceFreshnessValidator(*args, **kwargs)	Protocol for checking if an upstream resource has changed.

class fastrag.ingestion.cache.base.SourceFreshnessValidator(*args, **kwargs)

Bases: Protocol

Protocol for checking if an upstream resource has changed.

async is_source_modified(source_id, cached_etag, cached_last_modified, cached_content_hash)

Validates whether a source entry is stale against the upstream server.

Returns:

True if modified, False if cached version remains fresh.

Parameters:

• source_id (str)

• cached_etag (str | None)

• cached_last_modified (datetime | None)

• cached_content_hash (str)

Return type:

bool

class fastrag.ingestion.cache.base.CacheStore

Bases: AbstractAsyncContextManager[CacheStore], ABC

Pluggable abstract cache backend with distinct lifecycle support.

abstractmethod async get_source_entries(source_cache_id)

Fetches cached sources associated with a step run.

Parameters:

source_cache_id (str) – Unique hash of the source step configuration and target.

Returns:

The cached sources, [] if the id is known but holds no entries, or None on a cache miss. Each returned Source carries its persisted last_validated, so a caller can judge freshness without any per-process state.

Return type:

list[Source] | None

abstractmethod async put_source_entries(source_cache_id, sources)

Persists fetched sources and updates source reference tables.

Idempotent: writing the same source_cache_id again replaces its entry set rather than appending to it.

Parameters:

• source_cache_id (str) – Unique hash of the source step configuration.

• sources (Sequence[Source]) – Sequence of Source objects to persist. Each source’s last_validated is written verbatim when set; a source with last_validated=None is stamped with the current time, since being written at all means it was just validated.

Return type:

None

abstractmethod async get_document_entries(document_cache_id)

Fetches cached documents associated with a transformation step.

Parameters:

document_cache_id (str) – Hash composed of document ID and step configuration.

Returns:

The cached documents, [] if the id is known but holds no entries, or None on a cache miss.

Return type:

list[Document] | None

async get_source_entries_many(source_cache_ids)

Read multiple buckets, including misses and known-empty results.

Backends may override this with a bulk query. The default preserves compatibility with backends implementing only single-bucket reads. Duplicate ids are read once; an empty input performs no I/O.

Parameters:

source_cache_ids (Sequence[str])

Return type:

dict[str, list[Source] | None]

async get_document_entries_many(document_cache_ids)

Read document buckets with the same semantics as single reads.

The default uses single-bucket reads for existing custom backends. Duplicate ids are read once; an empty input performs no I/O.

Parameters:

document_cache_ids (Sequence[str])

Return type:

dict[str, list[Document] | None]

abstractmethod async put_document_entries(document_cache_id, documents)

Persists transformed documents and updates document cache references.

Idempotent: writing the same document_cache_id again replaces its entry set rather than appending to it.

Parameters:

• document_cache_id (str) – Hash composed of document ID and step configuration.

• documents (Sequence[Document]) – Transformed Document objects to persist.

Return type:

None

async get_source_headers_many(cache_ids)

Read source identities and validators, with lazy content when supported.

The fallback returns complete sources for existing custom backends.

Parameters:

cache_ids (Sequence[str])

Return type:

dict[str, list[Source] | None]

async get_bucket_sources(source_cache_id, source_ids)

Read only source_ids (with content) from one cached bucket.

The fallback loads the whole bucket; backends should override it with a targeted query so hydrating one page of a large crawl bucket does not transfer every sibling page.

Parameters:

• source_cache_id (str)

• source_ids (Sequence[str])

Return type:

list[Source]

async get_document_headers_many(cache_ids)

Read cached identities and routing metadata without text or vectors.

Optimized backends return lazy headers; the fallback returns complete documents. Ordinary cache getters always return complete outputs.

Parameters:

cache_ids (Sequence[str])

Return type:

dict[str, list[Document] | None]

abstractmethod async clear_sources()

Removes every cached source and every source cache id.

Cache administration only: emits nothing, retries nothing, and never touches a vector store.

Returns:

The number of source rows removed.

Return type:

int

abstractmethod async clear_documents()

Removes every cached document and every document cache id.

Cache administration only: emits nothing, retries nothing, and never touches a vector store.

Returns:

The number of document rows removed.

Return type:

int

abstractmethod async clear()

Removes everything this store holds: sources, documents, and both cache id tables.

Cache administration only: emits nothing, retries nothing, and never touches a vector store.

Returns:

The total number of source and document rows removed.

Return type:

int

abstractmethod async delete_sources(source_ids)

Removes the named cached sources.

Any source cache id left referencing none of its entries after this deletion is removed too, so a later get_source_entries on that id returns None (a miss) rather than [] (a known-empty result).

Idempotent and tolerant: a source_ids id that is not cached is silently skipped rather than raising. An empty source_ids is a no-op — no statement is sent.

Parameters:

source_ids (Sequence[str]) – The source ids to remove.

Returns:

The number of source rows actually removed.

Return type:

int

abstractmethod async delete_documents(source_ids)

Removes every cached document whose source_id is one of source_ids.

Any document cache id left referencing none of its entries after this deletion is removed too, so a later get_document_entries on that id returns None rather than [].

Idempotent and tolerant: a source_ids id with no cached documents is silently skipped rather than raising. An empty source_ids is a no-op — no statement is sent.

Parameters:

source_ids (Sequence[str]) – Remove every cached document belonging to one of these source ids.

Returns:

The number of document rows actually removed.

Return type:

int

abstractmethod async collect_garbage(max_age)

Removes every source and document whose last_accessed predates now - max_age.

An age-based sweep keyed on the column every read already touches. Any source or document cache id left referencing none of its entries after the sweep is removed too, with the same empty-bucket guarantee as delete_sources/ delete_documents.

Parameters:

max_age (timedelta) – How old an entry may be, by last_accessed, before it is collected. Must be positive.

Returns:

The total number of source and document rows removed.

Raises:

ValueError – If max_age is zero or negative.

Return type:

int