`yggdrasil.databricks.job.skeleton`¶

skeleton ¶

Prefect-style tasks & flows that deploy as Databricks serverless jobs.

The :func:task / :func:flow decorators wrap a function into a callable :class:Task / :class:Flow:

.local(x) runs the body in-process (tests, debugging, the on-cluster body of a deployed job);
.submit(...) runs in the background and returns a :class:Future (.result() blocks) — a flow fans tasks out with .submit / .map;
.deploy(client) registers it as a Databricks serverless job for a schedule / file-arrival trigger, shipping the live code as a wheel.

@flow(name="autoload", command=["databricks", "table", "autoload", "--table", "c.s.t", "--source", "s3://…"]) def autoloader(): ...

autoloader.deploy(client) # register as a serverless job

The deploy ships the live code as a wheel (built from the package on disk — dev checkout or installed), placed in the shared workspace pypi registry, or in a per-user folder + rebuilt when the package is an editable install. The single serverless python-wheel task invokes the ygg entry point with an explicit :attr:~Flow.command — the CLI subcommand the deployed job runs on the cluster, shipped verbatim as the wheel-task parameters (e.g. databricks table autoload --table … --source … for an Auto Loader job).

Class-based flows subclass :class:Flow and override :meth:~Flow.run (the body), :attr:~_Runnable.name, :meth:~_Runnable.command, :meth:~Flow.trigger.

Future ¶

Future(job: 'ThreadJob')

Bases: Generic[T]

Handle to a :meth:Task.submit / :meth:Flow.submit background run.

result ¶

result(timeout: Any = None, *, raise_error: bool = True) -> T

Block until the run finishes and return its result.

Task ¶

Task(
    fn: Callable[..., T],
    *,
    name: Optional[str] = None,
    retries: int = 0,
    retry_delay_seconds: float = 0.0,
    tags: "tuple[str, ...]" = (),
    key: Optional[str] = None,
    depends_on: "tuple[str, ...] | list[str]" = (),
    entry_point: Optional[str] = None,
    package_name: Optional[str] = None,
    command: list[str] | tuple[str, ...] | None = None,
    **task_options: Any
)

Bases: _Runnable, Generic[T]

A callable unit of work; also deployable as one databricks Task.

to_task ¶

to_task(parameters: list[str] | None = None) -> Any

Render a databricks Task (python-wheel) with explicit parameters and dependency edges — for hand-built multi-task job DAGs (the single-task deploy uses :meth:tasks). parameters are the ygg CLI subcommand the wheel-task runs on the cluster, shipped verbatim (e.g. ["databricks", "table", "autoload", …]).

run ¶

run(*args: Any, **kwargs: Any) -> Any

The body. Override it, or wrap a function with the decorator.

local ¶

local(*args: Any, **kwargs: Any) -> Any

Run in-process (honouring :attr:retries), skipping the remote routing — the escape hatch for tests and the on-cluster runner.

submit ¶

submit(*args: Any, **kwargs: Any) -> 'Future'

Run in the background; return a :class:Future.

map ¶

map(iterable: Any, **kwargs: Any) -> 'list[Future]'

:meth:submit once per item — Prefect-style fan-out.

with_options ¶

with_options(**overrides: Any) -> Any

Return a copy with attributes overridden.

command ¶

command() -> list[str] | None

The ygg CLI subcommand the deployed wheel-task runs on the cluster (e.g. ["databricks", "table", "autoload", "--table", …]), shipped verbatim as the python-wheel task parameters. None (the default) means no on-cluster command was configured — set command= on the flow / override this in a class-based flow.

wheel_package ¶

wheel_package() -> str

The top-level import package to wheel — where this task/flow is defined, so the deploy adapts to any project (the wheel is built from this package's live files on disk).

effective_dependencies ¶

effective_dependencies() -> list[str]

Shipped wheels once :meth:deploy has composed them, else the published :attr:dependencies (ygg pinned to the running version) + :attr:extra_dependencies.

environments ¶

environments() -> Optional[list]

Serverless environment list, or None when not serverless.

After :meth:deploy built the project's base environment(s), the default env references the version-pinned .yml by path (base_environment); with :attr:all_environments one env per supported Python (keyed py3XX) is appended so a task can run under any Python by key. Before a deploy (or for a non-built job) it falls back to an inline :meth:effective_dependencies list.

tasks ¶

tasks() -> list

The single serverless python-wheel task that runs the ygg :meth:command on the cluster (shipped verbatim as parameters).

definition ¶

definition() -> dict

Render the :meth:Jobs.create_or_update kwargs for this task/flow.

deploy ¶

deploy(client: Any) -> 'Job'

Get-or-create the live :class:Job from :meth:definition (without running it). When :attr:build_wheel is set, ships the live package as wheels (:meth:_serverless_dependencies) so the cluster runs this code.

Flow ¶

Flow(
    fn: Optional[Callable] = None,
    *,
    name: Optional[str] = None,
    trigger: Any = None,
    retries: int = 0,
    retry_delay_seconds: float = 0.0,
    command: tuple[str, ...] | list[str] | None = None,
    entry_point: Optional[str] = None,
    package_name: Optional[str] = None
)

Bases: _Runnable

A callable flow; deploys as a Databricks serverless job.

trigger ¶

trigger() -> Any

The databricks TriggerSettings (file-arrival / schedule), or None. Function-built flows carry the @flow(trigger=...) value.

definition ¶

definition() -> dict

:class:_Runnable.definition plus the schedule/file-arrival trigger.

run ¶

run(*args: Any, **kwargs: Any) -> Any

The body. Override it, or wrap a function with the decorator.

local ¶

local(*args: Any, **kwargs: Any) -> Any

Run in-process (honouring :attr:retries), skipping the remote routing — the escape hatch for tests and the on-cluster runner.

submit ¶

submit(*args: Any, **kwargs: Any) -> 'Future'

Run in the background; return a :class:Future.

map ¶

map(iterable: Any, **kwargs: Any) -> 'list[Future]'

:meth:submit once per item — Prefect-style fan-out.

with_options ¶

with_options(**overrides: Any) -> Any

Return a copy with attributes overridden.

command ¶

command() -> list[str] | None

The ygg CLI subcommand the deployed wheel-task runs on the cluster (e.g. ["databricks", "table", "autoload", "--table", …]), shipped verbatim as the python-wheel task parameters. None (the default) means no on-cluster command was configured — set command= on the flow / override this in a class-based flow.

wheel_package ¶

wheel_package() -> str

The top-level import package to wheel — where this task/flow is defined, so the deploy adapts to any project (the wheel is built from this package's live files on disk).

effective_dependencies ¶

effective_dependencies() -> list[str]

Shipped wheels once :meth:deploy has composed them, else the published :attr:dependencies (ygg pinned to the running version) + :attr:extra_dependencies.

environments ¶

environments() -> Optional[list]

Serverless environment list, or None when not serverless.

After :meth:deploy built the project's base environment(s), the default env references the version-pinned .yml by path (base_environment); with :attr:all_environments one env per supported Python (keyed py3XX) is appended so a task can run under any Python by key. Before a deploy (or for a non-built job) it falls back to an inline :meth:effective_dependencies list.

tasks ¶

tasks() -> list

The single serverless python-wheel task that runs the ygg :meth:command on the cluster (shipped verbatim as parameters).

deploy ¶

deploy(client: Any) -> 'Job'

Get-or-create the live :class:Job from :meth:definition (without running it). When :attr:build_wheel is set, ships the live package as wheels (:meth:_serverless_dependencies) so the cluster runs this code.

ensure_console_logging ¶

ensure_console_logging(
    name: str = "yggdrasil", level: int = logging.INFO
) -> None

Attach an INFO stdout handler to the name logger if it has none, so interactive deploys / job runs surface ygg logs (the default root config is WARNING-only). Idempotent and scoped — never touches the root logger.

task ¶

task(
    func: Optional[Callable] = None,
    *,
    name: Optional[str] = None,
    retries: int = 0,
    retry_delay_seconds: float = 0.0,
    key: Optional[str] = None,
    depends_on: "tuple[str, ...] | list[str]" = (),
    entry_point: Optional[str] = None,
    package_name: Optional[str] = None,
    **task_options: Any
) -> Any

Turn a function into a callable :class:Task (Prefect-style).

flow ¶

flow(
    func: Optional[Callable] = None,
    *,
    name: Optional[str] = None,
    trigger: Any = None,
    retries: int = 0,
    retry_delay_seconds: float = 0.0,
    command: tuple[str, ...] | list[str] | None = None,
    entry_point: Optional[str] = None,
    package_name: Optional[str] = None
) -> Any

Turn a function into a callable :class:Flow (Prefect-style).

yggdrasil.databricks.job.skeleton¶

skeleton ¶

Future ¶

result ¶

Task ¶

to_task ¶

run ¶

local ¶

submit ¶

map ¶

with_options ¶

command ¶

wheel_package ¶

effective_dependencies ¶

environments ¶

tasks ¶

definition ¶

deploy ¶

Flow ¶

trigger ¶

definition ¶

run ¶

local ¶

submit ¶

map ¶

with_options ¶

command ¶

wheel_package ¶

effective_dependencies ¶

environments ¶

tasks ¶

deploy ¶

ensure_console_logging ¶

task ¶

flow ¶

`yggdrasil.databricks.job.skeleton`¶