load_meta_data

The load_meta_data module provides the foundational logic for reading and parsing the YAMLA simple way to write configurations. It's basically a list that computers can read easily. configuration files that drive the automated platform. It translates your project definitions into the technical objects used by the Bronze, Silver, and Gold loaders.

clear_objects_cache

def clear_objects_cache() -> None

Clear the in-memory metadata cache.

Removes all cached TableConfig lists so subsequent calls to get_objects_by_folder will re-read from disk.

get_object_by_file

def get_object_by_file(
        tablefile: str,
        config_manager: ConfigManager = None) -> TableConfig | None

Load a single TableConfig from a YAML file path.

Delegates to get_objects_by_folder and returns the first result.

Arguments:

• tablefile - Path to the YAML table configuration file.
• config_manager - Optional ConfigManager instance; initialised automatically if not provided.

Returns:

The loaded TableConfig, or None if the file could not be loaded.

get_config_manager

def get_config_manager(config_filepath: str = None) -> ConfigManager

Load or return the global ConfigManager instance.

Arguments:

• config_filepath - Optional path to a config.yaml file. Uses the default location when not specified.

Returns:

The initialised ConfigManager singleton.

get_objects_by_folder

def get_objects_by_folder(
        table_folder: str | list[str],
        config_manager: ConfigManager = None,
        except_folders: list[str] = None,
        except_files: list[str] = None) -> list[TableConfig] | None

Loads metadata configurations from a specified folder or file and returns a list of TableConfig objects. If a single file is specified, metadata will be loaded only from that file. Otherwise, the function traverses the specified folder and loads metadata from all YAML files.

Each object file is loaded independently. When config_manager.skip_invalid_objects is True (the default), an object file that fails to parse or validate is logged and skipped, and the build continues with the remaining valid objects (a summary of the skipped files is logged). When it is False, the first invalid object definition aborts the build with a FabricConfigError naming the file, independent of stop_at_error — a malformed object definition is treated as an authoring error, not a recoverable data-load error.

Arguments:

• table_folder str - The path to the folder or file containing YAML metadata files. If it starts with "Files/", it will be resolved to the default 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. path.
• config_manager ConfigManager - An instance of ConfigManager used to initialize and validate configurations.
• except_folders List[str] - List of folders to exclude from the metadata loading.
• except_files List[str] - List of files to exclude from the metadata loading.

Returns:

• List[TableConfig] - A list of TableConfig dataclass instances created from the loaded YAML metadata.

Raises:

• Exception - If config_manager is not initialized or if an error occurs during metadata loading and config_manager.stop_at_error is set to True.
• FabricConfigError - If an object file is invalid and config_manager.skip_invalid_objects is False (raised regardless of stop_at_error).

get_tableconfig_by_inputfile

def get_tableconfig_by_inputfile(
        file_path: str,
        config_manager: ConfigManager = None) -> Optional[TableConfig]

Retrieves the best matching TableConfig for the given input filepath by extracting the folder path and finding the TableConfig whose bronzefolder is the longest prefix match to that folder path.

Arguments:

• file_path str - The path to the input file.
• config_manager ConfigManager - An instance of ConfigManager used to load configurations.

Returns:

• Optional[TableConfig] - The matching TableConfig if found, otherwise None.

Raises:

• Exception - If ConfigManager is not initialized.

get_object_trigger

def get_object_trigger(
        lakehousepath: str,
        config_manager: ConfigManager = None) -> list[TableConfig] | None

Find TableConfigs whose connections are triggerable for a lakehouse path.

Scans the Connections folder for YAML connection definitions, filters to those with istriggerable=True, and returns matching TableConfig objects.

Arguments:

• lakehousepath - The lakehouse file path to match against triggers.
• config_manager - Optional ConfigManager instance; initialised automatically if not provided.

Returns:

A list of matching TableConfigs, or None on error.

get_model_by_name

def get_model_by_name(model_name: str) -> Model | None

Load a model configuration by its name.

Looks up the model in the ConfigManager's model list and loads the corresponding YAML file.

Arguments:

• model_name - The name of the model as defined in config.yaml.

Returns:

The loaded Model instance, or None if loading fails and stop_at_error is False.

Raises:

• Exception - If the model name is not found or ConfigManager is not initialised.

get_model_by_file

def get_model_by_file(model_file: str,
                      config_manager: ConfigManager = None) -> Model | None

Loads a Model object by reading a given YAML file. Ensures that the configuration manager is initialized and sets up logging for the operation.

If the input file path starts with a specific prefix, it is converted to an absolute path. The function validates the file type to ensure it is a YAML file and parses its content to create a Model instance.

Arguments:

• model_file - The path to the metadata YAML file. It can be relative or absolute. If it starts with 'Files/', it is converted to the absolute path in the lakehouse directory.
• config_manager - An instance of ConfigManager that is responsible for initialization and managing configurations.

Returns:

• Model - The Model object created from the provided YAML file.

Raises:

• Exception - If the configuration manager is not initialized.
• Exception - If the provided file is not a YAML file.
• Exception - If the stop-at-error policy is enabled in the configuration manager and an error occurs during the process.

get_table_config_by_name

def get_table_config_by_name(
        table_name: str,
        config_manager: ConfigManager = None) -> Optional[TableConfig]

Retrieves the TableConfig for the given object name by searching in Files/Objects.

Arguments:

• table_name str - The name of the table (dataplatformobjectname).
• config_manager ConfigManager - An instance of ConfigManager.

Returns:

• Optional[TableConfig] - The matching TableConfig if found, otherwise None.

param_clean

def param_clean(value: Any) -> Any

One function. One truth. One source of sanity.

You call it like this in the child notebook: folder_path = fix_param(dbutils.widgets.get("folder_path")) layers = fix_param(dbutils.widgets.get("layers"))

Works perfectly when parent used json.dumps() on everything.