Skip to main content

protocol

Generic cache protocol for the pipeline result cache.

CacheProtocol defines the interface that any cache backend must implement. Stores type-hint their cache parameter against this protocol so that:

  • The concrete SqliteCache satisfies it in production.
  • Tests can supply lightweight mocks or in-memory implementations.
  • Future backends (Postgres, DuckDB) can be added without changing store code.

CacheAccessor defines the read-only interface used by interactive pipeline steps to consume data written by background steps. Concrete implementations (e.g. SqliteCacheAccessor) are constructed by the DAG runner when resolving <step_name>.cache references from the YAML; step code only ever sees the protocol.

Classes​

CacheAccessor​

class CacheAccessor(*args, **kwargs):

Read-only accessor for a single logical cache partition.

Instances are constructed by the DAG runner when resolving <background_step_name>.cache input references declared in the task-template YAML. Interactive step code receives a CacheAccessor and is free to consume it lazily (iter_rows / iter_dataframes) or eagerly (to_dataframe), without knowing which table or DB technology backs it.

All read methods accept an optional filters parameter — a list of SQLAlchemy ColumnElement expressions that are AND-ed with the partition's base filters at query time. This lets a step further narrow results without requiring a new accessor:

from sqlalchemy import func
rows = accessor.iter_rows(
    filters=[ModelInference.processed_at >= "2025-01-01"]
)

Tags can be queried using SQLAlchemy's func.json_extract:

rows = accessor.iter_rows(
    filters=[
        func.json_extract(ModelInference.tags, "$.laterality") == "LEFT"
    ]
)

Ancestors​

Variables​

  • table_name : str - The underlying table name (for logging and debugging only).

Methods​

count​

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

Return the number of rows in this partition.

Arguments

  • filters: Optional extra SQLAlchemy column expressions to apply on top of the accessor's base filters (AND-ed together).

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.

Useful for batch processing where you want DataFrame semantics but cannot afford to hold the full dataset in memory.

Arguments

  • chunk_size: Maximum number of rows per yielded DataFrame.
  • columns: If provided, only these columns are fetched. The projection is issued as SQL, so an unrequested blob column is never read off disk. Names the model does not map are ignored.
  • filters: Optional extra SQLAlchemy column expressions (AND-ed with the accessor's base filters).

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 underlying database session is held open for the duration of iteration and closed automatically when the iterator is exhausted or garbage-collected.

Arguments

  • chunk_size: Number of rows fetched from the DB per round-trip (passed to SQLAlchemy's yield_per).
  • filters: Optional extra SQLAlchemy column expressions (AND-ed with the accessor's 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 entire partition as a DataFrame.

For large datasets prefer iter_dataframes to avoid loading all rows into memory at once.

Arguments

  • columns: If provided, only these columns are fetched. The projection is issued as SQL, so an unrequested blob column is never read off disk. Names the model does not map are ignored.
  • filters: Optional extra SQLAlchemy column expressions (AND-ed with the accessor's base filters).

Returns A pandas.DataFrame with one row per cache record. Returns an empty DataFrame when the partition contains no rows — carrying the requested columns when those were given, so a caller can index them without first checking whether the partition was empty.

CacheProtocol​

class CacheProtocol(*args, **kwargs):

Interface for SQLAlchemy-backed cache implementations.

Implementations must provide:

  • A session() context manager yielding a SQLAlchemy Session that auto-commits on success and rolls back on exception.
  • An engine property for advanced use (e.g. pandas read_sql).
  • A close() method to release resources.

Ancestors​

Variables​

  • engine : sqlalchemy.engine.base.Engine - The underlying SQLAlchemy engine.

Methods​

close​

def close(self) ‑> None:

Dispose of connections and release resources.

ensure_types​

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

Create/migrate the physical tables for the given types.

For each registered type, the backend ensures its physical table exists and is migrated up to max(pinned_version, tracked_version) via each version's inbound upgrade(op) (an Alembic op-API migration autogenerated from the ORM), recording the resulting version in the type_versions tracker. Types absent from pinned are created at (at least) their current tracked version — never downgraded.

Arguments

  • pinned: Mapping of type name -> minimum required version (e.g. the versions a task's steps bind to). A type may be migrated higher than pinned if its table is already there.

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.

Arguments

  • orm_model: The SQLAlchemy ORM model class whose table holds the data (e.g. ModelInference). Not pinned to a single Base: each versioned record declares its own local declarative base, so there is no shared global base to type against.
  • base_filters: SQLAlchemy column expressions that scope the accessor to a specific partition (e.g. [ModelInference.task_hash == "abc", ModelInference.model_ref == "FoveaModel"]).

Returns A CacheAccessor instance for reading data from this partition.

session​

def session(self) ‑> contextlib.AbstractContextManager[sqlalchemy.orm.session.Session]:

Provide a transactional session scope.