Airflow

laktory.models.pipeline.AirflowOrchestrator

Bases: PipelineChild

Airflow used as an orchestrator to execute a Laktory pipeline.

Examples:

import laktory as lk

o = lk.models.AirflowOrchestrator(
    description="A Laktory pipeline",
    schedule={
        "cron": "0 0 * * *",
        "timezone": "utc",
    },
    start_date="2026-03-03",
    max_active_runs=1,
)
print(o)
'''
dataframe_backend_=None dataframe_api_=None type='AIRFLOW' description='A Laktory pipeline' schedule=CronSchedule(variables={}, cron='0 0 * * *', timezone='utc') start_date=datetime.datetime(2026, 3, 3, 0, 0, tzinfo=datetime.timezone.utc) end_date=None template_searchpath=None user_defined_macros=None user_defined_filters=None default_args=None max_active_tasks=None max_active_runs=1 max_consecutive_failed_dag_runs=None dagrun_timeout=None catchup=None doc_md=None access_control=None is_paused_upon_creation=None jinja_environment_kwargs=None render_template_as_native_obj=None tags=None owner_links=None auto_register=None fail_fast=None dag_display_name=None disable_bundle_versioning=None dataframe_backend=<DataFrameBackends.PYSPARK: 'PYSPARK'> dataframe_api='NARWHALS'
'''

References

• Data Pipeline

PARAMETER	DESCRIPTION
dataframe_backend_	Type of DataFrame backend TYPE: DataFrameBackends | VariableType DEFAULT: None
dataframe_api_	DataFrame API to use in DataFrame Transformer nodes. Either 'NATIVE' (backend-specific) or 'NARWHALS' (backend-agnostic). TYPE: Literal['NARWHALS', 'NATIVE'] | VariableType DEFAULT: None
type	Type of orchestrator TYPE: Literal['AIRFLOW'] | VariableType DEFAULT: 'AIRFLOW'
description	Optional human-readable description of the DAG. TYPE: str | VariableType DEFAULT: None
schedule	Schedule that defines when the DAG should run. Uses a cron expression and timezone. TYPE: CronSchedule | VariableType DEFAULT: None
start_date	The earliest date from which the scheduler will start creating DAG runs. TYPE: datetime | VariableType DEFAULT: None
end_date	Optional date after which no new DAG runs will be scheduled. TYPE: datetime | VariableType DEFAULT: None
template_searchpath	Directories to search for template files referenced in templated fields. TYPE: str | list[str] | VariableType DEFAULT: None
user_defined_macros	Custom macros made available in the Jinja template context for all tasks in the DAG. TYPE: dict | VariableType DEFAULT: None
user_defined_filters	Custom Jinja filters available when rendering templated fields. TYPE: dict | VariableType DEFAULT: None
default_args	Dictionary of default arguments applied to all tasks in the DAG unless overridden. TYPE: dict[str | VariableType, Any | VariableType] | VariableType DEFAULT: None
max_active_tasks	Maximum number of task instances allowed to run concurrently across all DAG runs. TYPE: int | VariableType DEFAULT: None
max_active_runs	Maximum number of active DAG runs allowed at the same time. TYPE: int | VariableType DEFAULT: None
max_consecutive_failed_dag_runs	Number of consecutive failed DAG runs allowed before the DAG is automatically paused. TYPE: int | VariableType DEFAULT: None
dagrun_timeout	Maximum time allowed for a DAG run to complete before it is marked as failed. TYPE: str | timedelta | VariableType DEFAULT: None
catchup	Whether the scheduler should create DAG runs for all past schedule intervals between the start date and the current time. TYPE: bool | VariableType DEFAULT: None
doc_md	Markdown documentation displayed in the Airflow UI for the DAG. TYPE: str | VariableType DEFAULT: None
access_control	Access control configuration defining roles and permissions for this DAG. TYPE: dict[str, dict[str, list[str]]] | dict[str, list[str]] | VariableType DEFAULT: None
is_paused_upon_creation	Whether the DAG should be paused when first created. TYPE: bool | VariableType DEFAULT: None
jinja_environment_kwargs	Additional keyword arguments used when creating the Jinja template environment. TYPE: dict | VariableType DEFAULT: None
render_template_as_native_obj	If true, Jinja templates will render native Python objects instead of strings. TYPE: bool | VariableType DEFAULT: None
tags	Tags used to organize and filter DAGs in the Airflow UI. TYPE: list[str | VariableType] | VariableType DEFAULT: None
owner_links	Mapping of DAG owners to URLs that provide additional information about them. TYPE: dict[str | VariableType, str | VariableType] | VariableType DEFAULT: None
auto_register	Whether the DAG should automatically be registered when parsed by Airflow. TYPE: bool | VariableType DEFAULT: None
fail_fast	If enabled, the DAG run will stop scheduling new tasks as soon as a task fails. TYPE: bool | VariableType DEFAULT: None
dag_display_name	Human-friendly name displayed in the Airflow UI instead of the DAG ID. TYPE: str | VariableType DEFAULT: None
disable_bundle_versioning	If enabled, disables versioning of DAG bundles when deploying. TYPE: bool | VariableType DEFAULT: None

--

laktory.models.pipeline.orchestrators.airfloworchestrator.CronSchedule

Bases: BaseModel

Cron Schedule.

PARAMETER	DESCRIPTION
variables	Dict of variables to be injected in the model at runtime TYPE: dict[str, Any] DEFAULT: {}
cron	Cron expression defining when the DAG should be scheduled to run. TYPE: str | VariableType
timezone	Timezone used to interpret the cron schedule. TYPE: str | VariableType DEFAULT: 'utc'

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
validate_assignment_disabled	Updating a model attribute inside a model validator when validate_assignment

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 typing import Union

from laktory import models


class Cluster(models.BaseModel):
    name: str = None
    size: Union[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 typing import Union

    from laktory import models


    class Cluster(models.BaseModel):
        name: str = None
        size: Union[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
    if vars is None:
        vars = {}

    vars = deepcopy(vars)
    vars.update(self.variables)

    # Fetching objs
    if objs is None:
        objs = {}

    # TODO: Review implementation as it results in serious performance hits
    from laktory.models.pipeline import Pipeline
    from laktory.models.pipeline import PipelineNode

    if isinstance(self, Pipeline):
        objs["pipeline"] = self

    if isinstance(self, PipelineNode):
        objs["pipeline_node"] = self

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

    # Inject into field values
    for k in list(self.model_fields_set):
        if k == "variables":
            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
            setattr(self, k, _resolve_value(o, vars, objs))

    # 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:
        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
    if vars is None:
        vars = {}
    vars = deepcopy(vars)
    vars.update(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: 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: 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 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

validate_assignment_disabled()

Updating a model attribute inside a model validator when validate_assignment is True causes an infinite recursion by design and must be turned off temporarily.

Source code in laktory/models/basemodel.py

@contextmanager
def validate_assignment_disabled(self):
    """
    Updating a model attribute inside a model validator when `validate_assignment`
    is `True` causes an infinite recursion by design and must be turned off
    temporarily.
    """
    original_state = self.model_config["validate_assignment"]
    self.model_config["validate_assignment"] = False
    try:
        yield
    finally:
        self.model_config["validate_assignment"] = original_state