Persistence#

The data store: root-level registry plus per-collection catalog, product tables, and immutable object store. Created only by adapt init; consumers read through StoreClient.

Store lifecycle: initialization and opening of the on-disk Adapt data store.

The layout is created only by init_store() (surfaced as adapt init); the pipeline refuses to run against an uninitialized root:

{root}/
├── registry.db      # collections + run lifecycle (store_registry.sql)
├── logs/
└── collections/     # provisioned per collection at run start
exception adapt.persistence.store.AlreadyInitializedError#

Bases: StoreError

The root already contains an initialized store.

class adapt.persistence.store.Collection(collection_id, path, catalog)#

Bases: object

One collection’s on-disk domain: catalog.db + products.db + decisions.db + objects/.

decisions.db is the analysis-only decision log (segmentation and tracking decisions); the pipeline never reads it.

Parameters:

  • collection_id (str)

  • path (Path)

  • catalog (Catalog)

close()#
Return type:

None

class adapt.persistence.store.Store(root)#

Bases: object

An initialized on-disk Adapt data store. Obtain via open().

Parameters:

root (Path)

close()#

Close every collection opened through this store.

Return type:

None

collection(collection_id)#

Provision (idempotently), validate, and return the collection’s domain.

Return type:

Collection

Parameters:

collection_id (str)

classmethod open(root)#

Open an initialized store, failing loudly on legacy or bare roots.

Return type:

Store

Parameters:

root (str | Path)

validate_collection(collection_id)#

Raise when catalog.db and products.db schema snapshots disagree.

Return type:

None

Parameters:

collection_id (str)

exception adapt.persistence.store.StoreError#

Bases: Exception

A store-level operation failed.

adapt.persistence.store.init_store(root)#

Create the store layout at root and return the resolved root.

Creates exactly registry.db, logs/, and collections/. Raises AlreadyInitializedError if a store exists, and StoreError on a root using the obsolete pre-store layout.

Return type:

Path

Parameters:

root (str | Path)

Store-level registry: collections and the single run lifecycle.

Backs {root}/registry.db (schema store_registry.sql). This is the ONE run lifecycle in the store — module executions and run events are recorded against it, and per-collection discovery lives in the collection catalog.

Thread-safe singleton per store root (WAL); close_all() releases the process-wide cache deterministically (FD-leak lesson from the old registry).

class adapt.persistence.store_registry.RunStart(run_id, collection_id, config_hash, config_json, pipeline_version, environment_json)#

Bases: object

Everything the registry records when a run begins.

Parameters:

  • run_id (str)

  • collection_id (str)

  • config_hash (str)

  • config_json (str)

  • pipeline_version (str)

  • environment_json (str)

collection_id: str#
config_hash: str#
config_json: str#
environment_json: str#
pipeline_version: str#
run_id: str#
class adapt.persistence.store_registry.StoreRegistry(root)#

Bases: SqliteStore

Registry over {root}/registry.db; requires an initialized store.

Parameters:

root (str | Path)

begin_run(start)#

Record a new run as running; unknown collection or reused run_id raise.

Return type:

None

Parameters:

start (RunStart)

classmethod close_all()#

Close every cached registry connection and empty the cache.

Return type:

None

ensure_collection_location(collection_id, *, lat, lon)#

Record the collection location once; an existing location is kept.

Deterministic: called with the location from pipeline inputs (the first gridded dataset), never from external lookup tables.

Return type:

None

Parameters:

  • collection_id (str)

  • lat (float)

  • lon (float)

finalize_run(run_id, status, *, scans_processed=None, scans_failed=None)#

Move a running run to a terminal status; double-finalize raises.

Return type:

None

Parameters:

  • run_id (str)

  • status (str)

  • scans_processed (int | None)

  • scans_failed (int | None)

get_collection(collection_id)#
Return type:

dict

Parameters:

collection_id (str)

classmethod get_instance(root)#

Return the cached registry for root, creating it on first use.

Return type:

StoreRegistry

Parameters:

root (str | Path)

get_run(run_id)#
Return type:

dict

Parameters:

run_id (str)

latest_run(collection_id=None)#

The most recently started run (optionally within one collection).

Return type:

dict | None

Parameters:

collection_id (str | None)

list_collections()#
Return type:

list[dict]

record_module_execution(run_id, module, *, status, duration_seconds, detail_json)#

Upsert one module’s execution record (re-runs replace their own row).

Return type:

None

Parameters:

  • run_id (str)

  • module (str)

  • status (str)

  • duration_seconds (float)

  • detail_json (str)

record_run_event(run_id, *, severity, module, message, context_json)#

Append one warning/error event for a run.

Return type:

None

Parameters:

  • run_id (str)

  • severity (str)

  • module (str)

  • message (str)

  • context_json (str)

register_collection(collection_id, *, source_kind, lat=None, lon=None, metadata_json=None)#

Register a collection; registering an existing id is a no-op.

Return type:

None

Parameters:

  • collection_id (str)

  • source_kind (str)

  • lat (float | None)

  • lon (float | None)

  • metadata_json (str | None)

resume_run(run_id)#

Re-open an existing run (--run-id continuation): running, no end time.

Return type:

None

Parameters:

run_id (str)

run_events(run_id)#
Return type:

list[dict]

Parameters:

run_id (str)

run_modules(run_id)#
Return type:

list[dict]

Parameters:

run_id (str)

Discovery + provenance catalog for one collection.

Backs collections/{id}/catalog.db (schema collection_catalog.sql). Science rows live in products.db; this catalog answers which artifacts and scans exist, which products each scan has, and where every object came from.

class adapt.persistence.collection_catalog.ArtifactRecord(artifact_id, artifact_type, producer, run_id, scan_id, observation_time, original_filename, source_uri, object_name, checksum_sha256, size_bytes)#

Bases: object

One complete artifacts row; built by the ObjectStore at commit.

Parameters:

  • artifact_id (str)

  • artifact_type (str)

  • producer (str)

  • run_id (str)

  • scan_id (str | None)

  • observation_time (str | None)

  • original_filename (str | None)

  • source_uri (str | None)

  • object_name (str)

  • checksum_sha256 (str)

  • size_bytes (int)

artifact_id: str#
artifact_type: str#
checksum_sha256: str#
object_name: str#
observation_time: str | None#
original_filename: str | None#
producer: str#
run_id: str#
scan_id: str | None#
size_bytes: int#
source_uri: str | None#
class adapt.persistence.collection_catalog.Catalog(collection_dir)#

Bases: SqliteStore

Catalog over {collection_dir}/catalog.db.

Parameters:

collection_dir (str | Path)

add_lineage(child_artifact_id, parent_artifact_id, relationship='derived_from')#

Record one provenance edge; both artifacts must exist, edges are unique.

Return type:

None

Parameters:

  • child_artifact_id (str)

  • parent_artifact_id (str)

  • relationship (str)

children_of(artifact_id)#
Return type:

list[str]

Parameters:

artifact_id (str)

complete_scans(run_id)#

Complete scans of one run, time-ordered.

Return type:

list[dict]

Parameters:

run_id (str)

find_artifact(run_id, scan_id, artifact_type)#

Newest artifact of one type for a scan (re-runs replace, newest wins).

Return type:

dict | None

Parameters:

  • run_id (str)

  • scan_id (str)

  • artifact_type (str)

find_artifact_by_scan(scan_id, artifact_type)#

Newest artifact of one type for a scan across ALL runs.

Raw volumes are content-identified (same bytes → same scan_id), so a later run reuses the object an earlier run committed.

Return type:

dict | None

Parameters:

  • scan_id (str)

  • artifact_type (str)

find_by_source_uri(source_uri)#

Newest artifact acquired from one source URI (skip-if-downloaded lookup).

Return type:

dict | None

Parameters:

source_uri (str)

get_artifact(artifact_id)#
Return type:

dict | None

Parameters:

artifact_id (str)

get_scan(run_id, scan_id)#
Return type:

dict | None

Parameters:

  • run_id (str)

  • scan_id (str)

Link one product to a scan; re-linking replaces (module re-run semantics).

Exactly one of artifact_id/table_name identifies where the product lives. Completeness is recomputed after every link.

Return type:

None

Parameters:

  • run_id (str)

  • scan_id (str)

  • product (str)

  • artifact_id (str | None)

  • table_name (str | None)

list_artifacts(run_id=None, artifact_type=None)#

Artifacts filtered by run and/or type, ordered by observation time.

Return type:

list[dict]

Parameters:

  • run_id (str | None)

  • artifact_type (str | None)

mark_scan_failed(run_id, scan_id)#

Record a scan-processing failure; a later successful re-run re-completes it.

Return type:

None

Parameters:

  • run_id (str)

  • scan_id (str)

parents_of(artifact_id)#
Return type:

list[str]

Parameters:

artifact_id (str)

register_artifact(record)#

Insert one artifacts row; duplicate artifact_id or object_name raise.

Return type:

None

Parameters:

record (ArtifactRecord)

register_scan(record)#

Upsert one scan keyed by (run_id, scan_id); starts pending.

scan_time is stored in the canonical format (to_scan_iso); coverage times are optional per-source metadata kept at full precision. Two different scans sharing one nominal second within a run raise.

Return type:

None

Parameters:

record (ScanRecord)

register_table_schema(snapshot)#

Mirror one frozen table-schema snapshot (written by the SchemaLedger).

Return type:

None

Parameters:

snapshot (dict)

scan_is_complete(run_id, scan_id)#
Return type:

bool

Parameters:

  • run_id (str)

  • scan_id (str)

scan_products(run_id, scan_id)#

All product links for one scan, keyed by product name.

Return type:

dict[str, dict]

Parameters:

  • run_id (str)

  • scan_id (str)

Immutable object store for one collection’s objects/ directory.

Write protocol: ObjectStore.begin() allocates the uuid4 artifact_id BEFORE any bytes exist and hands back a staging path inside store-owned objects/.staging/. Nothing is visible or cataloged until commit(), which checksums the staged bytes, atomically publishes them via os.replace, and registers the full catalog row. abort() removes the staging file and leaves no trace.

class adapt.persistence.objects.ArtifactMeta(artifact_type, producer, run_id, scan_id, observation_time, original_filename=None, source_uri=None)#

Bases: object

Caller-supplied metadata for one object; the store computes the rest.

Parameters:

  • artifact_type (str)

  • producer (str)

  • run_id (str)

  • scan_id (str | None)

  • observation_time (datetime | None)

  • original_filename (str | None)

  • source_uri (str | None)

artifact_type: str#
observation_time: datetime | None#
original_filename: str | None = None#
producer: str#
run_id: str#
scan_id: str | None#
source_uri: str | None = None#
class adapt.persistence.objects.ObjectStore(objects_dir, catalog)#

Bases: object

Atomic, catalog-registered writes into one collection’s objects/.

Parameters:

  • objects_dir (str | Path)

  • catalog (Catalog)

abort(handle)#

Discard the staged bytes; nothing was published or cataloged.

Return type:

None

Parameters:

handle (ObjectWriteHandle)

begin(*, suffix)#

Allocate an artifact_id and staging path; write bytes there, then commit.

Return type:

ObjectWriteHandle

Parameters:

suffix (str)

commit(handle, meta, parents=())#

Publish the staged bytes atomically, register the row and lineage edges.

Return type:

ArtifactRecord

Parameters:
class adapt.persistence.objects.ObjectWriteHandle(artifact_id, suffix, staging_path)#

Bases: object

One in-flight object write: id allocated, bytes staged, not yet visible.

Parameters:

  • artifact_id (str)

  • suffix (str)

  • staging_path (Path)

artifact_id: str#
staging_path: Path#
suffix: str#

Module-owned product tables with frozen first-frame schemas.

SchemaLedger freezes a table’s schema on its first non-empty frame — declared primary key + payload columns — and writes identical snapshots to products.db and catalog.db. Frozen types are deterministic: core identity columns are contracted by name (_CORE_COLUMN_TYPES), every other numeric column freezes as REAL, so the schema depends on the column set — never on the values in whichever scan arrived first. TableWriter stamps scan identity from PersistenceMeta, enforces the frozen schema (no new columns, no incompatible dtypes, no duplicate keys, no cross-module writes), and upserts on the declared key. There is no ALTER TABLE on the new path — ever.

Granularity derives from the declared primary key: valid_time in the key marks a time-granular product (interpolations, forecasts; the row’s own time is part of its identity), otherwise scan_id marks scan granularity, otherwise the table is run-granular. Identity columns of scan-granular per-scan writes are stamped by the writer — modules never pass them.

class adapt.persistence.products.SchemaLedger(products_path, catalog)#

Bases: object

Frozen first-frame schema registry with dual snapshots.

Parameters:

  • products_path (str | Path)

  • catalog (Catalog)

freeze(decl)#

Create the table + indexes and write identical snapshots to both dbs.

Return type:

None

Parameters:

decl (TableDeclaration)

frozen(table)#

The frozen declaration for table, or None before first freeze.

Return type:

TableDeclaration | None

Parameters:

table (str)

register_external(decl)#

Snapshot a table created by a bespoke writer (e.g. TrackStore DDL).

The table must already exist in products.db; only the dual snapshots are written here.

Return type:

None

Parameters:

decl (TableDeclaration)

class adapt.persistence.products.TableDeclaration(table, owner_module, granularity, primary_key, index_columns, columns)#

Bases: object

One frozen table schema: ownership, keys, and column types.

Parameters:

  • table (str)

  • owner_module (str)

  • granularity (str)

  • primary_key (tuple[str, ...])

  • index_columns (tuple[str, ...])

  • columns (tuple[tuple[str, str], ...])

columns: tuple[tuple[str, str], ...]#
granularity: str#
index_columns: tuple[str, ...]#
owner_module: str#
primary_key: tuple[str, ...]#
table: str#
class adapt.persistence.products.TableWriter(ledger, spec, owner_module)#

Bases: object

Writes one module’s product table; freezes on the first non-empty frame.

Parameters:

  • ledger (SchemaLedger)

  • spec (ProductTableWrite)

  • owner_module (str)

write(df, meta)#

Stamp identity, enforce the frozen schema, and upsert on the declared key.

Return type:

None

Parameters:

  • df (DataFrame)

  • meta (PersistenceMeta)

adapt.persistence.products.table_granularity(primary_key)#

Derive a table’s granularity from its declared primary key.

Return type:

str

Parameters:

primary_key (Sequence[str])

TrackStore — read/write the three track persistence tables.

Tables managed: - cells_by_scan : one row per active tracked cell per scan (wide canonical table) - cell_events : authoritative lineage edges (CONTINUE/SPLIT/MERGE/INITIATION/TERMINATION) - cell_tracks : convenience lifecycle summary per cell_uid

A “track” is a single connected chain of cell observations across scans identified by a stable cell_uid.

The tables live in the collection’s products.db: created with bespoke DDL (composite UNIQUE + retroactive-update SQL the generic writer cannot express) and frozen through the shared SchemaLedger on the first write — new columns are rejected forever after; there is no ALTER.

class adapt.persistence.track_store.TrackStore(db_path, ledger)#

Bases: object

Read/write track persistence tables in a collection’s products.db.

Thread-safe via SQLite WAL mode and an internal lock.

Parameters:
close()#
Return type:

None

get_cell_events(run_id, cell_uid=None)#
Return type:

DataFrame

Parameters:

  • run_id (str)

  • cell_uid (str | None)

get_cell_tracks(run_id)#
Return type:

DataFrame

Parameters:

run_id (str)

get_cells_by_scan(run_id, scan_id)#

All cells_by_scan rows for one scan, keyed by identity.

Return type:

DataFrame

Parameters:

  • run_id (str)

  • scan_id (str)

get_track_history(run_id, cell_uid)#
Return type:

DataFrame

Parameters:

  • run_id (str)

  • cell_uid (str)

write_scan(run_id, scan_time, cell_stats_df, tracked_cells_df, cell_events_df, cell_adjacency_df, *, scan_id)#

Persist one scan’s track outputs to the three tables.

Parameters:

  • run_id (str)

  • scan_time (datetime)

  • cell_stats_df (DataFrame)

  • tracked_cells_df (DataFrame)

  • cell_events_df (DataFrame)

  • cell_adjacency_df (DataFrame)

  • scan_id (str)

Return type:

None

Mechanical persistence of module-declared outputs.

The router knows spec TYPES, never module key names. Adding a module output means declaring a spec on the module — this file does not change.

Skip/raise rules: a key absent from the result means the module did not run this scan (skip); an empty DataFrame means no rows (skip); everything else that prevents a declared write is an error and raises.

class adapt.persistence.output_router.StoreOutputRouter(collection)#

Bases: object

Persist module-declared outputs into the store. No science names.

Dispatch: NetcdfArtifact → immutable object + catalog row + lineage to the scan’s raw volume + scan-product link; ProductTableWrite → products.db rows (frozen first-frame schema) + table link; TrackTablesWrite → TrackStore on products.db + auxiliary tracking link.

Parameters:

collection (Collection)

persist(modules, result, meta)#
Return type:

None

Parameters:

  • modules (Iterable)

  • result (dict)

  • meta (PersistenceMeta)

Execution History — a permanent, queryable scientific metadata product.

Records each module execution (run_modules aggregates) and structured warnings/errors (run_events) against the ONE run lifecycle in the store registry, so questions like “which runs failed?”, “which module got slower?”, “what config produced these outputs?” are answered with SQL, never by parsing log files. Consumes only contracts DTOs — never the observability implementation.

class adapt.persistence.execution_history.StoreExecutionHistory(registry)#

Bases: object

Execution history on the store registry (runs / run_modules / run_events).

The ONE run lifecycle lives in registry.runs (begun by the orchestrator via StoreRegistry.begin_run); this class records per-module aggregates and warning/error events against it and maps the end-of-run summary onto the terminal run status.

Parameters:

registry (StoreRegistry)

finalize_run(summary)#

Map the end-of-run summary onto the single registry run lifecycle.

Return type:

None

Parameters:

summary (RunSummary)

record_errors(run_id, errors)#
Return type:

None

Parameters:

  • run_id (str)

  • errors (list[ErrorEvent])

record_modules(run_id, scan_id, spans, recorded_at)#

Fold one scan’s module spans into the per-module run aggregates.

Return type:

None

Parameters:

  • run_id (str)

  • scan_id (str)

  • spans (list[SpanRecord])

  • recorded_at (datetime)

record_warnings(run_id, warnings)#
Return type:

None

Parameters:

  • run_id (str)

  • warnings (list[WarningEvent])

Store-level exceptions, shared by every persistence component.

exception adapt.persistence.errors.AlreadyInitializedError#

Bases: StoreError

The root already contains an initialized store.

exception adapt.persistence.errors.StoreError#

Bases: Exception

A store-level operation failed.

On this page
Show Source