CLI

laktory.cli.version()

Return installed laktory version and installed dependencies.

Examples:

laktory version

References

• CLI

Source code in laktory/cli/_version.py

@app.command()
def version():
    """
    Return installed laktory version and installed dependencies.

    Examples
    --------
    ```cmd
    laktory version
    ```

    References
    ----------
    * [CLI](https://www.laktory.ai/concepts/cli/)
    """
    print(show_version_info())

---

laktory.cli.build(environment=None, filepath='./stack.yaml')

Build temporary files and python packages into settings.build_root if specified or default cache dir if not. These files may also be used when deployment is delegated to third parties like Databricks Declarative Bundles.

PARAMETER	DESCRIPTION
environment	Name of the environment. TYPE: Annotated[str, Option(--env, -e, help='Name of the environment')] DEFAULT: None
filepath	Stack (yaml) filepath. TYPE: Annotated[str, Option(help='Stack (yaml) filepath.')] DEFAULT: './stack.yaml'

Examples:

laktory build --env dev

Source code in laktory/cli/_build.py

@app.command()
def build(
    environment: Annotated[
        str, typer.Option("--env", "-e", help="Name of the environment")
    ] = None,
    filepath: Annotated[
        str, typer.Option(help="Stack (yaml) filepath.")
    ] = "./stack.yaml",
):
    """
    Build temporary files and python packages into `settings.build_root` if
    specified or default cache dir if not. These files may also be used when
    deployment is delegated to third parties like Databricks Declarative Bundles.

    Parameters
    ----------
    environment:
        Name of the environment.
    filepath:
        Stack (yaml) filepath.

    Examples
    --------
    ```cmd
    laktory build --env dev
    ```
    """
    controller = CLIController(
        env=environment,
        stack_filepath=filepath,
    )

    # Call
    controller.build()

---

laktory.cli.validate(environment=None, filepath='./stack.yaml')

Validate configuration and resources.

PARAMETER	DESCRIPTION
environment	Name of the environment. TYPE: Annotated[str, Option(--env, -e, help='Name of the environment')] DEFAULT: None
filepath	Stack (yaml) filepath. TYPE: Annotated[str, Option(help='Stack (yaml) filepath.')] DEFAULT: './stack.yaml'

Examples:

laktory validate --env dev

References

• CLI

Source code in laktory/cli/_validate.py

@app.command()
def validate(
    environment: Annotated[
        str, typer.Option("--env", "-e", help="Name of the environment")
    ] = None,
    filepath: Annotated[
        str, typer.Option(help="Stack (yaml) filepath.")
    ] = "./stack.yaml",
):
    """
    Validate configuration and resources.

    Parameters
    ----------
    environment:
        Name of the environment.
    filepath:
        Stack (yaml) filepath.

    Examples
    --------
    ```cmd
    laktory validate --env dev
    ```

    References
    ----------
    * [CLI](https://www.laktory.ai/concepts/cli/)
    """
    CLIController(
        env=environment,
        stack_filepath=filepath,
    )

---

laktory.cli.preview(environment=None, filepath='./stack.yaml', options=None)

Validate configuration and resources and preview deployment.

PARAMETER	DESCRIPTION
environment	Name of the environment. TYPE: Annotated[str, Option(--env, -e, help='Name of the environment')] DEFAULT: None
filepath	Stack (yaml) filepath. TYPE: Annotated[str, Option(help='Stack (yaml) filepath.')] DEFAULT: './stack.yaml'
options	Comma separated IaC backend options (flags). TYPE: Annotated[str, Option(--options, help='Comma separated IaC backend options (flags).')] DEFAULT: None

Examples:

laktory preview --env dev

References

• CLI
• terraform plan

Source code in laktory/cli/_preview.py

@app.command()
def preview(
    environment: Annotated[
        str, typer.Option("--env", "-e", help="Name of the environment")
    ] = None,
    filepath: Annotated[
        str, typer.Option(help="Stack (yaml) filepath.")
    ] = "./stack.yaml",
    options: Annotated[
        str,
        typer.Option("--options", help="Comma separated IaC backend options (flags)."),
    ] = None,
):
    """
    Validate configuration and resources and preview deployment.

    Parameters
    ----------
    environment:
        Name of the environment.
    filepath:
        Stack (yaml) filepath.
    options:
        Comma separated IaC backend options (flags).

    Examples
    --------
    ```cmd
    laktory preview --env dev
    ```

    References
    ----------
    * [CLI](https://www.laktory.ai/concepts/cli/)
    * terraform [plan](https://developer.hashicorp.com/terraform/cli/commands/plan)
    """
    controller = CLIController(
        env=environment,
        stack_filepath=filepath,
        options_str=options,
    )

    # Call
    if controller.iac_backend == "terraform":
        controller.terraform_call("plan")
    else:
        raise ValueError(f"backend should be {SUPPORTED_BACKENDS}")

---

laktory.cli.deploy(environment=None, filepath='./stack.yaml', auto_approve=False, options=None)

Execute deployment.

PARAMETER	DESCRIPTION
environment	Name of the environment. TYPE: Annotated[str, Option(--env, -e, help='Name of the environment')] DEFAULT: None
filepath	Stack (yaml) filepath. TYPE: Annotated[str, Option(help='Stack (yaml) filepath.')] DEFAULT: './stack.yaml'
auto_approve	Automatically approve and perform the update after previewing it TYPE: Annotated[bool, Option(--yes, -y, help='Automatically approve and perform the update after previewing it')] DEFAULT: False
options	Comma separated IaC backend options (flags). TYPE: Annotated[str, Option(--options, help='Comma separated IaC backend options (flags).')] DEFAULT: None

Examples:

laktory deploy --env dev --filepath my-stack.yaml

References

• CLI
• terraform apply

Source code in laktory/cli/_deploy.py

@app.command()
def deploy(
    environment: Annotated[
        str, typer.Option("--env", "-e", help="Name of the environment")
    ] = None,
    filepath: Annotated[
        str, typer.Option(help="Stack (yaml) filepath.")
    ] = "./stack.yaml",
    auto_approve: Annotated[
        bool,
        typer.Option(
            "--yes",
            "-y",
            help="Automatically approve and perform the update after previewing it",
        ),
    ] = False,
    options: Annotated[
        str,
        typer.Option("--options", help="Comma separated IaC backend options (flags)."),
    ] = None,
):
    """
    Execute deployment.

    Parameters
    ----------
    environment:
        Name of the environment.
    filepath:
        Stack (yaml) filepath.
    auto_approve:
        Automatically approve and perform the update after previewing it
    options:
        Comma separated IaC backend options (flags).

    Examples
    --------
    ```cmd
    laktory deploy --env dev --filepath my-stack.yaml
    ```

    References
    ----------
    * [CLI](https://www.laktory.ai/concepts/cli/)
    * terraform [apply](https://developer.hashicorp.com/terraform/cli/commands/apply)
    """
    controller = CLIController(
        env=environment,
        auto_approve=auto_approve,
        stack_filepath=filepath,
        options_str=options,
    )

    # Call
    if controller.iac_backend == "terraform":
        controller.terraform_call("apply")
    else:
        raise ValueError(f"backend should be {SUPPORTED_BACKENDS}")

---

laktory.cli.run(databricks_job=None, databricks_pipeline=None, timeout=1200, raise_exception=True, full_refresh=False, current_run_action='WAIT', environment=None, filepath='./stack.yaml')

Execute remote job or DLT pipeline and monitor failures until completion.

PARAMETER	DESCRIPTION
databricks_job	Name of the job to run (mutually exclusive with dlt) TYPE: Annotated[str, Option(--dbks - job, -dbksj, help='Job name')] DEFAULT: None
databricks_pipeline	Name of the DLT pipeline to run (mutually exclusive with job) TYPE: Annotated[str, Option(--dbks - pipeline, -dbksp, help='Job name')] DEFAULT: None
timeout	Maximum allowed time (in seconds) for run. TYPE: Annotated[float, Option(--timeout, -t, help='Maximum allowed time (in seconds) for run')] DEFAULT: 1200
raise_exception	Raise exception on failure TYPE: Annotated[bool, Option('--raise', -r, help='Raise exception on failure')] DEFAULT: True
current_run_action	Action to take for currently running job or pipline. TYPE: Annotated[str, Option(--action, -a, help="Action to take if job currently running ['WAIT', 'CANCEL', 'FAIL']")] DEFAULT: 'WAIT'
full_refresh	Full tables refresh (pipline only) TYPE: Annotated[bool, Option(--full - refresh, --fr, help='Full tables refresh (pipeline only)')] DEFAULT: False
environment	Name of the environment. TYPE: Annotated[str, Option(--env, -e, help='Name of the environment')] DEFAULT: None
filepath	Stack (yaml) filepath. TYPE: Annotated[str, Option(help='Stack (yaml) filepath.')] DEFAULT: './stack.yaml'

Examples:

laktory run --env dev --dbks-pipeline pl-stock-prices --full_refresh --action CANCEL

References

• CLI

Source code in laktory/cli/_run.py

@app.command()
def run(
    databricks_job: Annotated[
        str, typer.Option("--dbks-job", "-dbksj", help="Job name")
    ] = None,
    databricks_pipeline: Annotated[
        str, typer.Option("--dbks-pipeline", "-dbksp", help="Job name")
    ] = None,
    timeout: Annotated[
        float,
        typer.Option(
            "--timeout", "-t", help="Maximum allowed time (in seconds) for run"
        ),
    ] = 1200,
    raise_exception: Annotated[
        bool, typer.Option("--raise", "-r", help="Raise exception on failure")
    ] = True,
    full_refresh: Annotated[
        bool,
        typer.Option(
            "--full-refresh", "--fr", help="Full tables refresh (pipeline only)"
        ),
    ] = False,
    current_run_action: Annotated[
        str,
        typer.Option(
            "--action",
            "-a",
            help="Action to take if job currently running ['WAIT', 'CANCEL', 'FAIL']",
        ),
    ] = "WAIT",
    environment: Annotated[
        str, typer.Option("--env", "-e", help="Name of the environment")
    ] = None,
    filepath: Annotated[
        str, typer.Option(help="Stack (yaml) filepath.")
    ] = "./stack.yaml",
):
    """
    Execute remote job or DLT pipeline and monitor failures until completion.

    Parameters
    ----------
    databricks_job:
        Name of the job to run (mutually exclusive with dlt)
    databricks_pipeline:
        Name of the DLT pipeline to run (mutually exclusive with job)
    timeout:
        Maximum allowed time (in seconds) for run.
    raise_exception:
        Raise exception on failure
    current_run_action:
        Action to take for currently running job or pipline.
    full_refresh:
        Full tables refresh (pipline only)
    environment:
        Name of the environment.
    filepath:
        Stack (yaml) filepath.

    Examples
    --------
    ```cmd
    laktory run --env dev --dbks-pipeline pl-stock-prices --full_refresh --action CANCEL
    ```

    References
    ----------
    * [CLI](https://www.laktory.ai/concepts/cli/)

    """

    # Set Resource Name
    if databricks_job and databricks_pipeline:
        raise ValueError("Only one of `job` or `dlt` should be set.")
    if not (databricks_job or databricks_pipeline):
        raise ValueError("One of `job` or `dlt` should be set.")

    # Set Dispatcher
    controller = CLIController(
        env=environment,
        stack_filepath=filepath,
    )
    dispatcher = Dispatcher(stack=controller.stack)
    dispatcher.get_resource_ids()

    if databricks_job:
        dispatcher.run_databricks_job(
            job_name=databricks_job,
            timeout=timeout,
            raise_exception=raise_exception,
            current_run_action=current_run_action,
        )

    if databricks_pipeline:
        dispatcher.run_databricks_dlt(
            dlt_name=databricks_pipeline,
            timeout=timeout,
            raise_exception=raise_exception,
            current_run_action=current_run_action,
            full_refresh=full_refresh,
        )