Skip to main content

sqlite

SQLite implementation of the cache using SQLAlchemy.

Thread-safe via SQLAlchemy's connection pooling (StaticPool for in-memory databases). Each call to session() creates an independent sessionmaker-produced session, so concurrent Prefect task threads can share one SqliteCache instance without session contention.

Typical usage:

cache = SqliteCache("/data/pods/site-a/pipeline_cache.db")
with cache.session() as session:
    session.add(EHRData(task_hash="abc", file_id="f1", ...))

Classes​

SqliteCache​

class SqliteCache(db_path: str | Path):

SQLAlchemy-backed SQLite cache.

Arguments

  • db_path: Path to the SQLite database file. Use ":memory:" for in-process testing.

Variables​

  • engine : sqlalchemy.engine.base.Engine - Expose the underlying engine for advanced queries (e.g. pandas).

Methods​

close​

def close(self) ‑> None:

Dispose of the connection pool.

ensure_types​

def ensure_types(self, pinned: Mapping[str, int]) ‑> None:

Create/migrate every registered type's table to the needed version.

Serialised on the per-path _init_lock so a call made after construction (e.g. a reader opening the pod cache and pinning directly-written records, or DAG setup pinning step versions) cannot race a concurrent open's DDL and fail one caller with a duplicate-column or batch-alter error. __init__ calls _ensure_types directly while already holding this lock; every other caller comes through here.

See _ensure_types for the migration semantics.

make_accessor​

def make_accessor(    self, orm_model: Any, base_filters: list[ColumnElement[Any]],) ‑> CacheAccessor:

Create a read-only accessor for a specific cache partition.

orm_model is any SQLAlchemy mapped class. It is intentionally not pinned to a single Base: each versioned record declares its own local declarative base (its own MetaData), so there is no shared global base the ORM could be typed against.

session​

def session(    self,) ‑> collections.abc.Generator[sqlalchemy.orm.session.Session, None, None]:

Provide a transactional session scope.

SqliteCacheAccessor​

class SqliteCacheAccessor(    cache: CacheProtocol, orm_model: Any, base_filters: list[ColumnElement[Any]],):

Read-only accessor for a single logical partition of a SQLite cache table.

Constructed by the DAG runner when resolving <step_name>.cache references declared in the task-template YAML. Step code receives this object typed as CacheAccessor (the protocol) and calls its read methods without knowing the underlying table or filter details.

Arguments

  • cache: The SqliteCache (or any CacheProtocol) backing the data.
  • orm_model: The SQLAlchemy ORM class whose table holds the data (e.g. ModelInference).
  • base_filters: SQLAlchemy column expressions that scope this accessor to a specific partition of the table (e.g. [ModelInference.task_hash == "abc", ModelInference.model_ref == "FoveaModel"]). These are AND-ed with any ad-hoc filters supplied at call time.

Variables​

  • table_name : str - The underlying SQLAlchemy table name.

Methods​

count​

def count(self, filters: list[ColumnElement[Any]] | None = None) ‑> int:

Return the number of rows in this partition.

Arguments

  • filters: Optional extra column expressions AND-ed with the base filters.

Returns Integer row count.

iter_dataframes​

def iter_dataframes(    self,    chunk_size: int = 1000,    columns: list[str] | None = None,    filters: list[ColumnElement[Any]] | None = None,) ‑> collections.abc.Iterator[pandas.core.frame.DataFrame]:

Stream the partition as a sequence of chunked DataFrames.

Arguments

  • chunk_size: Maximum rows per yielded DataFrame.
  • columns: If provided, only these columns are fetched, as SQL.
  • filters: Optional extra column expressions AND-ed with the base filters.

Raises

  • ValueError: If chunk_size is not positive. The degenerate-columns path below counts rows down by chunk_size, so a non-positive value would yield empty frames for ever instead of failing.

iter_rows​

def iter_rows(    self, chunk_size: int = 1000, filters: list[ColumnElement[Any]] | None = None,) ‑> collections.abc.Iterator[dict[str, typing.Any]]:

Stream rows as plain dicts without loading all into memory.

The database session is held open for the duration of iteration and closed automatically when the iterator is exhausted.

Arguments

  • chunk_size: Rows fetched per DB round-trip (yield_per).
  • filters: Optional extra column expressions AND-ed with the base filters.

to_dataframe​

def to_dataframe(    self,    columns: list[str] | None = None,    filters: list[ColumnElement[Any]] | None = None,) ‑> pandas.core.frame.DataFrame:

Materialise the partition as a DataFrame.

Arguments

  • columns: If provided, only these columns are fetched — the projection is issued as SQL, not applied afterwards, so unrequested blob columns are never read. Names the ORM does not map are ignored.
  • filters: Optional extra column expressions AND-ed with the base filters.

Returns A pandas.DataFrame — empty when the partition has no rows, and in that case still carrying the requested columns so callers can index them unconditionally.