Skip to content

read_instance_mixin

ReadInstancesMixin

Bases: BaseModel

This class defines the methods to read, validate and parse metadata definitions.

Source code in src/cloe_nessy/models/mixins/read_instance_mixin.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class ReadInstancesMixin(BaseModel):
    """This class defines the methods to read, validate and parse metadata definitions."""

    @classmethod
    def metadata_to_instance(cls, data: dict) -> tuple[Self | None, list[ValidationError]]:
        """Parses a Dictionary to an instance.

        Returns:
            An instance and potentially a list of errors.
        """
        errors = []
        try:
            instance = cls(**data)
        except ValidationError as e:
            instance = None
            errors.append(e)
        return instance, errors

    @classmethod
    def read_instance_from_file(
        cls,
        instance_path: pathlib.Path,
        **_: Any,  # allow subclasses to pass additional arguments
    ) -> tuple[Self | None, list[ValidationErrorType]]:
        """Read and instantiate a single YAML file for the given path.

        Arguments:
            instance_path: The path to the file to instantiate.

        Return:
            Returns a tuple of the instantiated model and errors.
        """
        errors: list[ValidationErrorType] = []
        try:
            with instance_path.open("r") as file:
                raw_string = file.read()
                yaml_str = cls._replace_variables(raw_string)
                data = yaml.safe_load(yaml_str)
                instance, sub_errors = cls.metadata_to_instance(data)
                errors += sub_errors
        except (ValidationError, yaml.parser.ParserError, yaml.scanner.ScannerError) as e:
            instance = None
            errors.append(e)
        return instance, errors

    @classmethod
    def read_instances_from_directory(
        cls,
        instance_path: str | pathlib.Path,
        fail_on_missing_subfolder: bool = True,
        **_: Any,  # allow subclasses to pass additional arguments
    ) -> tuple[list[Self], list[ValidationErrorType]]:
        """Read and instantiate all *.yaml files for the given path.

        Arguments:
            instance_path: Path to the directory containing the instance definitions as YAML files.
            fail_on_missing_subfolder: If False return a tuple with 2 empty
                    lists. Otherwise raise a FileNotFoundError.

        Return:
            Returns a tuple of the instantiated models and errors.
        """
        instances: list[Self] = []
        errors: list[ValidationErrorType] = []

        # Convert to Path if string
        if isinstance(instance_path, str):
            instance_path = pathlib.Path(instance_path)

        if not instance_path.exists() or not instance_path.is_dir():
            if fail_on_missing_subfolder:
                raise FileNotFoundError(f"Directory not found: {instance_path}")
            else:
                return instances, errors

        for instance_file in instance_path.iterdir():
            sub_errors: list[ValidationErrorType] = []
            if instance_file.is_file() and instance_file.suffix in (".yaml", ".yml"):
                instance, sub_errors = cls.read_instance_from_file(instance_file)
                instances += [] if instance is None else [instance]
            errors += sub_errors

        return instances, errors

    @staticmethod
    def _replace_variables(yaml_str: str) -> str:
        """Replace variable placeholders in a YAML string.

        Replaces environment variables with the pattern `{{env:var-name}}`. Where
        the var-name is the name of the environment variable.

        Args:
            yaml_str (str): A string that can be parsed in YAML format.

        Returns:
            The same YAML string with environment variable placeholders
            replaced.
        """
        env_var_pattern = r"\{\{env:([^}]+)\}\}"
        secret_ref_pattern = r"\{\{(?!step|env)([^}]+):([^}]+)\}\}"

        def replace_with_env_var(match):
            env_var_name = match.group(1)
            env_var_value = os.getenv(env_var_name)
            return env_var_value

        def replace_with_secret(match):
            secret_scope_name = match.group(1)
            secret_key = match.group(2)
            return SessionManager.get_utils().secrets.get(scope=secret_scope_name, key=secret_key)

        env_replaced_yaml_string = re.sub(env_var_pattern, replace_with_env_var, yaml_str)
        final_yaml_string = re.sub(secret_ref_pattern, replace_with_secret, env_replaced_yaml_string)
        return final_yaml_string

metadata_to_instance(data) classmethod

Parses a Dictionary to an instance.

Returns:

Type Description
tuple[Self | None, list[ValidationError]]

An instance and potentially a list of errors.
Source code in src/cloe_nessy/models/mixins/read_instance_mixin.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@classmethod
def metadata_to_instance(cls, data: dict) -> tuple[Self | None, list[ValidationError]]:
    """Parses a Dictionary to an instance.

    Returns:
        An instance and potentially a list of errors.
    """
    errors = []
    try:
        instance = cls(**data)
    except ValidationError as e:
        instance = None
        errors.append(e)
    return instance, errors

read_instance_from_file(instance_path, **_) classmethod

Read and instantiate a single YAML file for the given path.

Parameters:

Name Type Description Default
instance_path Path

The path to the file to instantiate.

 required
Return

Returns a tuple of the instantiated model and errors.

Source code in src/cloe_nessy/models/mixins/read_instance_mixin.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@classmethod
def read_instance_from_file(
    cls,
    instance_path: pathlib.Path,
    **_: Any,  # allow subclasses to pass additional arguments
) -> tuple[Self | None, list[ValidationErrorType]]:
    """Read and instantiate a single YAML file for the given path.

    Arguments:
        instance_path: The path to the file to instantiate.

    Return:
        Returns a tuple of the instantiated model and errors.
    """
    errors: list[ValidationErrorType] = []
    try:
        with instance_path.open("r") as file:
            raw_string = file.read()
            yaml_str = cls._replace_variables(raw_string)
            data = yaml.safe_load(yaml_str)
            instance, sub_errors = cls.metadata_to_instance(data)
            errors += sub_errors
    except (ValidationError, yaml.parser.ParserError, yaml.scanner.ScannerError) as e:
        instance = None
        errors.append(e)
    return instance, errors

read_instances_from_directory(instance_path, fail_on_missing_subfolder=True, **_) classmethod

Read and instantiate all *.yaml files for the given path.

Parameters:

Name Type Description Default
instance_path str | Path

Path to the directory containing the instance definitions as YAML files.

 required
fail_on_missing_subfolder bool

If False return a tuple with 2 empty lists. Otherwise raise a FileNotFoundError.

 True
Return

Returns a tuple of the instantiated models and errors.

Source code in src/cloe_nessy/models/mixins/read_instance_mixin.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@classmethod
def read_instances_from_directory(
    cls,
    instance_path: str | pathlib.Path,
    fail_on_missing_subfolder: bool = True,
    **_: Any,  # allow subclasses to pass additional arguments
) -> tuple[list[Self], list[ValidationErrorType]]:
    """Read and instantiate all *.yaml files for the given path.

    Arguments:
        instance_path: Path to the directory containing the instance definitions as YAML files.
        fail_on_missing_subfolder: If False return a tuple with 2 empty
                lists. Otherwise raise a FileNotFoundError.

    Return:
        Returns a tuple of the instantiated models and errors.
    """
    instances: list[Self] = []
    errors: list[ValidationErrorType] = []

    # Convert to Path if string
    if isinstance(instance_path, str):
        instance_path = pathlib.Path(instance_path)

    if not instance_path.exists() or not instance_path.is_dir():
        if fail_on_missing_subfolder:
            raise FileNotFoundError(f"Directory not found: {instance_path}")
        else:
            return instances, errors

    for instance_file in instance_path.iterdir():
        sub_errors: list[ValidationErrorType] = []
        if instance_file.is_file() and instance_file.suffix in (".yaml", ".yml"):
            instance, sub_errors = cls.read_instance_from_file(instance_file)
            instances += [] if instance is None else [instance]
        errors += sub_errors

    return instances, errors