Skip to main content

data.configmanager

initialize_config​

def initialize_config(file_path: str = "Files/Configuration/config.yaml")

Load and cache the global ConfigManager from a YAMLA simple way to write configurations. It's basically a list that computers can read easily. file.

Uses double-checked locking so only the first call reads from disk; subsequent calls return the cached instance.

Arguments:

  • file_path - Path to the config.yaml file.

Returns:

The initialised ConfigManager singleton.

Raises:

  • FileNotFoundError - If the config file does not exist.

get_config​

def get_config()

Get the cached ConfigManager instance.

Returns:

The previously initialised ConfigManager.

Raises:

  • RuntimeError - If initialize_config has not been called yet.

Model Objects​

@dataclass
class Model(DataClassFromDictMixin)

Model reference within ConfigManager.

Holds metadata about an analytical model including its workspace, compatibility level, and file location within the lakehouseA place where you store both "raw" data (like files) and "organized" data (like tables). It combines the best of a File Cabinet and a Database..

Function Objects​

@dataclass
class Function(DataClassFromDictMixin)

Azure function configuration settings.

Defines which azure functions are available

Backup Objects​

@dataclass
class Backup(DataClassFromDictMixin)

Backup configuration settings.

Defines which layer and storage account to back up to, including optional folder exclusions and schema overrides.

Lakehouse Objects​

@dataclass
class Lakehouse(DataClassFromDictMixin)

Lakehouse connection configuration for a single data layer.

Stores the workspace, lakehouse name, schema, and mount point details used to resolve ABFS paths for bronze, silver, or gold layers.

ConfigManager Objects​

@dataclass
class ConfigManager(DataClassFromDictMixin)

Central configuration holder loaded from config.yaml.

Manages tenant credentials, lakehouse mount points, model references, and global settings that control the data loading pipelineAn automated "conveyor belt" that moves data from one place to another or performs a task automatically. behaviour.

Attributes:

  • verboselogging bool - When true, log full schema and per-step detail instead of concise messages. Defaults to False.
  • run_prebronzenotebook bool - Whether the pre-bronze notebook hook runs before the bronze load. Defaults to True.
  • run_postbronzenotebook bool - Whether the post-bronze notebook hook runs after the bronze load. Defaults to True.
  • run_presilvernotebook bool - Whether the pre-silver notebook hook runs before the silver load. Defaults to True.
  • run_postsilvernotebook bool - Whether the post-silver notebook hook runs after the silver load. Defaults to True.
  • process_sample_rows int - When greater than 0, only this many rows are processed per object (useful for testing); 0 processes all rows. Defaults to 0.
  • silverloadretry int - Maximum retry attempts for a silver load before failing. Defaults to 5.
  • silverloaddelay int - Delay in seconds between silver load retry attempts. Defaults to 10.
  • objectprimarykeycolumn str - Name of the system primary-key column added to loaded tables. Defaults to "SYSTEMPRIMARYKEY".
  • objecttimestampcolumn str - Name of the system state-timestamp column used for history and snapshots. Defaults to "SYSTEMSTATETIMESTAMP".
  • skip_invalid_objects bool - When true, an object YAML that fails to parse or validate during a folder build is skipped with a warning and the build continues with the valid objects. When false, the first invalid object definition aborts the build with a FabricConfigError naming the file, independent of stop_at_error. Defaults to True.

from_yaml​

@classmethod
def from_yaml(cls, yaml_string: str)

Create a Model instance from a YAML string.

from_yaml_file​

@classmethod
def from_yaml_file(cls, file_path: str = "Files/Configuration/config.yaml")

Create a Model instance from a YAML file.

get_lakehouse_by_layer​

def get_lakehouse_by_layer(_layer: str) -> Lakehouse

Get the Lakehouse configuration for a given layer.

Arguments:

  • _layer - The layer name (e.g. "bronze", "silver", "gold").

Returns:

The matching Lakehouse instance.

Raises:

  • Exception - If no lakehouse is configured for the requested layer.

get_layer_by_lakehouse​

def get_layer_by_lakehouse(lakehouse_name: str) -> Optional[str]

Lookup the layer associated with a given lakehouse name.

Arguments:

  • lakehouse_name str - Name of the lakehouse to look up.

Returns:

  • Optional[str] - The layer name if found, otherwise None.

get_model​

def get_model(model_name: str) -> Model

Get a model by name from the models list.

Arguments:

  • model_name str - Name of the model to retrieve

Returns:

  • Model - The model object matching the given name

Raises:

  • Exception - If the model is not found in configuration

log_dataframe_schema​

def log_dataframe_schema(df) -> None

Logs the schema of a Spark DataFrame at debug level in pretty-printed JSON format.

:param df: Spark DataFrame

log_verbose_message​

def log_verbose_message(message: str) -> None

Logs message to debug or the info (if verbose logging is enabled).

:param message: string