Reference

dp package

Submodules

dp.annotations module

check_version(f)

Annotation to check for newer versions from PyPI.

This annotation checks for newer versions from PyPI and prints a message if a newer version is available. The check is performed once every 24 hours. You can disable this check by setting the environment variable DAPLA_CLI_NO_VERSION_CHECK to any value.

Return type:

Callable[..., Any]

Parameters:

f (Callable[[...], Any])

dryrunnable(f)

Decorator that prints an informative message if a command is running in dryrun mode.

Annotate functions with this decorator to print a message if the function is running in dryrun mode. Expects ‘dryrun’ boolean argument to be passed to the function.

Return type:

Callable[..., Any]

Parameters:

f (Callable[[...], Any])

ensure_helm_repos_updated(f)

Annotation to ensure Helm repo is updated if necessary before running a helm command that relies on helm repos.

Return type:

Callable[..., Any]

Parameters:

f (Callable[[...], Any])

dp.auth module

class Env(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Denotes the environment to operate on.

dev = 'dev'
prod = 'prod'
test = 'test'
local_access_token(env, ensure_valid=True)

Return the access token stored in the local configuration.

If the token is not found, the user is prompted to log in.

Parameters:

  • env (Env) – The environment to get the access token for.

  • ensure_valid (bool) – If True, the token is refreshed (if needed) before returning it.

Returns:

A valid JWT access token.

Return type:

str

Raises:

Exit – If no access token is found, prompts the user to log in and exits.

login(env=Env.prod)

Log in to Keycloak.

Return type:

None

Parameters:

env (Annotated[Env, <typer.models.OptionInfo object at 0x7f7d3f692720>])

logout(env=Env.prod)

Log out of Keycloak.

Return type:

None

Parameters:

env (Annotated[Env, <typer.models.OptionInfo object at 0x7f7d3f692720>])

show_access_token(env=Env.prod, decoded=False, to_clipboard=False)

Print the local access token (if logged in).

Return type:

None

Parameters:

  • env (Annotated[Env, <typer.models.OptionInfo object at 0x7f7d3f692720>])

  • decoded (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3e67bce0>])

  • to_clipboard (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3e67b9b0>])

dp.config module

get(section, key, namespace)

Retrieve the config value of a key in a section.

Return type:

Any

Parameters:

  • section (str)

  • key (str)

  • namespace (str | None)

put(section, key, value, namespace)

Set a config value for a key in a section.

Return type:

None

Parameters:

  • section (str)

  • key (str)

  • value (Any)

  • namespace (str | None)

remove(section, namespace)

Remove a section from the config.

Return type:

None

Parameters:

  • section (str)

  • namespace (str | None)

dp.lab module

class Env(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Denotes the environment to operate on.

dev = 'dev'
prod = 'prod'
test = 'test'
class OperationType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Denotes the operation type to perform on services when processing services.

kill = 'kill'
prune = 'prune'
suspend = 'suspend'
unsuspend = 'unsuspend'
class Service(**data)

Bases: BaseModel

A Dapla Lab service.

Parameters:

  • name (str)

  • namespace (str)

  • revision (str | None)

  • updated (datetime | None)

  • status (str | None)

  • suspended (bool | None)

  • chart (str | None)

  • app_version (str | None)

  • chart_version (str | None)

  • created (datetime | None)

app_version: str | None
chart: str | None
chart_version: str | None
created: datetime | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str
namespace: str
revision: str | None
status: str | None
suspended: bool | None
updated: datetime | None
add_chart_repos(verbose=False)

Add required dapla-lab service helm chart repositories.

The repositories must be added before we can modify helm releases (such as suspending services).

Return type:

None

Parameters:

verbose (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3f56f590>])

doctor()

Check if required tooling and environment is ok.

Return type:

None

kill_service(service_name, env, namespace, dryrun=False, verbose=False)

Kill a service in the specified namespace.

Return type:

None

Parameters:

  • service_name (str)

  • env (Annotated[Env, <typer.models.OptionInfo object at 0x7f7d3f53f470>])

  • namespace (Annotated[str, <typer.models.OptionInfo object at 0x7f7d3e257290>])

  • dryrun (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3e256a80>])

  • verbose (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3f56f590>])

kill_services(env, namespace, dryrun=False, verbose=False)

Kill all services in the specified namespace.

If the namespace is set to ‘all’, all user namespaces will be affected.

Return type:

None

Parameters:

  • env (Annotated[Env, <typer.models.OptionInfo object at 0x7f7d3f53f470>])

  • namespace (Annotated[str, <typer.models.OptionInfo object at 0x7f7d3e257290>])

  • dryrun (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3e256a80>])

  • verbose (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3f56f590>])

list_services(env, namespace, verbose=False)

List all services in the specified namespace.

Return type:

None

Parameters:

  • env (Annotated[Env, <typer.models.OptionInfo object at 0x7f7d3f53f470>])

  • namespace (Annotated[str, <typer.models.OptionInfo object at 0x7f7d3e257290>])

  • verbose (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3f56f590>])

prune_services(env, namespace, dryrun=False, verbose=False)

Prune services.

Return type:

None

Parameters:

  • env (Annotated[Env, <typer.models.OptionInfo object at 0x7f7d3f53f470>])

  • namespace (Annotated[str, <typer.models.OptionInfo object at 0x7f7d3e257290>])

  • dryrun (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3e256a80>])

  • verbose (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3f56f590>])

suspend_service(service_name, env, namespace, dryrun=False, verbose=False)

Suspend a service in the specified namespace.

Return type:

None

Parameters:

  • service_name (str)

  • env (Annotated[Env, <typer.models.OptionInfo object at 0x7f7d3f53f470>])

  • namespace (Annotated[str, <typer.models.OptionInfo object at 0x7f7d3e257290>])

  • dryrun (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3e256a80>])

  • verbose (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3f56f590>])

suspend_services(env, namespace, dryrun=False, verbose=False)

Suspend user services.

All services in the specified namespace will be suspended or unsuspended. If the namespace is set to ‘all’, all user namespaces will be affected.

Return type:

None

Parameters:

  • env (Annotated[Env, <typer.models.OptionInfo object at 0x7f7d3f53f470>])

  • namespace (Annotated[str, <typer.models.OptionInfo object at 0x7f7d3e257290>])

  • dryrun (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3e256a80>])

  • verbose (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3f56f590>])

unsuspend_services(env, namespace, dryrun=False, verbose=False)

Unsuspend user services.

All services in the specified namespace will be unsuspended. If the namespace is set to ‘all’, all user namespaces will be affected.

Return type:

None

Parameters:

  • env (Annotated[Env, <typer.models.OptionInfo object at 0x7f7d3f53f470>])

  • namespace (Annotated[str, <typer.models.OptionInfo object at 0x7f7d3e257290>])

  • dryrun (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3e256a80>])

  • verbose (Annotated[bool, <typer.models.OptionInfo object at 0x7f7d3f56f590>])

dp.main module

main(version=None)

Entrypoint for the Dapla CLI.

Return type:

None

Parameters:

version (Annotated[bool | None, <typer.models.OptionInfo object at 0x7f7d3e1257f0>])

version_callback(value)

Print the version.

Return type:

None

Parameters:

value (bool)

dp.utils module

class RunResult(**data)

Bases: BaseModel

The result of running a shell command.

Parameters:

  • stdout (str)

  • stderr (str)

  • returncode (int)

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

returncode: int
stderr: str
stdout: str
assert_successful_command(cmd, err_msg, success_msg)

Run a shell command and assert its success.

Parameters:

  • cmd (str) – The shell command to run.

  • err_msg (str) – The error message to display if the command fails.

  • success_msg (str) – The success message to display if the command succeeds.

Raises:

Exit – If the command fails.

Return type:

None

get_current_version()

Return the app version.

Return type:

Version | None

get_latest_pypi_version()

Fetches the latest version of a package from PyPI.

Returns:

The latest version of the package, or None if there is an error.

Return type:

str

green(text)

Returns text colored in green.

Return type:

str

Parameters:

text (Any)

grey(text)

Returns text colored in gray.

Return type:

str

Parameters:

text (Any)

hours_since(dt)

Calculate the number of hours since a given datetime.

Return type:

int

Parameters:

dt (datetime)

print_err(text)

Prints to standard error stream.

Return type:

None

Parameters:

text (Any)

red(text)

Returns text colored in red.

Return type:

str

Parameters:

text (Any)

run(command, dryrun=False, verbose=False)

Run a shell command.

Parameters:

  • command (str) – The shell command to run.

  • dryrun (bool) – Whether to perform a dry run.

  • verbose (bool) – Whether to print the command.

Returns:

The result of the command.

Return type:

RunResult

strip_ansi(text)

Strip ANSI escape sequences from text.

Return type:

str

Parameters:

text (str)

Module contents

Dapla CLI.