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, releasing its file descriptors.

Not an invalidation: the engine is shared per database file (see _engines) and stays usable, so disposing here frees this file's pooled connections without breaking any other holder — SQLAlchemy builds a fresh pool on the next use.

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 read session is held open for the duration of iteration and closed automatically when the iterator is exhausted. The touch-on-read write is not part of it — see below.

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.