exlab_wizard.lims.cache#

aiosqlite-backed cache for LIMS project rows. Backend Spec §7.2.4.

The cache lives at <xdg_cache_home>/exlab-wizard/lims_cache.db and holds a flat row-per-project table keyed by (lims_endpoint, short_id). Storing the endpoint as part of the primary key means a single workstation can switch between LIMS instances during testing without colliding rows.

Access pattern:

  • LIMSCache.upsert_many() is called by LIMSClient.list_projects on every successful refresh. The last_refreshed column is stamped with the wizard’s current UTC ISO time at write.

  • LIMSCache.list_projects() and LIMSCache.get_project() are consulted before the network for §7.2.4 cache-first lookups.

  • LIMSCache.is_fresh() returns True when the most recent last_refreshed for the endpoint is within ttl_hours. Callers use this to decide whether to short-circuit a network refresh.

  • The cache is never consulted for write paths – there are no writes to LIMS in v1.

The rows we materialize back into LIMSProject use the fetched_at column from last_refreshed. metadata is round-tripped through the JSON column metadata_json.

Classes

LIMSCache(db_path, *[, ttl_hours])

SQLite TTL cache for LIMS project rows.
class exlab_wizard.lims.cache.LIMSCache(db_path, *, ttl_hours=24)[source]#

Bases: object

SQLite TTL cache for LIMS project rows. Backend Spec §7.2.4.

The cache is async-native via aiosqlite so that LIMSClient can interleave cache reads with httpx network calls without blocking the event loop.

Parameters:
async close()[source]#

Close the underlying aiosqlite connection.

Return type:

None

async get_project(endpoint, uid_or_short_id)[source]#

Return one project by uid or short_id, or None if absent.

Parameters:

  • endpoint (str)

  • uid_or_short_id (str)

Return type:

LIMSProject | None

async init()[source]#

Create the table and index if absent. Idempotent.

Return type:

None

async is_fresh(endpoint)[source]#

True iff the most recent last_refreshed is within ttl_hours of the wizard’s current UTC time. False when the cache has no rows for endpoint.

Parameters:

endpoint (str)

Return type:

bool

async list_projects(endpoint, *, status_filter=None)[source]#

Return every cached project for endpoint, optionally filtered by status_filter (an OR of allowed status values).

Parameters:
Return type:

list[LIMSProject]

async upsert_many(endpoint, projects)[source]#

Insert or update every row. last_refreshed is taken from each LIMSProject.fetched_at; the caller stamps that value before invoking this method so a single refresh batch shares one timestamp.

Parameters:
Return type:

None