DataFrameDataSource

laktory.models.datasources.DataFrameDataSource

Bases: BaseDataSource

Data source using in-memory DataFrame.

PARAMETER	DESCRIPTION
dataframe_backend	Type of DataFrame backend TYPE: DataFrameBackends 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
variables	Dict of variables to be injected in the model at runtime TYPE: dict[str, Any] DEFAULT: {}
as_stream	If Truesource is read as a streaming DataFrame. Currently only supported by Spark DataFrame backend. TYPE: bool | VariableType DEFAULT: False
drop_duplicates	Remove duplicated rows from source using all columns if True or only the provided column names. TYPE: bool | list[str] | VariableType DEFAULT: None
drops	List of columns to drop TYPE: list | VariableType DEFAULT: None
filter	SQL expression used to select specific rows from the source table TYPE: str | VariableType DEFAULT: None
renames	Mapping between the source column names and desired column names TYPE: dict[Union[str, VariableType], Union[str, VariableType]] | VariableType DEFAULT: None
selects	Columns to select from the source. Can be specified as a list or as a dictionary to rename the source columns TYPE: list[str] | dict[str, str] | VariableType DEFAULT: None
type	Source Type TYPE: Literal['DATAFRAME'] | VariableType DEFAULT: 'DATAFRAME'
data	Serialized data used to build source TYPE: dict[str, list[Any]] | list[dict[str, Any]] | VariableType DEFAULT: None
df	DataFrame object acting as source TYPE: Any DEFAULT: None

Examples:

From data with PySpark backend.

import laktory as lk

data = {
    "x": [0, 1],
    "y": ["a", "b"],
}

# From data using PySpark
source = lk.models.DataFrameDataSource(
    data=data,
    dataframe_backend="PYSPARK",
)
df = source.read()
print(df.collect(backend="pandas"))
'''
┌──────────────────┐
|Narwhals DataFrame|
|------------------|
|        x  y      |
|     0  0  a      |
|     1  1  b      |
└──────────────────┘
'''

From Polars DataFrame

import polars as pl

import laktory as lk

data = {
    "x": [0, 1],
    "y": ["a", "b"],
}

source = lk.models.DataFrameDataSource(
    df=pl.DataFrame(data),
)
df = source.read()
print(df)
'''
┌──────────────────┐
|Narwhals DataFrame|
|------------------|
|    | x | y |     |
|    |---|---|     |
|    | 0 | a |     |
|    | 1 | b |     |
└──────────────────┘
'''

References

---

• Data Sources and Sinks

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
read	Read data with options specified in attributes.
validate_assignment_disabled	Updating a model attribute inside a model validator when validate_assignment

inject_vars(inplace=False, vars=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

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):
    """
    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.


    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)

    # 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)
        else:
            # Simple objects must be updated explicitly
            setattr(self, k, _resolve_value(o, vars))

    # 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)

    if not inplace:
        return self

inject_vars_into_dump(dump, inplace=False, vars=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

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
):
    """
    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.


    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)

    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) 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

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) -> 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

    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)
    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

read(**kwargs)

Read data with options specified in attributes.

RETURNS	DESCRIPTION
AnyFrame	Resulting dataframe

Source code in laktory/models/datasources/basedatasource.py

def read(self, **kwargs) -> AnyFrame:
    """
    Read data with options specified in attributes.

    Returns
    -------
    :
        Resulting dataframe
    """
    logger.info(
        f"Reading `{self.__class__.__name__}` {self._id} with {self.df_backend}"
    )
    df = self._read(**kwargs)

    # Convert to Narwhals
    if not isinstance(df, (nw.LazyFrame, nw.DataFrame)):
        df = nw.from_native(df)

    # Post read
    df = self._post_read(df)

    logger.info("Read completed.")

    return df

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