CLI

laktory.cli.app

---

laktory.cli.preview

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 pulumi_options "--show-reads,--show-config"

References

• pulumi preview
• terraform preview

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 pulumi_options "--show-reads,--show-config"
    ```

    References
    ----------
    - pulumi [preview](https://www.pulumi.com/docs/cli/commands/pulumi_preview/)
    - terraform [preview](https://developer.hashicorp.com/terraform/cli/commands/plan)
    """
    controller = CLIController(
        env=environment,
        stack_filepath=filepath,
        options_str=options,
    )

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

---

laktory.cli.deploy

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

• pulumi up
• 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
    ----------
    - pulumi [up](https://www.pulumi.com/docs/cli/commands/pulumi_up/)
    - 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.backend == "pulumi":
        controller.pulumi_call("up")
    elif controller.backend == "terraform":
        controller.terraform_call("apply")
    else:
        raise ValueError(f"backend should be {SUPPORTED_BACKENDS}")

---

laktory.cli.run

run(job=None, dlt=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
job	Name of the job to run (mutually exclusive with dlt) TYPE: Annotated[str, Option(--job, -j, help='Job name')] DEFAULT: None
dlt	Name of the DLT pipeline to run (mutually exclusive with job) TYPE: Annotated[str, Option(--dlt, -dlt, 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 --dlt pl-stock-prices --full_refresh --action CANCEL

Source code in laktory/cli/_run.py

@app.command()
def run(
    job: Annotated[str, typer.Option("--job", "-j", help="Job name")] = None,
    dlt: Annotated[str, typer.Option("--dlt", "-dlt", 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
    ----------
    job:
        Name of the job to run (mutually exclusive with dlt)
    dlt:
        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 --dlt pl-stock-prices --full_refresh --action CANCEL
    ```

    """

    # Set Resource Name
    if job and dlt:
        raise ValueError("Only one of `job` or `dlt` should be set.")
    if not (job or dlt):
        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 job:
        dispatcher.run_job(
            job_name=job,
            timeout=timeout,
            raise_exception=raise_exception,
            current_run_action=current_run_action,
        )

    if dlt:
        dispatcher.run_dlt(
            dlt_name=dlt,
            timeout=timeout,
            raise_exception=raise_exception,
            current_run_action=current_run_action,
            full_refresh=full_refresh,
        )