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_accountfederationpolicies={}, databricks_alerts={}, databricks_apps={}, databricks_budgets={}, databricks_budgetpolicies={}, databricks_catalogs={}, databricks_clusterpolicies={}, databricks_clusters={}, databricks_connections={}, databricks_currentusers={}, databricks_dashboards={}, databricks_dbfsfiles={}, databricks_directories={}, databricks_entitlements={}, databricks_pipelines={}, databricks_externallocations={}, databricks_gitcredentials={}, databricks_globalinitscripts={}, databricks_grant={}, databricks_grants={}, databricks_groups={}, databricks_instancepools={}, databricks_instanceprofiles={}, databricks_ipaccesslists={}, databricks_jobs={}, databricks_libraries={}, databricks_metastoreassignments={}, databricks_metastoredataaccesses={}, databricks_metastores={}, databricks_mlflowexperiments={}, databricks_mlflowmodels={}, databricks_mlflowwebhooks={}, databricks_modelservings={}, databricks_mwscredentials={}, databricks_mwscustomermanagedkeys={}, databricks_mwslogdeliveries={}, databricks_mwsnccprivateendpointrules={}, databricks_networkconnectivityconfig={}, databricks_mwsnetworks={}, databricks_mwsprivateaccesssettings={}, databricks_mwsstorageconfigurations={}, databricks_mwsvpcendpoints={}, databricks_mwsworkspaces={}, databricks_notebooks={}, databricks_notificationdestinations={}, databricks_obotokens={}, databricks_permissions={}, databricks_dataqualitymonitors={}, databricks_pythonpackages={}, databricks_queries={}, databricks_recipients={}, databricks_registeredmodels={}, databricks_repos={}, databricks_schemas={}, databricks_secrets={}, databricks_secretscopes={}, databricks_serviceprincipals={}, databricks_serviceprincipalfederationpolicies={}, databricks_shares={}, databricks_storagecredentials={}, databricks_tables={}, databricks_tokens={}, 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
inject_vars	Same as BaseModel.inject_vars(), but resolves self.settings and
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}})
        # Gated on `inject_vars` (not just run unconditionally) because
        # `to_terraform()` calls `env.build(env_name=None, inject_vars=False)`
        # internally on an already-resolved `env` - that internal call must
        # not re-run this with the wrong (None) env_name; to_terraform()
        # resolves it separately, beforehand, with its own correct env_name.
        #
        # Must run before `env.inject_vars()`: that call also caches
        # `core_resources` (and any `depends_on` baked into it) on every
        # resource in the stack, so resolving `workspace_root: "user_root"`
        # afterward would leave resource names computed from the
        # not-yet-resolved root frozen in already-cached `depends_on`
        # entries, out of sync with the final resource names.
        self._resolve_user_root(env, env_name)
        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:
                logger.warning(
                    f"Pipeline '{r.name}' has no orchestrator. Skipping."
                )
                continue

            # Orchestrator fields derived from `settings.workspace_root`
            # (e.g. a LAKEFLOW_JOB task's `filepath`, a
            # LAKEFLOW_DECLARATIVE_PIPELINE's `config_filepath`
            # configuration entry) are computed by `update_from_parent()`,
            # which first runs at initial model construction/validation -
            # before `_resolve_user_root()` (above) has had a chance to
            # resolve `workspace_root: "user_root"`. Re-run it now so
            # those fields reflect the resolved root.
            orchestrator.update_from_parent()

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

    logger.info("Rendering variable-templated file content...")
    for k, r in env.resources._get_all(providers_excluded=True).items():
        if isinstance(r, RenderableFileMixin) and r.render_vars:
            r.build(vars=env.variables)
        if isinstance(r, WorkspaceTree):
            for _r in r.core_resources:
                if isinstance(_r, RenderableFileMixin) and _r.render_vars:
                    _r.build(vars=env.variables)

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

inject_vars(inplace=False, vars=None, objs=None)

Same as BaseModel.inject_vars(), but resolves self.settings and pushes it onto the global laktory._settings.settings singleton first, deterministically, before resolving the rest of the stack.

BaseModel.inject_vars()'s field loop iterates model_fields_set (a set, unordered) and may skip re-resolution entirely via its result cache, so settings is not guaranteed to be resolved - and pushed onto the singleton - before other fields (e.g. a ${settings.x} reference elsewhere in the stack) are substituted in the same pass.

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

def inject_vars(self, inplace: bool = False, vars: dict = None, objs: dict = None):
    """
    Same as `BaseModel.inject_vars()`, but resolves `self.settings` and
    pushes it onto the global `laktory._settings.settings` singleton
    first, deterministically, before resolving the rest of the stack.

    `BaseModel.inject_vars()`'s field loop iterates `model_fields_set` (a
    `set`, unordered) and may skip re-resolution entirely via its result
    cache, so `settings` is not guaranteed to be resolved - and pushed
    onto the singleton - before other fields (e.g. a `${settings.x}`
    reference elsewhere in the stack) are substituted in the same pass.
    """
    if self.settings is not None:
        merged_vars = deepcopy({**(vars or {}), **self.variables})
        self.settings.inject_vars(vars=merged_vars, objs=objs).apply_settings()

    return super().inject_vars(inplace=inplace, vars=vars, objs=objs)

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}})
    # Must run before `env.inject_vars()` - see the comment in `build()`.
    _resolved_provider, _resolved_wc, _resolved_username = self._resolve_user_root(
        env, env_name
    )
    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 = {}
    virtual_children = {}
    for r in env.resources._get_all(providers_excluded=True).values():
        for _r in r.core_resources:
            resources[_r.resource_name] = _r
        # Virtual resources (e.g. WorkspaceTree, Pipeline) emit no Terraform
        # block of their own; keep track of the concrete children they expand
        # into so dependencies on them can be resolved.
        if not r.self_as_core_resources:
            virtual_children[r.resource_name] = [
                _r.resource_name for _r in r.core_resources
            ]

    # A `depends_on` reference to a virtual resource means a dependency on all
    # the resources it generates. Expand those references before they reach
    # Terraform, which has no block to point at otherwise.
    if virtual_children:
        self._expand_virtual_depends_on(resources, virtual_children)

    self._order_workspace_files_creation(resources)

    self._check_depends_on(resources, providers, virtual_children)

    # 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:
            # Reuse the provider/username `_resolve_user_root()` already
            # resolved above (if `workspace_root: "user_root"` was set),
            # rather than making a second, redundant `current_user.me()`
            # API call.
            if _resolved_provider is not None:
                db_provider, wc, username = (
                    _resolved_provider,
                    _resolved_wc,
                    _resolved_username,
                )
            else:
                db_provider, wc, username = (
                    _resolve_databricks_provider_and_username(env)
                )
            if db_provider is None:
                raise ValueError(
                    "backend.databricks_workspace requires a DatabricksProvider in the stack."
                )
            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."
                )

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

            state_path = (
                f"{_user_workspace_root(username, env.name, env_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_accountfederationpolicies	TYPE: dict[str | VariableType, AccountFederationPolicy | VariableType] | VariableType DEFAULT: {}
databricks_alerts	TYPE: dict[str | VariableType, Alert | VariableType] | VariableType DEFAULT: {}
databricks_apps	TYPE: dict[str | VariableType, App | VariableType] | VariableType DEFAULT: {}
databricks_budgetpolicies	TYPE: dict[str | VariableType, BudgetPolicy | VariableType] | VariableType DEFAULT: {}
databricks_budgets	TYPE: dict[str | VariableType, Budget | 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_gitcredentials	TYPE: dict[str | VariableType, GitCredential | VariableType] | VariableType DEFAULT: {}
databricks_globalinitscripts	TYPE: dict[str | VariableType, GlobalInitScript | 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_instancepools	TYPE: dict[str | VariableType, InstancePool | VariableType] | VariableType DEFAULT: {}
databricks_instanceprofiles	TYPE: dict[str | VariableType, InstanceProfile | VariableType] | VariableType DEFAULT: {}
databricks_ipaccesslists	TYPE: dict[str | VariableType, IpAccessList | VariableType] | VariableType DEFAULT: {}
databricks_jobs	TYPE: dict[str | VariableType, Job | VariableType] | VariableType DEFAULT: {}
databricks_libraries	TYPE: dict[str | VariableType, Library | 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_modelservings	TYPE: dict[str | VariableType, ModelServing | VariableType] | VariableType DEFAULT: {}
databricks_mwscredentials	TYPE: dict[str | VariableType, MwsCredentials | VariableType] | VariableType DEFAULT: {}
databricks_mwscustomermanagedkeys	TYPE: dict[str | VariableType, MwsCustomerManagedKeys | VariableType] | VariableType DEFAULT: {}
databricks_mwslogdeliveries	TYPE: dict[str | VariableType, MwsLogDelivery | VariableType] | VariableType DEFAULT: {}
databricks_mwsnccprivateendpointrules	TYPE: dict[str | VariableType, MwsNccPrivateEndpointRule | VariableType] | VariableType DEFAULT: {}
databricks_mwsnetworks	TYPE: dict[str | VariableType, MwsNetworks | VariableType] | VariableType DEFAULT: {}
databricks_mwsprivateaccesssettings	TYPE: dict[str | VariableType, MwsPrivateAccessSettings | VariableType] | VariableType DEFAULT: {}
databricks_mwsstorageconfigurations	TYPE: dict[str | VariableType, MwsStorageConfigurations | VariableType] | VariableType DEFAULT: {}
databricks_mwsvpcendpoints	TYPE: dict[str | VariableType, MwsVpcEndpoint | VariableType] | VariableType DEFAULT: {}
databricks_mwsworkspaces	TYPE: dict[str | VariableType, MwsWorkspaces | 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_registeredmodels	TYPE: dict[str | VariableType, RegisteredModel | 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_serviceprincipalfederationpolicies	TYPE: dict[str | VariableType, ServicePrincipalFederationPolicy | 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_tokens	TYPE: dict[str | VariableType, Token | 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	DataFrame API (NARWHALS / NATIVE). Default: NARWHALS (from LAKTORY_DATAFRAME_API). TYPE: Literal['NARWHALS', 'NATIVE'] | VariableType DEFAULT: None
dataframe_backend	DataFrame backend (POLARS / PYSPARK). Default: PYSPARK (from LAKTORY_DATAFRAME_BACKEND). TYPE: str DEFAULT: None
runtime_root	Laktory cache root directory. Used when a pipeline needs to write checkpoint files. The default assumes a DBFS FUSE mount, which is unavailable on newer/serverless Databricks workspaces - on those, set this explicitly to a Unity Catalog Volume path instead, e.g. /Volumes/{catalog}/{schema}/{volume}/{path}/. See Laktory Settings. 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. Set to the reserved value'user_root'to auto-compute/Users/{you}/.laktory/{stack_name}/{env_name}/instead. Requires aDatabricksProvider` in the stack (resolves your username via a live SDK call). See Laktory Settings. 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