Skip to main content

tasks

Prefect tasks for the file metadata runtime.

Each task is a discrete, independently-retryable unit of work:

collect_single_file_metadata Collects metadata for a single file (used for CSVSource).

iter_directory_file_metadata Streaming generator that recursively walks a directory and yields a WalkedFile for every file found (used for FileSystemIterableSource subclasses), carrying a record only for files whose stored row no longer describes them, so the flow can persist bounded, crash-resumable chunks.

collect_db_connection_metadata Records a DB connection string as a single metadata row (used for OMOPSource).

store_file_metadata Persists a batch of FileMetadataRecord instances into the cache via bulk_insert_file_metadata. The caller is responsible for constructing the cache instance and ensuring its parent directory exists before calling this task.

notify_orchestrator POSTs the flow's completion payload to the orchestrator's callback URL.

index_filesystem_datasource / collect_and_store_records Per-datasource indexing strategies that compose the tasks above. They live here, not in functions, because functions is the leaf layer this module imports — putting composers there would invert that arrow and force function-level imports.

Module​

Functions​

collect_and_store_records​

def collect_and_store_records(    datasource_cls: type,    pod_name: str,    datasource_name: str,    datasource_type: str,    cache: CacheProtocol,    task_hash: str,    run_id: str,    path: str | None,    connection_string: str | None,) ‑> int:

Index a non-filesystem datasource: collect a record list, store it once.

Covers the two datasource shapes whose entire input fits in memory — a DB connection string (_SQLSource) and a single file (e.g. CSVSource) — so the flow's routing reads as one dispatch rather than three inline bodies.

Arguments

  • datasource_cls: The resolved datasource class, already known not to be a FileSystemIterableSource.
  • pod_name: Pod name (written to every record).
  • datasource_name: Datasource name (written to every record).
  • datasource_type: Datasource type string, used in error messages.
  • cache: Open cache for the pod's background_cache.db.
  • task_hash: Provenance column stamped on each row.
  • run_id: The Run these records belong to.
  • path: File path for single-file datasources.
  • connection_string: Connection string for _SQLSource subclasses.

Returns The number of rows written.

Raises

  • ConfigError: When the datasource's required input is missing, or the type has neither a usable path nor a connection string.

collect_db_connection_metadata​

def collect_db_connection_metadata(    connection_string: str, pod_name: str, datasource_name: str, datasource_type: str,) ‑> list[FileMetadataRecord]:

Record a DB connection string as a single metadata row.

For DB-backed datasources (e.g. OMOPSource) there are no individual files to enumerate. Instead, the connection string itself is stored as the file_path so the datasource's presence is captured in the metadata cache.

Arguments

  • connection_string: The SQLAlchemy-compatible DB connection string.
  • pod_name: Name of the pod (written to the record for refresh lookups).
  • datasource_name: Name of the datasource (written to the record).
  • datasource_type: Datasource type string (written to the record).

Returns A list containing one FileMetadataRecord whose optional fields are all None.

collect_single_file_metadata​

def collect_single_file_metadata(    file_path: str, pod_name: str, datasource_name: str, datasource_type: str,) ‑> list[FileMetadataRecord]:

Collect filesystem metadata for a single file (CSV datasource).

Arguments

  • file_path: Absolute or relative path to the file.
  • pod_name: Name of the pod (written to the record for refresh lookups).
  • datasource_name: Name of the datasource (written to the record).
  • datasource_type: Datasource type string (written to the record).

Returns A list containing one FileMetadataRecord, or an empty list if the file cannot be stat'd (permissions / IO error).

Raises

  • FileNotFoundError: When file_path does not exist. A missing configured path is treated as a hard failure so the run is marked failed and the operator is notified, rather than silently completing with zero files processed.

index_filesystem_datasource​

def index_filesystem_datasource(    path: str,    pod_name: str,    datasource_name: str,    datasource_type: str,    cache: CacheProtocol,    task_hash: str,    run_id: str,    log: logging.Logger | logging.LoggerAdapter[Any],    only_paths: Collection[str] | None = None,    stats: IndexingStats | None = None,) ‑> int | None:

Walk path, persisting bounded chunks, then prune files that vanished.

Lives beside the tasks it composes rather than in functions, which is the leaf layer tasks itself imports. Kept out of file_metadata_runtime because it is the only branch of the routing that is more than "collect a list, store it": it streams, persists incrementally, and reconciles deletions, which together accounted for most of the flow's branching.

Chunked on purpose: records are persisted as the walk yields them rather than accumulated and written once at the end, so a kill mid-walk keeps every completed chunk. Unchanged files are neither re-hashed nor rewritten; that gate lives inside iter_directory_file_metadata.

An unscoped pass returns without walking when the tree has not changed since the last completed pass (see _can_short_circuit).

A path the walk cannot read is checked against the root. If the root is unreachable the walk stops, the prune is skipped and the failure is counted in stats.transient_failures, for the caller to end the run partial. If it is reachable the path is skipped and its rows are kept out of the prune.

Arguments

  • path: Root directory to walk.
  • pod_name: Pod name (written to every record).
  • datasource_name: Datasource name (written to every record).
  • datasource_type: Datasource type string (written to every record).
  • cache: Open cache for the pod's background_cache.db.
  • task_hash: Provenance column stamped on each row.
  • run_id: The Run this walk belongs to, stamped on each row.
  • log: The flow's run logger.
  • only_paths: When set, index exactly these paths and skip the deletion prune. A scoped pass sees only its own paths, so seen_paths is deliberately partial and pruning against it would delete every row the pass did not touch. Any spelling — _iter_stat_targets normalizes each path before checking it against the root.
  • stats: When set, per-file hashing and per-chunk flush figures are recorded on it, and a progress event is emitted per chunk.

Returns The number of files seen, written or unchanged, or None when the pass returned without walking.

iter_directory_file_metadata​

def iter_directory_file_metadata(    directory_path: str,    pod_name: str,    datasource_name: str,    datasource_type: str,    cache: CacheProtocol,    only_paths: Collection[str] | None = None,    stats: IndexingStats | None = None,    skips: WalkSkips | None = None,) ‑> collections.abc.Iterator[WalkedFile]:

Yield one WalkedFile per file under directory_path.

The directory is traversed recursively. Filtering by extension or other criteria is the responsibility of the datasource itself, not this runtime.

This is a plain generator (not a Prefect task) so the flow can persist bounded, crash-resumable chunks as it walks, rather than accumulating every record in memory and writing once at the end.

Unchanged skip: a file whose stored row still describes it (see _is_unchanged) yields a WalkedFile with no record, so the caller neither rewrites the row nor prunes it. Hash reuse: a file that is rewritten but whose stored row has a matching file_size_bytes and modified_at keeps its stored file_hash rather than being re-hashed. So a re-run (or a resume after a kill) pays the hashing and write cost only for files that actually changed. The rare mtime-spoof miss (same size + mtime, new content) is accepted.

Arguments

  • directory_path: Root directory to walk.
  • pod_name: Pod name (written to every record; derives the task_hash).
  • datasource_name: Datasource name (written to every record).
  • datasource_type: Datasource type string (written to every record).
  • cache: Open cache — read once up-front to preload prior rows.
  • only_paths: When set, index exactly these paths instead of walking directory_path. Paths outside the root, and paths that have vanished, are skipped with a warning. The prior-row preload is narrowed to these paths, so a scoped pass never reads the whole inventory. Relative, ..-containing and symlinked spellings are all accepted: each path is resolved before it is checked against the root and against the stored rows, so it matches rather than failing closed (see _iter_stat_targets).
  • stats: When set, records per-file hashing, reuse and skip counts on it.
  • skips: Told of every path the walk could not read. When one is skipped with the datasource root unreachable, the walk stops there.

Raises

  • FileNotFoundError: When directory_path does not exist. A missing configured path is treated as a hard failure so the run is marked failed and the operator is notified, rather than silently completing with zero files processed.

notify_orchestrator​

def notify_orchestrator(callback_url: str, payload: dict[str, Any]) ‑> None:

POST the completion payload to the orchestrator callback URL.

Retried up to three times with a 5-second delay between attempts. If all retries are exhausted Prefect marks the task as failed, but the parent flow run is unaffected — metadata has already been written to the cache successfully.

Arguments

  • callback_url: Full URL of the orchestrator's POST /pod/file-metadata-complete endpoint.
  • payload: JSON-serialisable dict containing pod_name, datasource_name, task_hash, files_processed, status, and error.

store_file_metadata​

def store_file_metadata(    records: list[FileMetadataRecord],    task_hash: str,    cache: CacheProtocol,    run_id: str | None = None,) ‑> int:

Persist records into the metadata cache.

Each record already contains the provenance fields pod_name, datasource_name and datasource_type, populated by the collection tasks.

The caller is responsible for constructing the cache instance and ensuring its backing store (e.g. parent directory for a SQLite file) exists before invoking this task.

Arguments

  • records: FileMetadataRecord instances as returned by the collection tasks.
  • task_hash: Provenance column stamped on each row (the inventory's primary key is file_path alone).
  • cache: An open cache instance satisfying CacheProtocol.
  • run_id: Optional UUID of the Run that produced these records. Stored on each row for provenance tracking.

Returns The number of rows written (0 when records is empty).

Classes​

WalkSkips​

class WalkSkips(probe: RootProbe):

What a walk could not read, and whether that was the root going away.

Fed from the walk's pool threads and from the consumer. Asked by the consumer after each file whether the root has been lost; the root probe is consulted only when something new was skipped, so a clean walk never probes.

Arguments

  • probe: Answers whether the datasource root is reachable.

Variables​

  • paths : list[str] - Every path skipped so far, in the walk's spelling.
  • probe : RootProbe - The root probe consulted.

Methods​

record​

def record(self, path: str, error: OSError) ‑> None:

Note that path could not be read. Safe from any thread.

root_lost​

def root_lost(self) ‑> bool:

Whether a skip so far happened with the root unreachable.

WalkedFile​

class WalkedFile(    path: ForwardRef('str'), record: ForwardRef('FileMetadataRecord | None'),):

One file seen by iter_directory_file_metadata.

Variables​

  • path : str - Alias for field number 0