Core

Package: databricks.bundles.core

Classes

class Resources

Resources is a collection of resources in a bundle.

Resources class is returned by ‘load_resources’ function specified in databricks.yml. Each element in ‘python/resources’ list is a fully qualified function name that returns an instance of Resources class. If there are multiple functions specified in ‘python/resources’ list, the resources from all functions are combined into a single Resources object.

Example:

experimental:
  python:
    resources:
      - "resources:load_resources"

load_resources function can be implemented using built-in functions:

Programmatic construction of resources is supported using add_resource() and add_job() methods.

Example:

def load_resources(bundle: Bundle) -> Resources:
    resources = Resources()

    for resource_name, config in get_configs():
        job = create_job(config)

        resources.add_job(resource_name, job)

    return resources
property jobs: dict[str, Job]
property pipelines: dict[str, Pipeline]
property diagnostics: Diagnostics

Returns diagnostics. If there are any diagnostic errors, bundle validation fails.

add_resource(
resource_name: str,
resource: Resource,
*,
location: Location | None,
) → None

Adds a resource to the collection of resources. Resource name must be unique across all resources of the same type.

Parameters:

  • resource_name – unique identifier for the resource

  • resource – the resource to add

  • location – optional location of the resource in the source code

add_job(
resource_name: str,
job: JobParam,
*,
location: Location | None,
) → None

Adds a job to the collection of resources. Resource name must be unique across all jobs.

Parameters:

  • resource_name – unique identifier for the job

  • job – the job to add, can be Job or dict

  • location – optional location of the job in the source code

add_pipeline(
resource_name: str,
pipeline: PipelineParam,
*,
location: Location | None,
) → None

Adds a pipeline to the collection of resources. Resource name must be unique across all pipelines.

Parameters:

  • resource_name – unique identifier for the pipeline

  • pipeline – the pipeline to add, can be Pipeline or dict

  • location – optional location of the pipeline in the source code

add_location(
path: tuple[str, ...],
location: Location,
) → None

Associate source code location with a path in the bundle configuration.

add_diagnostics(
other: Diagnostics,
) → None

Add diagnostics from another Diagnostics object. :param other: :return:

add_diagnostic_error(
msg: str,
*,
detail: str | None,
path: tuple[str, ...] | None,
location: Location | None,
) → None

Report a diagnostic error. If there are any diagnostic errors, bundle validation fails.

Parameters:

  • msg – short summary of the error

  • detail – optional detailed description of the error

  • path – optional path in bundle configuration where the error occurred

  • location – optional location in the source code where the error occurred

add_diagnostic_warning(
msg: str,
*,
detail: str | None,
path: tuple[str, ...] | None,
location: Location | None,
) → None

Report a diagnostic warning. Warnings are informational and do not cause bundle validation to fail.

Parameters:

  • msg – short summary of the warning

  • detail – optional detailed description of the warning

  • path – optional path in bundle configuration where the warning occurred

  • location – optional location in the source code where the warning occurred

add_resources(
other: Resources,
) → None

Add resources from another Resources object.

Adds error to diagnostics if there are duplicate resource names.

class Resource

Base class for all resources.

class ResourceMutator

Resource mutators are used to modify resources before they are deployed.

Mutators are applied both to resources defined in YAML and Python. Mutators are applied in the order they are defined in databricks.yml.

Example:

experimental:
    python:
        mutators:
        - "resources:my_job_mutator"

from databricks.bundles.core import Bundle, job_mutator
from databricks.bundles.jobs import Job


@job_mutator
def my_job_mutator(bundle: Bundle, job: Job) -> Job:
    return replace(job, name="my_job")

See databricks.bundles.core.job_mutator().

resource_type: Type[_T]

Resource type that this mutator applies to.

function: Callable

Underling function that was decorated. Can be accessed for unit-testing.

class Bundle

Bundle contains information about a bundle accessible in functions loading and mutating resources.

target: str

Selected target where the bundle is being loaded. E.g.: ‘development’, ‘staging’, or ‘production’.

variables: dict[str, Any]

Values of bundle variables resolved for selected target. Bundle variables are defined in databricks.yml. For accessing variables as structured data, use resolve_variable().

Example:

variables:
  default_dbr_version:
    description: Default version of Databricks Runtime
    default: "14.3.x-scala2.12"
resolve_variable(
variable: Variable[_T] | _T,
) → _T

Resolve a variable to its value.

If the value is a variable, it will be resolved and returned. Otherwise, the value will be returned as is.

resolve_variable_list(
variable: Variable[list[Variable[_T] | _T]] | list[Variable[_T] | _T],
) → list[_T]

Resolve a list variable to its value.

If the value is a variable, or the list item is a variable, it will be resolved and returned. Otherwise, the value will be returned as is.

class Variable

Reference to a bundle variable.

See: Databricks Asset Bundles configuration

path: str

Path to the variable, e.g. “var.my_variable”.

type: Type[_T]

Type of the variable.

property value: str

Returns the variable path in the format of “${path}”

class Diagnostics

Diagnostics is a collection of errors and warnings we print to users.

Each item can have source location or path associated, that is reported in output to indicate where the error or warning occurred.

items: tuple[Diagnostic, ...]
extend(
diagnostics: Self,
) → Self

Extend items with another diagnostics. This pattern allows to accumulate errors and warnings.

Example:

def foo() -> Diagnostics: ...
def bar() -> Diagnostics: ...

diagnostics = Diagnostics()
diagnostics = diagnostics.extend(foo())
diagnostics = diagnostics.extend(bar())
extend_tuple(
pair: tuple[_T, Self],
) → tuple[_T, Self]

Extend items with another diagnostics. This variant is useful when methods return a pair of value and diagnostics. This pattern allows to accumulate errors and warnings.

Example:

def foo() -> (int, Diagnostics): ...

diagnostics = Diagnostics()
value, diagnostics = diagnostics.extend_tuple(foo())
has_error() → bool

Returns True if there is at least one error in diagnostics.

classmethod create_error(
msg: str,
*,
detail: str | None,
location: Location | None,
path: tuple[str, ...] | None,
) → Self

Create an error diagnostics.

classmethod create_warning(
msg: str,
*,
detail: str | None,
location: Location | None,
path: tuple[str, ...] | None,
) → Self

Create a warning diagnostics.

classmethod from_exception(
exc: Exception,
*,
summary: str,
location: Location | None,
path: tuple[str, ...] | None,
explanation: str | None,
) → Self

Create diagnostics from an exception.

Parameters:

  • exc – exception to create diagnostics from

  • summary – short summary of the error

  • location – optional location in the source code where the error occurred

  • path – optional path to relevant property in databricks.yml

  • explanation – optional explanation to add to the details

class Diagnostic
severity: Severity

Severity of the diagnostics item.

summary: str

Short summary of the error or warning.

detail: str | None = None

Explanation of the error or warning.

path: tuple[str, ...] | None = None

Path in databricks.yml where the error or warning occurred.

location: Location | None = None

Source code location where the error or warning occurred.

as_dict() → dict
class Location
file: str
line: int | None = None
column: int | None = None
static from_callable(
fn: Callable,
) → Location | None

Capture location of callable. This is useful for creating diagnostics of decorated functions.

static from_stack_frame(
depth: int = 0,
) → Location

Capture location of the caller

as_dict() → dict
class Severity
WARNING = 'warning'
ERROR = 'error'
class T

TypeVar for variable value

Methods

core.load_resources_from_current_package_module() → Resources

Load resources from all submodules of the current package module.

core.load_resources_from_module(
module: ModuleType,
) → Resources

Load resources from the given module.

For recursive loading of resources from submodules, use load_resources_from_package_module.

Parameters:

module – module to load resources from

core.load_resources_from_modules(
modules: Iterable[ModuleType],
) → Resources

Load resources from the given modules.

For recursive loading of resources from submodules, use load_resources_from_package_module.

Parameters:

modules – list of modules to load resources from

core.load_resources_from_package_module(
package_module: ModuleType,
) → Resources

Load resources from all submodules of the given package module.

Parameters:

package_module – package module to load resources from

Decorators

@job_mutator(
function: Callable[[Bundle, Job], Job],
) → ResourceMutator[Job]
@variables(
cls: type[_T],
) → type[_T]

A decorator that initializes each annotated attribute in a class with Variable type. Variables are initialized with a path that corresponds to the attribute name. Variables should specify their type, or else they will be treated as Any. Complex types like data classes, lists or dictionaries are supported.

For example, if your databricks.yml file contains:

variables:
  warehouse_id:
    description: Warehouse ID for SQL tasks
    default: ...

You can define a class with a warehouse_id attribute:

@variables
class MyVariables:
  warehouse_id: Variable[str] # ${var.warehouse_id}

And later use it in your code as MyVariables.warehouse_id.

For accessing bundle variable values, see Bundle.resolve_variable().