pysae-cli-tools 0.1.24

Reusable utilities for Pysae Python CLIs (k8s pod dispatch, …).

pysae-cli-tools

Reusable utilities for Pysae Python CLIs.

Installation

pip install pysae-cli-tools
# or
poetry add pysae-cli-tools

What's included

pysae_cli_tools.k8s — run any Typer command in an ephemeral pod

The @k8s_support decorator injects three flags into a Typer command — --k8s / --no-k8s, --k8s-environment {dev|prod}, --k8s-from-local-sources — and dispatches the call into a freshly-spawned Kubernetes pod when --k8s is set. A command can default to k8s with @k8s_support(default_k8s=True), in which case it dispatches unless --no-k8s is passed (see below).

It is meant for CLIs that need to run inside the same network as their target infrastructure (private-link databases, VPC-only APIs, …) without rewriting the command for kubectl run.

Usage with build_k8s_support (recommended)

Most projects share the same K8sConfig across every decorated command — declare it once and reuse the bound decorator everywhere:

from pathlib import Path

from typer import Typer

from pysae_cli_tools.k8s import K8sConfig, build_k8s_support

K8S_CONFIG = K8sConfig(
    default_image="<registry>/<project>:latest",
    project_root=Path(__file__).resolve().parents[1],
    local_sources=("my_pkg", "pyproject.toml", "poetry.lock"),
    install_script=(
        "apt-get update -qq && pip install poetry && "
        "poetry config virtualenvs.create false && "
        "poetry install --only main --no-interaction"
    ),
    forwarded_envvars=("MY_API_KEY", "MY_DB_URI"),
    redacted_options=("--api-key", "--password"),
    env_secret_bindings={
        "dev": {"MONGO_URI": "k8s:secret:dev/dev-secrets:api-mongo-uri"},
        "prod": {"MONGO_URI": "k8s:secret:prod/prod-secrets:api-mongo-uri"},
    },
)

k8s_support = build_k8s_support(K8S_CONFIG)
app = Typer()


@app.command()
@k8s_support()
def my_command() -> None:
    ...


@app.command()
@k8s_support(pod_name_prefix="my-second-command")  # override per-command
def my_second_command() -> None:
    ...

Usage with the explicit form

When you want to use a different config per command, pass it directly:

from pysae_cli_tools.k8s import K8sConfig, k8s_support

@app.command()
@k8s_support(config=K8S_CONFIG)
def my_command() -> None:
    ...

Defaulting to k8s (--no-k8s to opt out)

For commands that should almost always run in a pod (private-link only, long-running, …), flip the default so the operator doesn't have to remember --k8s every time:

@app.command()
@k8s_support(default_k8s=True)
def restore() -> None:
    ...

The injected flag becomes a --k8s/--no-k8s toggle defaulting to True: restore … dispatches to a pod, restore --no-k8s … forces local execution. The decorator suffixes the in-pod invocation with --no-k8s, so the command inside the pod runs its body locally instead of dispatching again. Set default_k8s per command — it composes with pod_name_prefix, tty, etc., and only the commands that opt in get the flipped default.

Dynamic image resolution

K8sConfig.default_image accepts either a literal string (used verbatim for every environment) or a callable (env_value: str) -> str resolved at dispatch time, after --k8s-environment has been parsed:

def resolve_image(env: str) -> str:
    return _get_deployed_image(env)  # e.g. kubectl get deployment/{env}-foo -o jsonpath=...

K8S_CONFIG = K8sConfig(
    default_image=resolve_image,
    project_root=Path(__file__).resolve().parents[1],
)

Use the callable form when the image must mirror what is actually running on the target environment — typically by reading a deployed Kubernetes Deployment via kubectl — so the pod never drifts from the runtime image. The callable is invoked only when --k8s-from-local-sources is not set (the local-sources mode extracts the base image from the Dockerfile).

local_sources vs copy — what lands in the pod and when

Two distinct paths drive what gets copied into the ephemeral pod:

• local_sources is honoured only when --k8s-from-local-sources is set. Use it for the project's source layout (the package, the pyproject.toml, the lock file…) — everything Poetry needs to rebuild the project from scratch in the pod. After the copy, install_script runs (if defined) so the in-pod environment matches the operator's local checkout. This mode is mostly for development.
• copy is honoured in every mode (deployed image and --k8s-from-local-sources). Use it when the deployed image is missing runtime assets that the script needs at import time — typically extra CLI helpers (tooling/, scripts/, …) that live outside the published wheel. The copy happens right after pod spawn, before the script is executed, with no install step. Empty tuple disables it.

K8S_CONFIG = K8sConfig(
    default_image=resolve_image,
    project_root=Path(__file__).resolve().parents[1],
    local_sources=("my_pkg", "tooling", "pyproject.toml", "poetry.lock"),
    install_script="poetry install --only main --no-interaction",
    copy=("tooling",),  # tooling/ is not in the deployed wheel
)

In this example, a --k8s invocation copies tooling/ into the pod (deployed image mode) so python -m tooling.foo resolves, and a --k8s --k8s-from-local-sources invocation copies both the full local_sources set (followed by install_script) and tooling/ again via copy — the second copy is idempotent in practice because local_sources already contains tooling.

Per-environment secret bindings

K8sConfig.env_secret_bindings maps an environment value to a dict of envvar -> value-or-pattern. Values can be:

• a literal string forwarded verbatim into the pod,
• a k8s:secret:[<namespace>/]<secret-name>:<key> reference resolved via kubectl get secret on the operator's machine before the pod is created (base64-decoded automatically),
• a k8s:secret:mount:[<namespace>/]<secret-name>:<key> reference, which materialises as a secretKeyRef entry in the pod spec — the value never transits through the operator's machine, the kubelet reads it directly from the API server. The secret must live in the pod's namespace (no cross-namespace secretKeyRef), and the pod's ServiceAccount must have RBAC get secrets on it. Mount bindings are skipped by the eager-inject hook, so they cannot serve a required Argument(envvar=…) — use Option(envvar=…) with a default, or a non-mount form, when the value must be available during Typer's argv parsing,
• an aws:secret:[<region>:]<secret-id>:<key> reference resolved via aws secretsmanager get-secret-value on the operator's machine, or
• a Callable[[Sequence[str]], str] that receives the filtered argv and returns one of the above forms.

Local environment wins: if the operator already exported the envvar locally, that value is propagated as-is. Kubectl resolution is the fallback, not the override. This matters for two reasons:

1. Argument(envvar="X") in Typer keeps working in both modes — the eager-inject hook seeds os.environ before Typer parses argv.
2. The operator can override a binding for a one-off run without editing the config.

Use forwarded_envvars for simple value-only propagation (no kubectl fallback) and env_secret_bindings whenever you want the convenience of pulling from a Kubernetes secret automatically.

What happens at runtime

When k8s mode is active — --k8s is passed, or default_k8s=True and --no-k8s is not — the decorator:

1. Spawns an ephemeral pod using K8sConfig.default_image (or the Dockerfile base image when --k8s-from-local-sources is also set).
2. Forwards every envvar listed in K8sConfig.forwarded_envvars from your local shell into the pod's env block.
3. Runs python -m <your.cli.module> <subcommand> <filtered argv> inside the pod, with values matching K8sConfig.redacted_options masked in the [K8S] Running: log line.
4. Streams stdout/stderr back to your terminal and deletes the pod on exit.

See pysae_cli_tools/k8s/config.py for the complete K8sConfig reference.

Development

poetry install
poetry run pre-commit install
poetry run pytest

CI publishes a new version to PyPI on every push to main — see .gitlab-ci.yml. The version is computed from git describe via pysae_cli_tools.compute_version.