FileDataSource

laktory.models.FileDataSource

Bases: BaseDataSource

Data source using disk files, such data events (json/csv) or full dataframes.

Examples:

from laktory import models

source = models.FileDataSource(
    path="/Volumes/sources/landing/events/yahoo-finance/stock_price",
    format="JSON",
    dataframe_backend="POLARS",
)
# df = source.read()

# With Explicit Schema
source = models.FileDataSource(
    path="/Volumes/sources/landing/events/yahoo-finance/stock_price",
    format="JSON",
    dataframe_backend="PYSPARK",
    schema={
        "columns": {
            "symbol": "String",
            "open": "Float64",
            "close": "Float64",
        }
    },
)
# df = source.read()

References

• Data Sources and Sinks
• autoloader schema inference

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
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['FILE'] | VariableType DEFAULT: 'FILE'
format	Format of the data files. TYPE: Literal['AVRO', 'BINARYFILE', 'CSV', 'DELTA', 'EXCEL', 'IPC', 'JSON', 'JSONL', 'NDJSON', 'ORC', 'PARQUET', 'PYARROW', 'TEXT', 'XML'] | VariableType
has_header	Indicate if the first row of the dataset is a header or not. Only applicable to 'CSV' format. TYPE: bool | VariableType DEFAULT: True
infer_schema	When True, the schema is inferred from the data. When False, the schema is not inferred and will be string if not specified in schema_definition. Only applicable to some format like CSV and JSON. TYPE: bool | VariableType DEFAULT: False
path	File path on a local disk, remote storage or Databricks volume. TYPE: str | VariableType
reader_kwargs	Keyword arguments passed directly to dataframe backend reader. Passed to .options() method when using PySpark. TYPE: dict[Union[str, VariableType], Union[Any, VariableType]] | VariableType DEFAULT: {}
schema_definition	Target schema specified as a list of columns, as a dict or a json serialization. Only used when reading data from non-strongly typed files such as JSON or csv files. TYPE: DataFrameSchema | VariableType DEFAULT: None
schema_location_	Path for schema inference when reading data as a stream. If None, parent directory of path is used. TYPE: str | Path | VariableType DEFAULT: None
reader_methods	DataFrame backend reader methods. TYPE: list[Union[ReaderWriterMethod, VariableType]] | VariableType 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
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.dataframe_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