BaseModel

laktory.models.BaseModel

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

PARAMETER	DESCRIPTION
variables	Dict of variables to be injected in the model at runtime TYPE: dict[str, Any] DEFAULT: {}

METHOD	DESCRIPTION
inject_vars	Inject model variables values into a model attributes.
inject_vars_into_dump	Inject model variables values into a model dump.
model_validate_json_file	Load model from json file object
model_validate_yaml	Load model from yaml file object using laktory.yaml.RecursiveLoader. Supports
push_vars	Push variable values to all child recursively
resolve_string	Resolve ${vars.x} / ${{ expr }} placeholders in an arbitrary

inject_vars(inplace=False, vars=None, objs=None)

Inject model variables values into a model attributes.

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
objs	A dictionary of objects available when resolving expressions. TYPE: dict DEFAULT: None

RETURNS	DESCRIPTION
	Model instance.

Examples:

from __future__ import annotations

from laktory import models


class Cluster(models.BaseModel):
    name: str = None
    size: int | str = None


c = Cluster(
    name="cluster-${vars.my_cluster}",
    size="${{ 4 if vars.env == 'prod' else 2 }}",
    variables={
        "env": "dev",
    },
).inject_vars()
print(c)
# > variables={'env': 'dev'} name='cluster-${vars.my_cluster}' size=2

References

• variables

Source code in laktory/models/basemodel.py

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

    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.
    objs:
        A dictionary of objects available when resolving expressions.


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

    Examples
    --------
    ```py
    from __future__ import annotations

    from laktory import models


    class Cluster(models.BaseModel):
        name: str = None
        size: int | str = None


    c = Cluster(
        name="cluster-${vars.my_cluster}",
        size="${{ 4 if vars.env == 'prod' else 2 }}",
        variables={
            "env": "dev",
        },
    ).inject_vars()
    print(c)
    # > variables={'env': 'dev'} name='cluster-${vars.my_cluster}' size=2
    ```

    References
    ----------
    * [variables](https://www.laktory.ai/concepts/variables/)
    """

    # Fetching vars
    # Merge first, deep-copy once: _resolve_values mutates mutable values
    # (list/dict) in-place when resolving nested variable references, so both
    # the caller's vars and self.variables values must be protected. Two
    # separate deepcopies (caller then update) would leave self.variables
    # unprotected; merging into a new dict first and copying once covers both.
    vars = deepcopy({**(vars or {}), **self.variables})

    # Fetching objs - subclasses override _inject_vars_objs() to inject
    # context objects (e.g. pipeline, pipeline_node) without circular imports
    _caller_objs = objs
    if objs is None:
        objs = self._inject_vars_objs()
    else:
        objs = {**self._inject_vars_objs(), **objs}

    # Cache check: skip re-resolution when vars and objs haven't changed
    cache_key = None
    if not inplace and _caller_objs is None:
        cache_key = json.dumps(vars, sort_keys=True)
        if self._inject_vars_cache_key == cache_key:
            return self._inject_vars_cache_value.model_copy(deep=True)

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

    # Inject into field values
    for k in list(self.model_fields_set):
        if k == "variables":
            continue
        # Frozen fields (e.g. `type` literals) are constants and can
        # never contain a variable to resolve. Skipping them also
        # avoids Pydantic's frozen-field check on setattr below.
        field = type(self).model_fields.get(k)
        if field is not None and field.frozen:
            continue
        o = getattr(self, k)

        if isinstance(o, BaseModel) or isinstance(o, dict) or isinstance(o, list):
            # Mutable objects will be updated in place
            _resolve_values(o, vars, objs)
        else:
            # Simple objects must be updated explicitly, but only if
            # the resolved value actually changed
            new_o = _resolve_value(o, vars, objs)
            if new_o != o:
                setattr(self, k, new_o)

    # Inject into child resources
    if hasattr(self, "core_resources"):
        for r in self.core_resources:
            if r == self:
                continue
            r.inject_vars(vars=vars, inplace=True, objs=objs)

    if not inplace:
        if cache_key is not None:
            original._inject_vars_cache_key = cache_key
            original._inject_vars_cache_value = self
        return self

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

Inject model variables values into a model dump.

PARAMETER	DESCRIPTION
dump	Model dump (or any other general purpose mutable object) 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
objs	A dictionary of objects available when resolving expressions. TYPE: dict[str, Any] DEFAULT: None

RETURNS	DESCRIPTION
	Model dump with injected variables.

Examples:

from laktory import models

m = models.BaseModel(
    variables={
        "env": "dev",
    },
)
data = {
    "name": "cluster-${vars.my_cluster}",
    "size": "${{ 4 if vars.env == 'prod' else 2 }}",
}
print(m.inject_vars_into_dump(data))
# > {'name': 'cluster-${vars.my_cluster}', 'size': 2}

References

• 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,
    objs: dict[str, Any] = None,
):
    """
    Inject model variables values into a model dump.

    Parameters
    ----------
    dump:
        Model dump (or any other general purpose mutable object)
    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.
    objs:
        A dictionary of objects available when resolving expressions.


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


    Examples
    --------
    ```py
    from laktory import models

    m = models.BaseModel(
        variables={
            "env": "dev",
        },
    )
    data = {
        "name": "cluster-${vars.my_cluster}",
        "size": "${{ 4 if vars.env == 'prod' else 2 }}",
    }
    print(m.inject_vars_into_dump(data))
    # > {'name': 'cluster-${vars.my_cluster}', 'size': 2}
    ```

    References
    ----------
    * [variables](https://www.laktory.ai/concepts/variables/)
    """

    # Setting vars - same merge-then-copy pattern as inject_vars()
    vars = deepcopy({**(vars or {}), **self.variables})

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

    # Inject into field values
    _resolve_values(dump, vars, objs)

    if not inplace:
        return dump

model_validate_json_file(fp) classmethod

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: typing.Type[Model], 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)

model_validate_yaml(fp, vars=None) classmethod

Load model from yaml file object using laktory.yaml.RecursiveLoader. Supports reference to external yaml and sql files using !use, !extend and !update tags. Path to external files can be defined using model or environment variables.

Referenced path should always be relative to the file they are referenced from.

Custom Tags

• !use {filepath}: Directly inject the content of the file at filepath

• - !extend {filepath}: Extend the current list with the elements found in the file at filepath. Similar to python list.extend method.

• <<: !update {filepath}: Merge the current dictionary with the content of the dictionary defined at filepath. Similar to python dict.update method.

PARAMETER	DESCRIPTION
fp	file object structured as a yaml file TYPE: TextIO
vars	Dict of variables available when parsing filepaths references in yaml files i.e. !use catalog_${vars.env}.yaml DEFAULT: None

RETURNS	DESCRIPTION
Model	Model instance

Examples:

businesses:
  apple:
    symbol: aapl
    address: !use addresses.yaml
    <<: !update common.yaml
    emails:
      - jane.doe@apple.com
      - extend! emails.yaml
  amazon:
    symbol: amzn
    address: !use addresses.yaml
    <<: update! common.yaml
    emails:
      - john.doe@amazon.com
      - extend! emails.yaml

Source code in laktory/models/basemodel.py

@classmethod
def model_validate_yaml(cls: typing.Type[Model], fp: TextIO, vars=None) -> Model:
    """
    Load model from yaml file object using laktory.yaml.RecursiveLoader. Supports
    reference to external yaml and sql files using `!use`, `!extend` and `!update` tags.
    Path to external files can be defined using model or environment variables.

    Referenced path should always be relative to the file they are referenced from.

    Custom Tags
    -----------
    - `!use {filepath}`:
        Directly inject the content of the file at `filepath`

    - `- !extend {filepath}`:
        Extend the current list with the elements found in the file at `filepath`.
        Similar to python list.extend method.

    - `<<: !update {filepath}`:
        Merge the current dictionary with the content of the dictionary defined at
        `filepath`. Similar to python dict.update method.

    Parameters
    ----------
    fp:
        file object structured as a yaml file
    vars:
        Dict of variables available when parsing filepaths references in yaml files
        i.e. `!use catalog_${vars.env}.yaml`

    Returns
    -------
    :
        Model instance

    Examples
    --------
    ```yaml
    businesses:
      apple:
        symbol: aapl
        address: !use addresses.yaml
        <<: !update common.yaml
        emails:
          - jane.doe@apple.com
          - extend! emails.yaml
      amazon:
        symbol: amzn
        address: !use addresses.yaml
        <<: update! common.yaml
        emails:
          - john.doe@amazon.com
          - extend! emails.yaml
    ```
    """

    data = RecursiveLoader.load(fp, vars=vars)
    return cls.model_validate(data)

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 type(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

resolve_string(text, vars=None, objs=None)

Resolve ${vars.x} / ${{ expr }} placeholders in an arbitrary string (e.g. raw file content that is not itself a model field) using this model's variables merged with any additional vars/objs.

Unlike inject_vars/inject_vars_into_dump on a typed field - where a placeholder resolving to a non-string (dict, list, bool, ...) replaces the whole field value with that Python object - a non-string resolution here is JSON-serialized and substituted in place, since the return value must always be text. JSON is valid embedded syntax for both JSON and YAML content; other file types may need the value pre-formatted as a string instead (e.g. via a ${{ }} expression).

PARAMETER	DESCRIPTION
text	Raw string to resolve. TYPE: str
vars	Additional variables to merge with self.variables (self.variables wins on conflict, same precedence as inject_vars/ inject_vars_into_dump). TYPE: dict[str, Any] DEFAULT: None
objs	A dictionary of objects available when resolving expressions. TYPE: dict[str, Any] DEFAULT: None

RETURNS	DESCRIPTION
str	Resolved string.

Examples:

from laktory import models

m = models.BaseModel(
    variables={
        "env": "dev",
    },
)
print(m.resolve_string("catalog: ${vars.env}"))
# > catalog: dev

A variable resolving to a dict or list is JSON-serialized in place:

from laktory import models

m = models.BaseModel(
    variables={
        "tags": {"bu": "finance", "env": "dev"},
    },
)
print(m.resolve_string('{"tags": ${vars.tags}}'))
# > {"tags": {"bu": "finance", "env": "dev"}}

Source code in laktory/models/basemodel.py

def resolve_string(
    self, text: str, vars: dict[str, Any] = None, objs: dict[str, Any] = None
) -> str:
    """
    Resolve `${vars.x}` / `${{ expr }}` placeholders in an arbitrary
    string (e.g. raw file content that is not itself a model field)
    using this model's `variables` merged with any additional
    `vars`/`objs`.

    Unlike `inject_vars`/`inject_vars_into_dump` on a typed field -
    where a placeholder resolving to a non-string (dict, list, bool,
    ...) replaces the whole field value with that Python object - a
    non-string resolution here is JSON-serialized and substituted in
    place, since the return value must always be text. JSON is valid
    embedded syntax for both JSON and YAML content; other file types
    may need the value pre-formatted as a string instead (e.g. via a
    `${{ }}` expression).

    Parameters
    ----------
    text:
        Raw string to resolve.
    vars:
        Additional variables to merge with `self.variables` (`self.variables`
        wins on conflict, same precedence as `inject_vars`/
        `inject_vars_into_dump`).
    objs:
        A dictionary of objects available when resolving expressions.

    Returns
    -------
    :
        Resolved string.

    Examples
    --------
    ```py
    from laktory import models

    m = models.BaseModel(
        variables={
            "env": "dev",
        },
    )
    print(m.resolve_string("catalog: ${vars.env}"))
    # > catalog: dev
    ```

    A variable resolving to a dict or list is JSON-serialized in place:

    ```py
    from laktory import models

    m = models.BaseModel(
        variables={
            "tags": {"bu": "finance", "env": "dev"},
        },
    )
    print(m.resolve_string('{"tags": ${vars.tags}}'))
    # > {"tags": {"bu": "finance", "env": "dev"}}
    ```
    """
    vars = deepcopy({**(vars or {}), **self.variables})
    return _resolve_value(text, vars, objs, stringify=True)