Skip to main content

functions

Shared helper functions for the file metadata runtime.

This module is the single source of truth for pure helpers and internal utilities used across the other modules in this package:

  • flow.py — Prefect flow entry-point
  • refresh.py — cron-driven refresh flow
  • tasks.py — Prefect tasks

Module​

Functions​

generate_prefect_task_hash​

def generate_prefect_task_hash(pod_name: str, datasource_name: str) ‑> str:

Return a stable, fixed-length task hash for a (pod, datasource) pair.

The hash is derived from "{pod_name}/{datasource_name}" so that it is deterministic across restarts and serves as the composite primary-key prefix in FileMetadata and Run.

This is the datasource leaf hash, and it plays two roles:

  • It keys the datasource/run-level tables directly — file_metadata, scan_metadata, runs, schema_versions — none of which a DAG step writes.
  • It is the leaf mixed into every per-step Merkle task_hash (bitfount.flows.dag.hashing), which is what keeps two datasources on the same pod in separate partitions of a step's table.

Arguments

  • pod_name: The name of the pod.
  • datasource_name: The name of the datasource within the pod.

Returns A 32-character hex string (the first 128 bits of the SHA-256 digest).

make_dataset_identifier​

def make_dataset_identifier(pod_name: str, datasource_name: str) ‑> str:

Return the "{pod_name}/{datasource_name}" dataset identifier.

This is the same slash-joined key generate_prefect_task_hash hashes, and the value stored in FileMetadata.dataset_identifier.

open_background_cache​

def open_background_cache(    pod_name: str, *, pod_key_path: Path | None = None, create_dir: bool = False,) ‑> CacheProtocol:

Open the per-pod background cache DB.

Single source of truth for resolving the per-pod background cache path (PODS_CACHE_ROOT/<pod_name>/BACKGROUND_CACHE_FILENAME) and opening it, shared by the DATASET_PROJECT_LINKED gRPC pod handler and the run_background_task CLI so the two entry points cannot drift.

Arguments

  • pod_name: Name of the pod whose cache DB to open.
  • pod_key_path: Path to the pod's pod_rsa.pem (typically pod.pod_key_path). When provided, it is bound into the cache-encryption layer so encrypted columns can derive their key from this pod's RSA private key. Optional so test/CLI flows that don't need encryption can still call with just pod_name.
  • create_dir: When True the parent directory is created if missing — the live pod handler path, where the DB is created on first write. When False the cache file must already exist or FileNotFoundError is raised — the CLI path, which requires the pod to have been started at least once to populate file metadata.

Returns A CacheProtocol instance for the pod's background cache.

Raises

  • FileNotFoundError: When create_dir is False and the cache file does not exist.

resolve_datasource_inputs​

def resolve_datasource_inputs(datasource: Any) ‑> tuple[str | None, str | None]:

Return the (path, connection_string) a datasource should be indexed by.

file_metadata_runtime takes both as separate parameters and routes on whichever is set, so every caller holding a live datasource has to make the same decision: filesystem sources index by their directory path, SQL sources by their connection string, anything else by neither.

Written once here because it was previously repeated at each trigger site, and a site that forgot a branch would not fail — it would submit a run with both inputs None, which the flow rejects only once it is already running.

Note this resolves from a live datasource instance. The flow itself routes from a resolved class (via issubclass), because it receives a type string rather than an object; the two are deliberately not merged.

Arguments

  • datasource: A constructed datasource instance.

Returns (path, connection_string) with at most one set. Both are None for a datasource of neither kind, which callers may treat as "nothing to index".

split_dataset_identifier​

def split_dataset_identifier(dataset_identifier: str) ‑> tuple[str, str]:

Split a dataset identifier back into (pod_name, datasource_name).

Splits on the first "/" so re-joining reproduces the original string (and therefore the same task hash) even when the datasource name itself contains a "/".

Global variables​

  • BACKGROUND_CACHE_FILENAME : str - Filename of the per-pod background cache DB.
  • PODS_CACHE_ROOT : pathlib.Path - Root directory under which per-pod cache directories are stored.