fastrag.ingestion.cache.postgres#

PostgreSQL-backed CacheStore.

The schema is self-managed: __aenter__ creates the target Postgres schema and its tables if they are absent (docs/specification/API.md: “Databases schemas are created automatically if they don’t exist … The user can customize the target database schema”). The cache owns its own schema and Alembic history is untouched by it.

Statements are written against the ORM models in fastrag.ingestion.cache.schema; the engine’s schema_translate_map places them in the configured schema.

Classes

PostgresCache(*, url[, schema, pool_size, ...])

Cache pipeline sources and documents on a PostgreSQL server.
class fastrag.ingestion.cache.postgres.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