Skip to main content

bookkeeping

Shared run scaffolding for the background metadata flows.

file_metadata_runtime and scan_metadata_runtime differ only in the work they do in the middle. Around that work they were identical: open the cache, insert a bookkeeping runs row, complete it on success, fail it on error (without letting a second failure in that write mask the original), close the cache, log a one-line summary, optionally POST a completion callback, and re-raise so Prefect records the run as failed rather than successful.

metadata_run owns all of it, so the two flows contain their routing and nothing else. Getting any step of that sequence subtly wrong is a real hazard — the ordering matters (notify must happen after the cache is closed but before the re-raise, or a failed run never reports) and it was previously written out twice.

Module​

Functions​

metadata_run​

def metadata_run(    *,    flow_name: str,    run_type: str,    pod_name: str,    datasource_name: str,    datasource_type: str,    cache_db_path: str,    log: logging.Logger | logging.LoggerAdapter[Any],    count_key: str,    orchestrator_callback_url: str | None = None,) ‑> collections.abc.Iterator[MetadataRun]:

Wrap a metadata flow body in its cache, bookkeeping and reporting.

On entry: opens the cache and inserts a running bookkeeping row. On a clean exit: completes that row with run.count. On an exception: fails the row, logs the summary, sends the callback, then re-raises so Prefect records the run as failed.

dedup=False on the runs insert is deliberate and applies to both flows: mutual exclusion lives at the trigger layer, backed by the heartbeat and the zombie-reaper automation. Keeping the row's own guard would re-introduce the crash-lockout it was demoted to fix — a crashed run's stale running row would block a fresh, already-deduped run. The rows a crash leaves behind are cleaned by bitfount.runtimes.recovery at the next process start.

Arguments

  • flow_name: Flow name used in the start/finish log lines.
  • run_type: runs-table discriminator, e.g. "file_metadata".
  • pod_name: Name of the pod.
  • datasource_name: Name of the datasource within the pod.
  • datasource_type: Datasource type string, logged for context.
  • cache_db_path: Path to the pod's background_cache.db.
  • log: The flow's run logger.
  • count_key: Payload key for the row count ("files_processed" / "scans_processed"), which the desktop app reads.
  • orchestrator_callback_url: When set, a completion payload is POSTed here whether the flow succeeded or failed.

Raises

  • RuntimeError: If the bookkeeping insert returns no run_id. With dedup=False that is documented not to happen, so it means the store's contract changed — failing loudly beats indexing with no row to record the outcome against.
  • BaseException: Whatever the flow body raised, re-raised after the row is failed and the callback sent.

Classes​

MetadataRun​

class MetadataRun(    cache: CacheProtocol,    run_id: str,    task_hash: str,    count: int = 0,    status: str = 'success',):

Mutable handle a flow uses to report progress back to metadata_run.

The flow body sets count to the number of rows it persisted, or calls skip() when there is legitimately nothing to do. Everything else — cache, run_id, status — is managed by the context manager.

Variables​

  • static cache : CacheProtocol - Open cache for the pod's background_cache.db, closed on exit.
  • static count : int - Rows the flow persisted. The flow body assigns this.
  • static run_id : str - The bookkeeping runs row this flow is recording against.
  • static status : str - "success", "skipped" or "error" — reported in the summary log and the completion callback.
  • static task_hash : str - Provenance hash for the (pod, datasource) pair.

Methods​

skip​

def skip(self, reason: str) ‑> None:

Mark this run a no-op success and record why.

Used when the flow has nothing to do but has not failed — e.g. scan_metadata handed a non-ophthalmology datasource. Completes the bookkeeping row with zero rows so it does not linger as running.