fastrag.ingestion.cache.schema#

The ORM models behind PostgresCache.

This schema is owned by the cache itself, not by a repository-wide Alembic setup: PostgresCache creates its schema and tables on __aenter__ if they are absent (see docs/specification/API.md, “Databases schemas are created automatically”). It therefore declares its own CacheBase rather than attaching to a shared declarative base — doing the latter would make alembic revision --autogenerate try to manage these tables too, which is exactly what must not happen.

The models carry no Postgres schema. Each PostgresCache engine maps them onto its own target schema through schema_translate_map, so the same models serve as many Postgres schemas as the caller wants.

Sources and documents share one three-table layout:

A bucket (SourceBucket, DocumentBucket) is one cache id: the output of one step configuration. A bucket with no snapshots is a known-empty result, distinct from a miss.
A snapshot (SourceSnapshot, DocumentSnapshot) is one item of a bucket’s output, with its full payload.
A record (SourceRecord, DocumentRecord) is the logical identity shared by every snapshot of the same id. Its last_accessed is the garbage-collection clock for sources.

Module Attributes

DEFAULT_SCHEMA

Schema name used when a PostgresCache caller does not pick one.

Classes

`CacheBase`(**kwargs)	Declarative base owning the cache tables, and only them.
`DocumentBucket`(**kwargs)	One transformation step configuration's cache id.
`DocumentRecord`(**kwargs)	A document's logical identity, shared by every stage that emits it.
`DocumentSnapshot`(**kwargs)	A document as one stage produced it.
`SourceBucket`(**kwargs)	One source step configuration's cache id.
`SourceRecord`(**kwargs)	A source's logical identity and garbage-collection clock.
`SourceSnapshot`(**kwargs)	A source as one configuration rendered it: bytes and validators.

fastrag.ingestion.cache.schema.DEFAULT_SCHEMA = 'fastrag_cache'#: Schema name used when a PostgresCache caller does not pick one.

class fastrag.ingestion.cache.schema.CacheBase(**kwargs)[source]#

Bases: DeclarativeBase

Declarative base owning the cache tables, and only them.

Parameters:: kwargs (Any)

type_annotation_map = {<class 'bytes'>: <class 'sqlalchemy.sql.sqltypes.LargeBinary'>, <class 'datetime.datetime'>: DateTime(timezone=True), <class 'str'>: <class 'sqlalchemy.sql.sqltypes.Text'>, dict[str, typing.Any]: <class 'sqlalchemy.dialects.postgresql.json.JSONB'>, list[float]: ARRAY(Float())}#

class fastrag.ingestion.cache.schema.SourceBucket(**kwargs)[source]#

Bases: CacheBase

One source step configuration’s cache id.

id: Mapped[str]#

type_annotation_map = {<class 'bytes'>: <class 'sqlalchemy.sql.sqltypes.LargeBinary'>, <class 'datetime.datetime'>: DateTime(timezone=True), <class 'str'>: <class 'sqlalchemy.sql.sqltypes.Text'>, dict[str, typing.Any]: <class 'sqlalchemy.dialects.postgresql.json.JSONB'>, list[float]: ARRAY(Float())}#

class fastrag.ingestion.cache.schema.SourceRecord(**kwargs)[source]#

Bases: CacheBase

A source’s logical identity and garbage-collection clock.

The payload columns mirror the latest snapshot written for this id; reads serve payloads from SourceSnapshot, never from here.

id: Mapped[str]#

uri: Mapped[str]#

content: Mapped[bytes]#

content_encoding: Mapped[str | None]#

content_type: Mapped[str]#

content_hash: Mapped[str]#

etag: Mapped[str | None]#

last_modified: Mapped[datetime | None]#

metadata_: Mapped[dict[str, Any]]#

last_validated: Mapped[datetime]#

last_accessed: Mapped[datetime]#

type_annotation_map = {<class 'bytes'>: <class 'sqlalchemy.sql.sqltypes.LargeBinary'>, <class 'datetime.datetime'>: DateTime(timezone=True), <class 'str'>: <class 'sqlalchemy.sql.sqltypes.Text'>, dict[str, typing.Any]: <class 'sqlalchemy.dialects.postgresql.json.JSONB'>, list[float]: ARRAY(Float())}#

class fastrag.ingestion.cache.schema.SourceSnapshot(**kwargs)[source]#

Bases: CacheBase

A source as one configuration rendered it: bytes and validators.

bucket_id: Mapped[str]#

source_id: Mapped[str]#

uri: Mapped[str]#

content: Mapped[bytes]#

content_encoding: Mapped[str | None]#

content_type: Mapped[str]#

content_hash: Mapped[str]#

etag: Mapped[str | None]#

last_modified: Mapped[datetime | None]#

metadata_: Mapped[dict[str, Any]]#

last_validated: Mapped[datetime]#

type_annotation_map = {<class 'bytes'>: <class 'sqlalchemy.sql.sqltypes.LargeBinary'>, <class 'datetime.datetime'>: DateTime(timezone=True), <class 'str'>: <class 'sqlalchemy.sql.sqltypes.Text'>, dict[str, typing.Any]: <class 'sqlalchemy.dialects.postgresql.json.JSONB'>, list[float]: ARRAY(Float())}#

class fastrag.ingestion.cache.schema.DocumentBucket(**kwargs)[source]#

Bases: CacheBase

One transformation step configuration’s cache id.

id: Mapped[str]#

type_annotation_map = {<class 'bytes'>: <class 'sqlalchemy.sql.sqltypes.LargeBinary'>, <class 'datetime.datetime'>: DateTime(timezone=True), <class 'str'>: <class 'sqlalchemy.sql.sqltypes.Text'>, dict[str, typing.Any]: <class 'sqlalchemy.dialects.postgresql.json.JSONB'>, list[float]: ARRAY(Float())}#

class fastrag.ingestion.cache.schema.DocumentRecord(**kwargs)[source]#

Bases: CacheBase

A document’s logical identity, shared by every stage that emits it.

The payload columns are legacy: they stay blank, because one logical id has a different payload per stage (an embedder keeps its input id but adds a vector). Payloads live on DocumentSnapshot.

id: Mapped[str]#

source_id: Mapped[str]#

step_config_hash: Mapped[str]#

content: Mapped[str]#

embedding: Mapped[list[float] | None]#

metadata_: Mapped[dict[str, Any]]#

last_accessed: Mapped[datetime]#

type_annotation_map = {<class 'bytes'>: <class 'sqlalchemy.sql.sqltypes.LargeBinary'>, <class 'datetime.datetime'>: DateTime(timezone=True), <class 'str'>: <class 'sqlalchemy.sql.sqltypes.Text'>, dict[str, typing.Any]: <class 'sqlalchemy.dialects.postgresql.json.JSONB'>, list[float]: ARRAY(Float())}#

class fastrag.ingestion.cache.schema.DocumentSnapshot(**kwargs)[source]#

Bases: CacheBase

A document as one stage produced it.

payload holds the compressed text and metadata; the vector and the header fields live in their own columns so index-only reads skip it.

bucket_id: Mapped[str]#

document_id: Mapped[str]#

payload: Mapped[bytes]#

header_metadata: Mapped[dict[str, Any]]#

step_config_hash: Mapped[str]#

embedding: Mapped[list[float] | None]#

source_id: Mapped[str]#

last_accessed: Mapped[datetime]#

type_annotation_map = {<class 'bytes'>: <class 'sqlalchemy.sql.sqltypes.LargeBinary'>, <class 'datetime.datetime'>: DateTime(timezone=True), <class 'str'>: <class 'sqlalchemy.sql.sqltypes.Text'>, dict[str, typing.Any]: <class 'sqlalchemy.dialects.postgresql.json.JSONB'>, list[float]: ARRAY(Float())}#