Skip to main content

datasource_manager

Per-pod datasource lifecycle manager.

Owns the datasource collection (base_datasources) and the operations that load, schema-check, and register them with the hub.

Responsibilities:

  • Processing datasource arguments and constructing DatasourceContainers.
  • Triggering schema generation (delegated to SchemaManager).
  • Registering finalised schemas with the hub.
  • Detecting which datasources still need schema generation.

Does NOT own:

  • Schema creation/validation logic (see schema_manager.py).
  • Config-file polling and hot-reload (see pod_config_reload.py).

Classes​

DatasourceManager​

class DatasourceManager(    *, hub: BitfountHub, pod_name: str, schema_manager: SchemaManager,):

Datasource collection owner: load, schema-check, register.

Holds base_datasources and coordinates with SchemaManager for schema generation. Used by both Pod and DatasourceConfigReloader for datasource lifecycle operations.

Variables​

  • datasources_needing_schema_generation : dict[str, DatasourceContainer] - Datasources that need schema generation via Prefect.

    Filters out datasources with pre-defined schemas.

Methods​

load_basesource_schema_if_necessary​

def load_basesource_schema_if_necessary(    self,    ds: DatasourceContainerConfig,    update_schema: bool = False,    pod: Pod | None = None,) ‑> DatasourceContainer:

Load schema for a base datasource.

Arguments

  • ds: The datasource container config to load schema for.
  • update_schema: Whether to force schema regeneration.
  • pod: The Pod instance (used for hook callbacks).

process_datasource_args​

def process_datasource_args(    self, datasources: Iterable[DatasourceContainerConfig],) ‑> list[DatasourceContainerConfig]:

Load supplied datasources into expected format by datasource type. Note Mutates data_config on each datasource in-place to apply schema config migration (legacy nested-by-name format).

register_schema​

def register_schema(    self,    datasource_container: DatasourceContainer,    *,    ds_add_number_of_records: bool = False,) ‑> None:

Update the schema for a datasource on the hub.

Arguments

  • datasource_container: The datasource container to register.
  • ds_add_number_of_records: Whether to include record count.

trigger_background_schema_generation​

def trigger_background_schema_generation(    self, *, force_stop_existing: bool = False,) ‑> None:

Kick off full background schema generation for eligible datasources.

Arguments

  • force_stop_existing: If True, stop any in-progress schema generation before starting.

update_pod_name​

def update_pod_name(self, pod_name: str) ‑> None:

Update the cached pod name (e.g. after identifier-length enforcement).

The manager caches the pod name at construction; enforcement can shorten it afterwards. Keeping the cache in sync ensures the fallback display name and description used for datasets reflect the enforced pod name.