Stack

laktory.models.stacks.Stack

Bases: BaseModel

The Stack defines a collection of deployable resources, the deployment configuration, some variables and the environment-specific settings.

Examples:

from laktory import models

stack = models.Stack(
    name="workspace",
    resources={
        "databricks_pipelines": {
            "pl-stock-prices": {
                "name": "pl-stock-prices",
                "development": "${vars.is_dev}",
                "libraries": [
                    {"notebook": {"path": "/pipelines/dlt_brz_template.py"}},
                ],
            }
        },
        "databricks_jobs": {
            "job-stock-prices": {
                "name": "job-stock-prices",
                "job_clusters": [
                    {
                        "job_cluster_key": "main",
                        "new_cluster": {
                            "spark_version": "16.3.x-scala2.12",
                            "node_type_id": "Standard_DS3_v2",
                        },
                    }
                ],
                "tasks": [
                    {
                        "task_key": "ingest",
                        "job_cluster_key": "main",
                        "notebook_task": {
                            "notebook_path": "/.laktory/jobs/ingest_stock_prices.py",
                        },
                    },
                    {
                        "task_key": "pipeline",
                        "depends_on": [{"task_key": "ingest"}],
                        "pipeline_task": {
                            "pipeline_id": "${resources.dlt-pl-stock-prices.id}",
                        },
                    },
                ],
            }
        },
    },
    variables={
        "org": "okube",
    },
    environments={
        "dev": {
            "variables": {
                "is_dev": True,
            }
        },
        "prod": {
            "variables": {
                "is_dev": False,
            }
        },
    },
)

References

---

• Stack

PARAMETER	DESCRIPTION
description	Description of the stack TYPE: str | VariableType DEFAULT: None
environments	Environment-specific overwrite of config, resources or variables arguments. TYPE: dict[str | VariableType, EnvironmentSettings | VariableType] | VariableType DEFAULT: {}
iac_backend	IaC backend used for deployment. TYPE: Literal['terraform'] | VariableType DEFAULT: 'terraform'
name	Name of the stack. TYPE: str | VariableType
organization	Organization TYPE: str | None | VariableType DEFAULT: None
resources	Dictionary of resources to be deployed. Each key should be a resource type and each value should be a dictionary of resources who's keys are the resource names and the values the resources definitions. TYPE: StackResources | None | VariableType DEFAULT: StackResources(variables={}, databricks_accesscontrolrulesets={}, databricks_alerts={}, databricks_apps={}, databricks_catalogs={}, databricks_clusterpolicies={}, databricks_clusters={}, databricks_connections={}, databricks_currentusers={}, databricks_dashboards={}, databricks_dbfsfiles={}, databricks_directories={}, databricks_entitlements={}, databricks_pipelines={}, databricks_externallocations={}, databricks_grant={}, databricks_grants={}, databricks_groups={}, databricks_jobs={}, databricks_metastoreassignments={}, databricks_metastoredataaccesses={}, databricks_metastores={}, databricks_mlflowexperiments={}, databricks_mlflowmodels={}, databricks_mlflowwebhooks={}, databricks_networkconnectivityconfig={}, databricks_notebooks={}, databricks_notificationdestinations={}, databricks_obotokens={}, databricks_permissions={}, databricks_dataqualitymonitors={}, databricks_pythonpackages={}, databricks_queries={}, databricks_recipients={}, databricks_repos={}, databricks_schemas={}, databricks_secrets={}, databricks_secretscopes={}, databricks_serviceprincipals={}, databricks_shares={}, databricks_storagecredentials={}, databricks_tables={}, databricks_users={}, databricks_vectorsearchendpoints={}, databricks_vectorsearchindexes={}, databricks_volumes={}, databricks_warehouses={}, databricks_workspacebindings={}, databricks_workspacefiles={}, databricks_workspacetrees={}, pipelines={}, providers={})
settings	Laktory settings TYPE: LaktorySettings | VariableType DEFAULT: None
terraform	Terraform-specific settings TYPE: Terraform | VariableType DEFAULT: Terraform(variables={}, backend=None)

METHOD	DESCRIPTION
apply_settings	Required to apply settings before instantiating resources and setting default values
build	Build stack artifacts before preview or deploy.
get_env	Complete definition the stack for a given environment. It takes into
to_terraform	Create a terraform stack for a given environment env.

apply_settings(data) classmethod

Required to apply settings before instantiating resources and setting default values

Source code in laktory/models/stacks/stack.py

@model_validator(mode="before")
@classmethod
def apply_settings(cls, data: Any) -> Any:
    """Required to apply settings before instantiating resources and setting default values"""
    settings = data.get("settings", None)
    if settings:
        if not isinstance(settings, dict):
            settings = settings.model_dump(exclude_unset=True)
        LaktorySettings(**settings)

    return data

build(env_name, inject_vars=True, vars=None)

Build stack artifacts before preview or deploy.

Pipeline config JSON files are written to the location determined by settings.build_root (when set in stack.yaml under settings:) or the default Laktory cache directory. For Databricks Asset Bundles users, set settings.build_root to a project-local path (e.g. .laktory/.resources/) so that DABs can sync the files to the workspace.

PARAMETER	DESCRIPTION
env_name	Name of the environment TYPE: str | None
inject_vars	Inject stack variables TYPE: bool DEFAULT: True
vars	Additional variables that override stack and environment variables. TYPE: dict DEFAULT: None

Source code in laktory/models/stacks/stack.py

def build(self, env_name: str | None, inject_vars: bool = True, vars: dict = None):
    """
    Build stack artifacts before preview or deploy.

    Pipeline config JSON files are written to the location determined by
    ``settings.build_root`` (when set in ``stack.yaml`` under
    ``settings:``) or the default Laktory cache directory. For Databricks
    Asset Bundles users, set ``settings.build_root`` to a
    project-local path (e.g. ``.laktory/.resources/``) so that DABs can
    sync the files to the workspace.

    Parameters
    ----------
    env_name:
        Name of the environment
    inject_vars:
        Inject stack variables
    vars:
        Additional variables that override stack and environment variables.
    """

    logger.info("Building artifacts...")

    env = self.get_env(env_name=env_name)
    if inject_vars:
        if vars:
            env = env.model_copy(update={"variables": {**env.variables, **vars}})
        env = env.inject_vars()

    if env.resources is None:
        return

    for k, r in env.resources._get_all(providers_excluded=True).items():
        if isinstance(r, PythonPackage):
            r.build()

    logger.info("Writing pipeline config files...")
    for k, r in env.resources._get_all(providers_excluded=True).items():
        if isinstance(r, Pipeline):
            orchestrator = r.orchestrator
            if not orchestrator:
                continue

            config_file = getattr(r.orchestrator, "config_file", None)
            if config_file:
                config_file.build()

    logger.info("Build completed.")

get_env(env_name)

Complete definition the stack for a given environment. It takes into account both the default stack values and environment-specific overwrites.

PARAMETER	DESCRIPTION
env_name	Name of the environment TYPE: str | None

RETURNS	DESCRIPTION
	Environment definitions.

Source code in laktory/models/stacks/stack.py

def get_env(self, env_name: str | None):
    """
    Complete definition the stack for a given environment. It takes into
    account both the default stack values and environment-specific
    overwrites.

    Parameters
    ----------
    env_name:
        Name of the environment

    Returns
    -------
    :
        Environment definitions.
    """

    if env_name is None:
        return self

    if env_name not in self.environments.keys():
        raise ValueError(f"Environment '{env_name}' is not declared in the stack.")

    return self._apply_env_overrides(self.environments[env_name])

to_terraform(env_name=None, vars=None)

Create a terraform stack for a given environment env.

PARAMETER	DESCRIPTION
env_name	Target environment. If None, used default stack values only. TYPE: str | None DEFAULT: None
vars	Additional variables that override stack and environment variables. TYPE: dict DEFAULT: None

RETURNS	DESCRIPTION
TerraformStack	Terraform-specific stack definition

Source code in laktory/models/stacks/stack.py

def to_terraform(self, env_name: str | None = None, vars: dict = None):
    """
    Create a terraform stack for a given environment `env`.

    Parameters
    ----------
    env_name:
        Target environment. If `None`, used default stack values only.
    vars:
        Additional variables that override stack and environment variables.

    Returns
    -------
    : TerraformStack
        Terraform-specific stack definition
    """
    from laktory.models.stacks.terraformstack import TerraformStack

    env = self.get_env(env_name=env_name)
    if vars:
        env = env.model_copy(update={"variables": {**env.variables, **vars}})
    env = env.inject_vars()
    env.build(env_name=None, inject_vars=False)

    # Providers
    providers = {}
    for r in env.resources._get_all(providers_only=True).values():
        for _r in r.core_resources:
            rname = _r.resource_name
            providers[rname] = _r

    # Resources
    resources = {}
    for r in env.resources._get_all(providers_excluded=True).values():
        for _r in r.core_resources:
            resources[_r.resource_name] = _r

    self._check_depends_on(resources, providers)

    # Auto-configure Databricks workspace state backend when
    # backend.databricks_workspace is set.
    ws_backend = None
    if env.terraform.backend and "databricks_workspace" in env.terraform.backend:
        ws_backend = env.terraform.backend.pop("databricks_workspace")
        if not env.terraform.backend:
            env.terraform.backend = None
        if ws_backend is False:
            ws_backend = None

    if ws_backend is not None:
        if env.terraform.backend:
            logger.warning(
                "'databricks_workspace' is set alongside other backend keys; "
                "the other keys take precedence and databricks_workspace is ignored."
            )
        else:
            db_provider = next(
                (
                    p
                    for p in providers.values()
                    if isinstance(p, DatabricksProvider)
                ),
                None,
            )
            if db_provider is None:
                raise ValueError(
                    "backend.databricks_workspace requires a DatabricksProvider in the stack."
                )
            wc = db_provider.workspace_client
            host = (db_provider.host or wc.config.host or "").rstrip("/")
            if not host:
                raise ValueError(
                    "Could not resolve Databricks host for backend.databricks_workspace. "
                    "Set 'host' explicitly in the DatabricksProvider or ensure the profile "
                    "in ~/.databrickscfg includes a host."
                )

            username = wc.current_user.me().user_name

            # The Terraform HTTP backend uses Basic auth (Authorization:
            # Basic base64("token:{password}")). Databricks accepts this
            # for PAT tokens but NOT for Azure AD / OAuth tokens, which
            # require Bearer auth. When no PAT is configured, create a
            # Databricks PAT via the Tokens API and cache it on disk so the
            # backend configuration stays stable across laktory init / deploy
            # invocations. The cache is scoped to CACHE_ROOT (next to
            # stack.tf.json) and must be kept out of source control.
            if db_provider.token:
                token = db_provider.token
            else:
                from laktory.constants import CACHE_ROOT

                cache_path = os.path.join(CACHE_ROOT, ".terraform", _WS_TOKEN_CACHE)
                try:
                    token = _get_cached_ws_token(wc, cache_path)
                except Exception as exc:
                    raise ValueError(
                        "backend.databricks_workspace: could not create or "
                        "retrieve a Databricks PAT token for the Terraform "
                        "HTTP backend. This is required when using service "
                        "principal authentication because Azure AD tokens are "
                        "not accepted via Basic auth. Ensure the service "
                        "principal has permission to create tokens (workspace "
                        "Admin Console > Settings > Advanced > 'Allow service "
                        "principals to use personal access tokens'), or set an "
                        "explicit 'token' in your DatabricksProvider. "
                        f"Original error: {exc}"
                    ) from exc

            if ws_backend is not True:
                raise ValueError(
                    "backend.databricks_workspace must be set to `true`. "
                    "To use a fully custom backend, configure the `http` backend directly."
                )

            if env_name:
                state_path = f"/Users/{username}/.laktory/{self.name}/{env_name}/state/terraform.tfstate"
            else:
                state_path = f"/Users/{username}/.laktory/{self.name}/state/terraform.tfstate"
            ws_parent = state_path.rsplit("/", 1)[0]
            wc.workspace.mkdirs(path=ws_parent)

            url = f"{host}/api/2.0/workspace-files{state_path}"
            env.terraform.backend = {
                "http": {
                    "address": url,
                    "username": "token",
                    "password": token,
                    "retry_wait_min": "5",
                }
            }

    # Update terraform
    return TerraformStack(
        terraform={"backend": env.terraform.backend},
        providers=providers,
        resources=resources,
    )

---

laktory.models.stacks.StackResources

Bases: BaseModel

Resources definition for a given stack or stack environment.

PARAMETER	DESCRIPTION
databricks_accesscontrolrulesets	TYPE: dict[str | VariableType, AccessControlRuleSet | VariableType] | VariableType DEFAULT: {}
databricks_alerts	TYPE: dict[str | VariableType, Alert | VariableType] | VariableType DEFAULT: {}
databricks_apps	TYPE: dict[str | VariableType, App | VariableType] | VariableType DEFAULT: {}
databricks_catalogs	TYPE: dict[str | VariableType, Catalog | VariableType] | VariableType DEFAULT: {}
databricks_clusterpolicies	TYPE: dict[str | VariableType, ClusterPolicy | VariableType] | VariableType DEFAULT: {}
databricks_clusters	TYPE: dict[str | VariableType, Cluster | VariableType] | VariableType DEFAULT: {}
databricks_connections	TYPE: dict[str | VariableType, Connection | VariableType] | VariableType DEFAULT: {}
databricks_currentusers	TYPE: dict[str | VariableType, CurrentUser | VariableType] | VariableType DEFAULT: {}
databricks_dashboards	TYPE: dict[str | VariableType, Dashboard | VariableType] | VariableType DEFAULT: {}
databricks_dataqualitymonitors	TYPE: dict[str | VariableType, DataQualityMonitor | VariableType] | VariableType DEFAULT: {}
databricks_dbfsfiles	TYPE: dict[str | VariableType, DbfsFile | VariableType] | VariableType DEFAULT: {}
databricks_directories	TYPE: dict[str | VariableType, Directory | VariableType] | VariableType DEFAULT: {}
databricks_entitlements	TYPE: dict[str | VariableType, Entitlements | VariableType] | VariableType DEFAULT: {}
databricks_externallocations	TYPE: dict[str | VariableType, ExternalLocation | VariableType] | VariableType DEFAULT: {}
databricks_grant	TYPE: dict[str | VariableType, Grant | VariableType] | VariableType DEFAULT: {}
databricks_grants	TYPE: dict[str | VariableType, Grants | VariableType] | VariableType DEFAULT: {}
databricks_groups	TYPE: dict[str | VariableType, Group | VariableType] | VariableType DEFAULT: {}
databricks_jobs	TYPE: dict[str | VariableType, Job | VariableType] | VariableType DEFAULT: {}
databricks_metastoreassignments	TYPE: dict[str | VariableType, MetastoreAssignment | VariableType] | VariableType DEFAULT: {}
databricks_metastoredataaccesses	TYPE: dict[str | VariableType, MetastoreDataAccess | VariableType] | VariableType DEFAULT: {}
databricks_metastores	TYPE: dict[str | VariableType, Metastore | VariableType] | VariableType DEFAULT: {}
databricks_mlflowexperiments	TYPE: dict[str | VariableType, MLflowExperiment | VariableType] | VariableType DEFAULT: {}
databricks_mlflowmodels	TYPE: dict[str | VariableType, MLflowModel | VariableType] | VariableType DEFAULT: {}
databricks_mlflowwebhooks	TYPE: dict[str | VariableType, MLflowWebhook | VariableType] | VariableType DEFAULT: {}
databricks_networkconnectivityconfig	TYPE: dict[str | VariableType, MwsNetworkConnectivityConfig | VariableType] | VariableType DEFAULT: {}
databricks_notebooks	TYPE: dict[str | VariableType, Notebook | VariableType] | VariableType DEFAULT: {}
databricks_notificationdestinations	TYPE: dict[str | VariableType, NotificationDestination | VariableType] | VariableType DEFAULT: {}
databricks_obotokens	TYPE: dict[str | VariableType, OboToken | VariableType] | VariableType DEFAULT: {}
databricks_permissions	TYPE: dict[str | VariableType, Permissions | VariableType] | VariableType DEFAULT: {}
databricks_pipelines	TYPE: dict[str | VariableType, Pipeline | VariableType] | VariableType DEFAULT: {}
databricks_pythonpackages	TYPE: dict[str | VariableType, PythonPackage | VariableType] | VariableType DEFAULT: {}
databricks_queries	TYPE: dict[str | VariableType, Query | VariableType] | VariableType DEFAULT: {}
databricks_recipients	TYPE: dict[str | VariableType, Recipient | VariableType] | VariableType DEFAULT: {}
databricks_repos	TYPE: dict[str | VariableType, Repo | VariableType] | VariableType DEFAULT: {}
databricks_schemas	TYPE: dict[str | VariableType, Schema | VariableType] | VariableType DEFAULT: {}
databricks_secrets	TYPE: dict[str | VariableType, Secret | VariableType] | VariableType DEFAULT: {}
databricks_secretscopes	TYPE: dict[str | VariableType, SecretScope | VariableType] | VariableType DEFAULT: {}
databricks_serviceprincipals	TYPE: dict[str | VariableType, ServicePrincipal | VariableType] | VariableType DEFAULT: {}
databricks_shares	TYPE: dict[str | VariableType, Share | VariableType] | VariableType DEFAULT: {}
databricks_storagecredentials	TYPE: dict[str | VariableType, StorageCredential | VariableType] | VariableType DEFAULT: {}
databricks_tables	TYPE: dict[str | VariableType, Table | VariableType] | VariableType DEFAULT: {}
databricks_users	TYPE: dict[str | VariableType, User | VariableType] | VariableType DEFAULT: {}
databricks_vectorsearchendpoints	TYPE: dict[str | VariableType, VectorSearchEndpoint | VariableType] | VariableType DEFAULT: {}
databricks_vectorsearchindexes	TYPE: dict[str | VariableType, VectorSearchIndex | VariableType] | VariableType DEFAULT: {}
databricks_volumes	TYPE: dict[str | VariableType, Volume | VariableType] | VariableType DEFAULT: {}
databricks_warehouses	TYPE: dict[str | VariableType, Warehouse | VariableType] | VariableType DEFAULT: {}
databricks_workspacebindings	TYPE: dict[str | VariableType, WorkspaceBinding | VariableType] | VariableType DEFAULT: {}
databricks_workspacefiles	TYPE: dict[str | VariableType, WorkspaceFile | PythonPackage | VariableType] | VariableType DEFAULT: {}
databricks_workspacetrees	TYPE: dict[str | VariableType, WorkspaceTree | PythonPackage | VariableType] | VariableType DEFAULT: {}
pipelines	TYPE: dict[str | VariableType, Pipeline | VariableType] | VariableType DEFAULT: {}
providers	TYPE: dict[str | VariableType, AWSProvider | AzureProvider | DatabricksProvider | VariableType] | VariableType DEFAULT: {}

METHOD	DESCRIPTION
route_providers_by_key	Use the provider key name (Terraform convention) as the discriminator.

route_providers_by_key(v) classmethod

Use the provider key name (Terraform convention) as the discriminator.

AWSProvider and DatabricksProvider share fields like profile and token, so Pydantic's union matching is ambiguous. The key name is the explicit source of truth: "databricks[.]" → DatabricksProvider, "aws[.]" → AWSProvider, "azure[rm][.*]" → AzureProvider.

Source code in laktory/models/stacks/stack.py

@field_validator("providers", mode="before")
@classmethod
def route_providers_by_key(cls, v):
    """Use the provider key name (Terraform convention) as the discriminator.

    AWSProvider and DatabricksProvider share fields like `profile` and `token`,
    so Pydantic's union matching is ambiguous. The key name is the explicit
    source of truth: "databricks[.*]" → DatabricksProvider, "aws[.*]" →
    AWSProvider, "azure[rm][.*]" → AzureProvider.
    """
    if not isinstance(v, dict):
        return v
    result = {}
    for key, value in v.items():
        if not isinstance(value, dict):
            result[key] = value
            continue
        base = key.split(".")[0].lower()
        if base == "databricks":
            result[key] = DatabricksProvider.model_validate(value)
        elif base == "aws":
            result[key] = AWSProvider.model_validate(value)
        elif base in ("azure", "azurerm"):
            result[key] = AzureProvider.model_validate(value)
        else:
            result[key] = value
    return result

---

laktory.models.stacks.stack.LaktorySettings

Bases: BaseModel

Laktory Settings

PARAMETER	DESCRIPTION
build_root	Local directory where pipeline config JSON and resource files are written during build. Defaults to the Laktory cache directory. Use when deployment is delegated to third parties like Databricks Declarative Bundles. TYPE: str | VariableType DEFAULT: ''
dataframe_api	TYPE: Literal['NARWHALS', 'NATIVE'] | VariableType DEFAULT: None
dataframe_backend	DataFrame backend TYPE: str DEFAULT: None
runtime_root	Laktory cache root directory. Used when a pipeline needs to write checkpoint files. TYPE: str | VariableType DEFAULT: '/laktory/'
workspace_root	Root directory of a Databricks Workspace (excluding `'/Workspace') to which databricks objects like notebooks and workspace files are deployed. TYPE: str | VariableType DEFAULT: '/.laktory/'

---

laktory.models.stacks.stack.EnvironmentSettings

Bases: BaseModel

Settings overwrite for a specific environments

PARAMETER	DESCRIPTION
resources	Dictionary of resources to be deployed. Each key should be a resource type and each value should be a dictionary of resources who's keys are the resource names and the values the resources definitions. TYPE: Any DEFAULT: None
terraform	Terraform-specific settings TYPE: Terraform | VariableType DEFAULT: Terraform(variables={}, backend=None)

---

laktory.models.stacks.stack.Terraform

Bases: BaseModel

PARAMETER

DESCRIPTION

backend

Terraform backend configuration. Accepts any standard Terraform backend block (e.g. azurerm, s3, http). Additionally, the special key databricks_workspace: true auto-configures a Terraform HTTP backend that stores state as a workspace file in the current Databricks user's directory at /Users/{user}/.laktory/{stack}/{env}/state/terraform.tfstate. No external storage account is required. TYPE: dict[str, Any] | None | VariableType DEFAULT: None