PipelineTask

laktory.models.pipeline.PipelineTask

Bases: BaseModel

A pipeline task is a unit of execution within a pipeline, defined by a set of nodes to be executed together.

PARAMETER	DESCRIPTION
name	Pipeline task name TYPE: str | VariableType
node_names	List of node names in sorted order of execution TYPE: list[str | VariableType] | VariableType
pipeline	Pipeline TYPE: Annotated[Pipeline, SkipValidation] | VariableType

METHOD	DESCRIPTION
execute	Execute the pipeline task.

ATTRIBUTE	DESCRIPTION
has_sinks	True if at least one sink is found in task nodes. TYPE: bool
is_sql_expressible	True if all nodes in the task can run on a SQL warehouse. TYPE: bool
nodes	Task nodes
upstream_task_names	Get upstream task names TYPE: list[str]

has_sinks property

True if at least one sink is found in task nodes.

is_sql_expressible property

True if all nodes in the task can run on a SQL warehouse.

nodes property

Task nodes

upstream_task_names property

Get upstream task names

execute(write_sinks=True, refresh='INCREMENTAL', named_dfs=None, update_tables_metadata=True, reset_mode=None)

Execute the pipeline task.

PARAMETER	DESCRIPTION
write_sinks	If False writing of node sinks will be skipped DEFAULT: True
refresh	What the run does: INCREMENTAL, FULL or RESET. See Pipeline.execute(). TYPE: str DEFAULT: 'INCREMENTAL'
named_dfs	Named DataFrames to be passed to pipeline nodes transformer. TYPE: dict[str, AnyFrame] DEFAULT: None
update_tables_metadata	Update tables metadata TYPE: bool DEFAULT: True
reset_mode	Optional override for sinks reset_mode with refresh FULL or RESET. TYPE: str | None DEFAULT: None

Source code in laktory/models/pipeline/pipelinetask.py

def execute(
    self,
    write_sinks=True,
    refresh: str = "INCREMENTAL",
    named_dfs: dict[str, AnyFrame] = None,
    update_tables_metadata: bool = True,
    reset_mode: str | None = None,
) -> None:
    """
    Execute the pipeline task.

    Parameters
    ----------
    write_sinks:
        If `False` writing of node sinks will be skipped
    refresh:
        What the run does: `INCREMENTAL`, `FULL` or `RESET`. See
        `Pipeline.execute()`.
    named_dfs:
        Named DataFrames to be passed to pipeline nodes transformer.
    update_tables_metadata:
        Update tables metadata
    reset_mode:
        Optional override for sinks `reset_mode` with `refresh` `FULL` or `RESET`.
    """

    logger.info(f"Executing pipeline task '{self.name}'")

    # Execute nodes
    for node_name in self.node_names:
        node = self.pipeline.nodes_dict[node_name]
        if named_dfs is None:
            named_dfs = {}

        node.execute(
            write_sinks=write_sinks,
            refresh=refresh,
            named_dfs=named_dfs,
            update_tables_metadata=update_tables_metadata,
            reset_mode=reset_mode,
        )