ConfigLoader

Reference for the ConfigLoader class, which contains the logic used for parsing a pipeline configuration file into a format that can be understood by the Runner.

aineko.ConfigLoader

ConfigLoader(pipeline_config_file: str)

Class to read yaml config files.

Parameters:

Name	Type	Description	Default
pipeline_config_file	str	path of pipeline config file. Defaults to DEFAULT_PIPELINE_CONFIG.	required

Attributes:

Name	Type	Description
pipeline_config_file	str	path to the pipeline configuration file

Methods:

Name	Description
load_config	loads and validates the pipeline config from a yaml file
inject_env_vars	injects environment variables into node params

Initialize ConfigLoader.

Source code in aineko/core/config_loader.py

def __init__(
    self,
    pipeline_config_file: str,
):
    """Initialize ConfigLoader."""
    self.pipeline_config_file = pipeline_config_file or AINEKO_CONFIG.get(
        "DEFAULT_PIPELINE_CONFIG"
    )

pipeline_config_file instance-attribute

pipeline_config_file = pipeline_config_file or get(
    "DEFAULT_PIPELINE_CONFIG"
)

inject_env_vars

inject_env_vars(
    node_params: Optional[
        Union[Dict, List, str, int, float, bool]
    ] = None
) -> Optional[Union[Dict, List, str, int, float, bool]]

Inject environment variables into node params.

This function is used to recursively inject environment variables into strings passed through node params via the pipeline config. We only recursively parse strings, dicts, and lists, as these are the only types that can contain environment variables (i.e. excluding ints, floats, and Nones).

Environment variables are identified in strings by the pattern {$ENV_VAR} where ENV_VAR is the name of the environment variable to inject. For example, given the following environment variables:

$ export SECRET1=secret1
$ export SECRET2=secret2

The following node params dict:

```
{
    "key1": "A string with a {$SECRET1} and a {$SECRET2}.",
    "key2": {
        "key3": "A string with a {$SECRET1} and a {$SECRET2}.",
        "key4": [
            "A string with a {$SECRET1} and a {$SECRET2}.",
            "A string with a {$SECRET1} and a {$SECRET2}."
        ]
    }
}
```

Will be transformed to:

    ```
    {
        "key1": "A string with a secret1 and a secret2.",
        "key2": {
            "key3": "A string with a secret1 and a secret2.",
            "key4": [
                "A string with a secret1 and a secret2.",
                "A string with a secret1 and a secret2."
            ]
        }
    }
    ```

Source code in aineko/core/config_loader.py

def inject_env_vars(
    self,
    node_params: Optional[Union[Dict, List, str, int, float, bool]] = None,
) -> Optional[Union[Dict, List, str, int, float, bool]]:
    """Inject environment variables into node params.

    This function is used to recursively inject environment variables
    into strings passed through node params via the pipeline config.
    We only recursively parse strings, dicts, and lists, as these are
    the only types that can contain environment variables (i.e.
    excluding ints, floats, and Nones).

    Environment variables are identified in strings by the pattern
    {$ENV_VAR} where ENV_VAR is the name of the environment variable
    to inject. For example, given the following environment variables:

    ```
    $ export SECRET1=secret1
    $ export SECRET2=secret2
    ```

    The following node params dict:

        ```
        {
            "key1": "A string with a {$SECRET1} and a {$SECRET2}.",
            "key2": {
                "key3": "A string with a {$SECRET1} and a {$SECRET2}.",
                "key4": [
                    "A string with a {$SECRET1} and a {$SECRET2}.",
                    "A string with a {$SECRET1} and a {$SECRET2}."
                ]
            }
        }
        ```

    Will be transformed to:

            ```
            {
                "key1": "A string with a secret1 and a secret2.",
                "key2": {
                    "key3": "A string with a secret1 and a secret2.",
                    "key4": [
                        "A string with a secret1 and a secret2.",
                        "A string with a secret1 and a secret2."
                    ]
                }
            }
            ```
    """
    if isinstance(node_params, dict):
        for k, v in list(node_params.items()):
            node_params[k] = self.inject_env_vars(v)
    elif isinstance(node_params, list):
        for i, v in enumerate(node_params):
            node_params[i] = self.inject_env_vars(v)
    elif isinstance(node_params, str):
        env_var_pattern = r"\{\$.*?\}"
        env_var_match = re.search(env_var_pattern, node_params, re.DOTALL)
        if env_var_match:
            env_var_env_str = env_var_match.group()
            env_var_value = os.getenv(
                env_var_env_str[2:][:-1], default=None
            )
            if env_var_value is None:
                raise ValueError(
                    "Failed to inject environment variable. "
                    f"{env_var_env_str[2:][:-1]} was not found."
                )
            node_params = node_params.replace(
                env_var_env_str, env_var_value
            )
            return self.inject_env_vars(node_params)

    return node_params

load_config

load_config() -> Config

Load and validate the pipeline config.

Raises:

Type	Description
ValidationError	If the config does not match the schema

Returns:

Type	Description
Config	The validated pipeline config as a Pydantic Config object

Source code in aineko/core/config_loader.py

def load_config(self) -> Config:
    """Load and validate the pipeline config.

    Raises:
        ValidationError: If the config does not match the schema

    Returns:
        The validated pipeline config as a Pydantic Config object
    """
    raw_config = load_yaml(self.pipeline_config_file)

    try:
        config = Config(**raw_config)
    except ValidationError as e:
        logger.error(
            "Schema validation failed for pipeline `%s` loaded from %s. "
            "See detailed error below.",
            raw_config["pipeline"]["name"],
            self.pipeline_config_file,
        )
        raise e

    # Inject environment variables into node params
    for node in config.pipeline.nodes.values():
        if node.node_params is not None:
            node.node_params = self.inject_env_vars(node.node_params)

    return config