BaseModel

laktory.models.BaseModel

Parent class for all Laktory models offering generic functions and properties. This BaseModel class is derived from pydantic.BaseModel.

ATTRIBUTE	DESCRIPTION
variables	Variable values to be resolved when using inject_vars method. TYPE: dict[str, Any]

METHOD	DESCRIPTION
model_validate_yaml	Load model from yaml file object. Other yaml files can be referenced
model_validate_json_file	Load model from json file object
push_vars	Push variable values to all child recursively
inject_vars	Inject variables values into a model attributes.
inject_vars_into_dump	Inject variables values into a model dump.

Functions

model_validate_yaml classmethod

model_validate_yaml(fp)

Load model from yaml file object. Other yaml files can be referenced
using the ${include.other_yaml_filepath} syntax. You can also merge
lists with -< ${include.other_yaml_filepath} and dictionaries with
<<: ${include.other_yaml_filepath}. Including multi-lines and
commented SQL files is also possible.

Parameters

fp:
    file object structured as a yaml file

Returns

:
    Model instance

Examples:

businesses:
  apple:
    symbol: aapl
    address: ${include.addresses.yaml}
    <<: ${include.common.yaml}
    emails:
      - jane.doe@apple.com
      -< ${include.emails.yaml}
  amazon:
    symbol: amzn
    address: ${include.addresses.yaml}
    <<: ${include.common.yaml}
    emails:
      - john.doe@amazon.com
      -< ${include.emails.yaml}

Source code in laktory/models/basemodel.py

@classmethod
def model_validate_yaml(cls, fp: TextIO) -> Model:
    """
        Load model from yaml file object. Other yaml files can be referenced
        using the ${include.other_yaml_filepath} syntax. You can also merge
        lists with -< ${include.other_yaml_filepath} and dictionaries with
        <<: ${include.other_yaml_filepath}. Including multi-lines and
        commented SQL files is also possible.

        Parameters
        ----------
        fp:
            file object structured as a yaml file

        Returns
        -------
        :
            Model instance

    Examples
    --------
    ```yaml
    businesses:
      apple:
        symbol: aapl
        address: ${include.addresses.yaml}
        <<: ${include.common.yaml}
        emails:
          - jane.doe@apple.com
          -< ${include.emails.yaml}
      amazon:
        symbol: amzn
        address: ${include.addresses.yaml}
        <<: ${include.common.yaml}
        emails:
          - john.doe@amazon.com
          -< ${include.emails.yaml}
    ```
    """

    if hasattr(fp, "name"):
        dirpath = os.path.dirname(fp.name)
    else:
        dirpath = "./"

    def inject_includes(lines):
        _lines = []
        for line in lines:
            line = line.replace("\n", "")
            indent = " " * (len(line) - len(line.lstrip()))
            if line.strip().startswith("#"):
                continue

            if "${include." in line:
                pattern = r"\{include\.(.*?)\}"
                matches = re.findall(pattern, line)
                path = matches[0]
                path0 = path
                if not os.path.isabs(path):
                    path = os.path.join(dirpath, path)
                path_ext = path.split(".")[-1]
                if path_ext not in ["yaml", "yml", "sql"]:
                    raise ValueError(
                        f"Include file of format {path_ext} ({path}) is not supported."
                    )

                # Merge include
                if "<<: ${include." in line or "-< ${include." in line:
                    with open(path, "r", encoding="utf-8") as _fp:
                        new_lines = _fp.readlines()
                        _lines += [
                            indent + __line for __line in inject_includes(new_lines)
                        ]

                # Direct Include
                else:
                    if path.endswith(".sql"):
                        with open(path, "r", encoding="utf-8") as _fp:
                            new_lines = _fp.read()
                        _lines += [
                            line.replace(
                                "${include." + path0 + "}",
                                '"' + new_lines.replace("\n", "\\n") + '"',
                            )
                        ]

                    elif path.endswith(".yaml") or path.endswith("yml"):
                        indent = indent + " " * 2
                        _lines += [line.split("${include")[0]]
                        with open(path, "r", encoding="utf-8") as _fp:
                            new_lines = _fp.readlines()
                        _lines += [
                            indent + __line for __line in inject_includes(new_lines)
                        ]

            else:
                _lines += [line]

        return _lines

    lines = inject_includes(fp.readlines())
    data = yaml.safe_load("\n".join(lines))

    return cls.model_validate(data)

model_validate_json_file classmethod

model_validate_json_file(fp)

Load model from json file object

PARAMETER	DESCRIPTION
fp	file object structured as a json file TYPE: TextIO

RETURNS	DESCRIPTION
Model	Model instance

Source code in laktory/models/basemodel.py

@classmethod
def model_validate_json_file(cls, fp: TextIO) -> Model:
    """
    Load model from json file object

    Parameters
    ----------
    fp:
        file object structured as a json file

    Returns
    -------
    :
        Model instance
    """
    data = json.load(fp)
    return cls.model_validate(data)

push_vars

push_vars(update_core_resources=False)

Push variable values to all child recursively

Source code in laktory/models/basemodel.py

def push_vars(self, update_core_resources=False) -> Any:
    """Push variable values to all child recursively"""

    def _update_model(m):
        if not isinstance(m, BaseModel):
            return
        for k, v in self.variables.items():
            m.variables[k] = m.variables.get(k, v)
        m.push_vars()

    def _push_vars(o):
        if isinstance(o, list):
            for _o in o:
                _push_vars(_o)
        elif isinstance(o, dict):
            for _o in o.values():
                _push_vars(_o)
        else:
            _update_model(o)

    for k in self.model_fields.keys():
        _push_vars(getattr(self, k))

    if update_core_resources and hasattr(self, "core_resources"):
        for r in self.core_resources:
            if r != self:
                _push_vars(r)

    return None

inject_vars

inject_vars(inplace=False, vars=None)

Inject variables values into a model attributes.

There are 2 types of variables:

• User defined variables expressed as ${vars.variable_name} and defined in self.variables (pulled from stack variables) or as environment variables. Stack variables have priority over environment variables.
• Resources output properties expressed as ${resources.resource_name.output}.

PARAMETER	DESCRIPTION
inplace	If True model is modified in place. Otherwise, a new model instance is returned. TYPE: bool DEFAULT: False
vars	A dictionary of variables to be injected in addition to the model internal variables. TYPE: dict DEFAULT: None

RETURNS	DESCRIPTION
	Model instance.

Source code in laktory/models/basemodel.py

def inject_vars(self, inplace: bool = False, vars: dict = None):
    """
    Inject variables values into a model attributes.

    There are 2 types of variables:

    - User defined variables expressed as `${vars.variable_name}` and
      defined in `self.variables` (pulled from stack variables) or as
      environment variables. Stack variables have priority over environment
      variables.
    - Resources output properties expressed as
     `${resources.resource_name.output}`.

    Parameters
    ----------
    inplace:
        If `True` model is modified in place. Otherwise, a new model
        instance is returned.
    vars:
        A dictionary of variables to be injected in addition to the
        model internal variables.


    Returns
    -------
    :
        Model instance.
    """

    # Setting vars
    if vars is None:
        vars = {}
    for k, v in self.variables.items():
        vars[k] = v

    # Create copy
    if not inplace:
        self = self.model_copy(deep=True)

    # Inject into field values
    for k in self.model_fields_set:
        o = getattr(self, k)
        if isinstance(o, BaseModel) or isinstance(o, dict) or isinstance(o, list):
            self._inject_vars(o, vars)
        else:
            setattr(self, k, self._replace(o, vars))

    if not inplace:
        return self

inject_vars_into_dump

inject_vars_into_dump(dump, inplace=False, vars=None)

Inject variables values into a model dump.

There are 2 types of variables:

• User defined variables expressed as ${vars.variable_name} and defined in self.variables (pulled from stack variables) or as environment variables. Stack variables have priority over environment variables.
• Resources output properties expressed as ${resources.resource_name.output}.

PARAMETER	DESCRIPTION
dump	Model dump (or any other general purpose dictionary) TYPE: dict[str, Any]
inplace	If True model is modified in place. Otherwise, a new model instance is returned. TYPE: bool DEFAULT: False
vars	A dictionary of variables to be injected in addition to the model internal variables. TYPE: dict[str, Any] DEFAULT: None

RETURNS	DESCRIPTION
	Model dump with injected variables.

Source code in laktory/models/basemodel.py

def inject_vars_into_dump(
    self, dump: dict[str, Any], inplace: bool = False, vars: dict[str, Any] = None
):
    """
    Inject variables values into a model dump.

    There are 2 types of variables:

    - User defined variables expressed as `${vars.variable_name}` and
      defined in `self.variables` (pulled from stack variables) or as
      environment variables. Stack variables have priority over environment
      variables.
    - Resources output properties expressed as
     `${resources.resource_name.output}`.

    Parameters
    ----------
    dump:
        Model dump (or any other general purpose dictionary)
    inplace:
        If `True` model is modified in place. Otherwise, a new model
        instance is returned.
    vars:
        A dictionary of variables to be injected in addition to the
        model internal variables.


    Returns
    -------
    :
        Model dump with injected variables.
    """

    # Setting vars
    if vars is None:
        vars = {}
    for k, v in self.variables.items():
        vars[k] = v
    _vars = self._get_patterns(vars)

    # Create copy
    if not inplace:
        dump = copy.deepcopy(dump)

    # Inject into field values
    self._inject_vars(dump, vars)

    if not inplace:
        return dump