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, hash-reusing generator that recursively walks a directory and yields metadata for every file found (used for FileSystemIterableSource subclasses), 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,) ‑> int:

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. Hash reuse lives inside iter_directory_file_metadata — unchanged files are not re-hashed.

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.

Returns The number of rows written across all chunks.

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,) ‑> collections.abc.Iterator[FileMetadataRecord]:

Yield one FileMetadataRecord 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.

Hash reuse: a file's stored file_hash is reused when a persisted inventory row exists for the same file_path with a matching file_size_bytes and modified_at. Only new or changed files are hashed, so a re-run (or a resume after a kill) pays the hashing 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 hashes.
  • 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-hash 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).

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).