Skip to main content

flow

Prefect flow for the file metadata runtime.

Entrypoint​

file_metadata_runtime Top-level flow — call this to collect and persist metadata for a single datasource. The orchestrator triggers it via a Prefect deployment (run_deployment("background/file_metadata_runtime", ...)) immediately after a pod is successfully started.

Datasource routing​

The flow receives a datasource_type string (e.g. "bitfount.CSVSource") and resolves it to the actual class using the same lookup path that setup_datasource uses in bitfount.runners.config_utils:

  1. DataSourceType(datasource_type).name — resolves the enum value (e.g. "bitfount.CSVSource") to its enum member name (e.g. "CSVSource"), falling back to the raw string for plugin datasources not in the enum.
  2. getattr(importlib.import_module("bitfount.data"), cls_name) — imports the class from the bitfount.data public surface.
  3. issubclass(cls, FileSystemIterableSource) — determines whether the datasource walks a directory of files, without maintaining a hardcoded allowlist of type names.
  4. issubclass(cls, _SQLSource) — identifies DB-backed datasources whose only input is a connection string.
  5. Everything else with a path arg falls through to single-file metadata (e.g. CSVSource).

Run-state tracking​

Run is written to the cache DB at flow start (status "running") and updated on completion ("complete") or failure ("failed"). The refresh flow reads this table before re-triggering to avoid running two indexing passes for the same datasource concurrently.

Completion signalling​

When orchestrator_callback_url is provided, the flow POSTs a completion payload to that URL via the notify_orchestrator task (retries=3, retry_delay_seconds=5) so the orchestrator can forward a file_metadata_complete Socket.IO event to the desktop app. The notification is sent regardless of whether the flow succeeded or failed, before any exception is re-raised.

Flow timeout​

timeout_seconds is read from settings.metadata_flow_timeout_seconds, which defaults to None (no timeout) — the same as Prefect's own default. A walk over a very large tree is legitimately long-running and cannot be told apart from a wedged run by duration alone, so no cap is baked in; liveness is detected via flow-run heartbeats and the zombie-reaper automation instead. Operators who know their expected upper bound may set one via BITFOUNT_METADATA_FLOW_TIMEOUT_SECONDS. Note this is read at import time, so the environment must be set before the flow module loads.

Config-error handling​

Misconfigured datasources (unsupported type, missing path, missing connection string) raise ConfigError (a ValueError subclass shared with scan_metadata_runtime; see bitfount.runtimes.deployments). This propagates through the except Exception handler which calls mark_run_failed and stores the error message, then is re-raised so that Prefect records the flow run as failed rather than successful. Operators monitoring Prefect flow state therefore see an accurate signal without needing to inspect the callback payload or the cache DB.

Module​

Functions​

file_metadata_runtime​

def file_metadata_runtime(    pod_name: str,    datasource_name: str,    datasource_type: str,    cache_db_path: str,    orchestrator_callback_url: str | None = None,    path: str | None = None,    connection_string: str | None = None,    only_paths: list[str] | None = None,) ‑> int:

Collect and persist file metadata for a single datasource.

This flow is triggered as a fire-and-forget background task immediately after a pod is successfully started (i.e. the dataset has been created). The orchestrator submits it via run_deployment("background/file_metadata_runtime", timeout=0) so it never blocks the HTTP response.

A Run is written at the start (status "running") and updated on completion or failure. The refresh flow checks this record before re-triggering to avoid concurrent indexing passes for the same datasource.

On completion the flow POSTs a file_metadata_complete payload to orchestrator_callback_url (when provided) so the orchestrator can forward the event to the desktop app via Socket.IO. The notification is sent whether the flow succeeds or fails.

Datasource routing uses issubclass checks against the resolved class so that newly added subclasses are handled automatically without any changes to this module:

  • FileSystemIterableSource subclasses → directory walk.
  • _SQLSource subclasses → DB connection string row.
  • Everything else with a path → single-file metadata (e.g. CSV).
  • Unresolvable or unsupported types → logged and skipped.

Arguments

  • pod_name: Name of the pod (used to derive task_hash).
  • datasource_name: Name of the datasource within the pod.
  • datasource_type: Datasource type string from the pod config, e.g. "bitfount.CSVSource" or "bitfount.DICOMSource".
  • cache_db_path: Filesystem path to the pod's background_cache.db SQLite file. The flow opens its own cache connection for the duration of the run and closes it on exit.
  • orchestrator_callback_url: Optional URL of the orchestrator's POST /pod/file-metadata-complete endpoint. When set, the flow POSTs a completion payload so the orchestrator can emit a file_metadata_complete Socket.IO event to the desktop app.
  • path: File path (single-file sources) or directory path (FileSystemIterableSource subclasses).
  • connection_string: DB connection string (_SQLSource subclasses).
  • only_paths: When set, index exactly these paths (files or directories) instead of walking path, and skip the deletion prune. Used by the file watcher, which has already identified what is new. Ignored for non-FileSystemIterableSource datasources, which have no walk to scope. Any spelling is accepted — these and path are both normalized before being compared — and a path that lands outside path is refused with a warning. See bitfount.runtimes.file_metadata.tasks._iter_stat_targets.

Returns The number of FileMetadata records written to the cache.