Metadata-Version: 2.4
Name: pysae-cli-tools
Version: 0.1.46
Summary: Reusable utilities for Pysae Python CLIs (k8s pod dispatch, …).
Project-URL: homepage, https://gitlab.com/pysae/tools/cli-tools
Project-URL: repository, https://gitlab.com/pysae/tools/cli-tools
Author-email: Rémi Alvergnat <remi.alvergnat@pysae.com>
License-Expression: MIT
Requires-Python: <4,>=3.11
Requires-Dist: typer<1.0,>=0.16
Description-Content-Type: text/markdown

# pysae-cli-tools

Reusable utilities for Pysae Python CLIs.

[![PyPI](https://img.shields.io/pypi/v/pysae-cli-tools.svg)](https://pypi.org/project/pysae-cli-tools/)

## Installation

```bash
pip install pysae-cli-tools
# or
uv 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:

```python
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", "uv.lock"),
    install_script=(
        "pip install uv && "
        "uv sync --frozen --no-dev"
    ),
    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:

```python
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:

```python
@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:

```python
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 uv 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.

```python
K8S_CONFIG = K8sConfig(
    default_image=resolve_image,
    project_root=Path(__file__).resolve().parents[1],
    local_sources=("my_pkg", "tooling", "pyproject.toml", "uv.lock"),
    install_script="uv sync --frozen --no-dev",
    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.

#### Resolving local env vars for the Typer parse (`resolve_env`)

`env_secret_bindings` decides what the **pod** receives. A separate helper,
`resolve_env`, decides how the **operator's** `os.environ` is seeded *before
Typer parses argv* — so a required `Argument(envvar="X")` resolves even when `X`
is not exported locally. This is what lets a `k8s:secret:mount:…` binding (never
materialised on the operator side) back a required argument: the mount serves
the pod, `resolve_env` seeds the local parse.

Call it at the top of the module that declares the Typer app (or inside the
entry point), before `app()` runs. It takes **bare resolver functions**, each
`(environment: str) -> Mapping[str, str]`, tried in order. It runs on every
invocation except `--help`.

Per variable: an absent one is seeded; a value already handled by an earlier
resolver is left as-is; a value **already present with the same value** is left
as-is; a value **present with a different value** prompts the operator (keep
current vs use resolved, **defaulting to the resolved value**) rather than
silently overwriting — and **errors when there is no TTY** to ask. The two values
are shown with URL passwords masked (`user:***@host`).

A seeded (or accepted) value then lives in `os.environ` and is **indistinguishable
from a raw operator export**: it reaches the pod and is merged there by
`secret_merge` exactly like a user-defined one (there is no "resolved-by-tooling"
special case).

The target environment is detected from argv **independently of the `--k8s`
flags**: a bare token equal to one of `environments`' values (`dev` / `prod` /
`testing` by default, see `ResolveEnvironment`) selects it — so `mycli restore
prod` resolves `prod` — otherwise the **first enum member** is the default
(`dev`). The detected value is passed to each resolver.

The built-in `pysae_ai_tools_env_resolver` loads the project's whitelisted vars
(from `.pysae-ai-tools.yaml`) via `pysae-ai-tools env activate <environment>`:

```python
import typer
from pysae_cli_tools.k8s import pysae_ai_tools_env_resolver, resolve_env

app = typer.Typer()

@app.command()
def restore(mongo_uri: str = typer.Argument(envvar="MONGO_URI")): ...

# fire-and-forget: seed once, fine for a one-shot CLI that exits
resolve_env(pysae_ai_tools_env_resolver)
```

`resolve_env` returns an **optional context manager**: wrap the app invocation
in it to restore the original environment once the command completes (recommended
when the CLI module is imported into a longer-lived process, e.g. tests) —
otherwise ignore the return value.

```python
def main() -> None:
    with resolve_env(pysae_ai_tools_env_resolver):
        app()
```

Projects with a custom environments enum pass it explicitly:
`resolve_env(pysae_ai_tools_env_resolver, environments=MyEnv)` — the enum's
first member is the default. Write your own resolver for other backends — any
`(environment) -> Mapping[str, str]` returning the subset it could resolve
(best-effort; unresolved names fall through to the next resolver, then to Typer's
own error).

##### `EnhancedTyper` — the automatic form

To avoid the top-of-module call or the `main()` wrapper entirely, declare the
app with `EnhancedTyper` (a `typer.Typer` subclass). It seeds before Click parses
and restores after every invocation, so the console-script entry point stays
`app`:

```python
import typer
from pysae_cli_tools.k8s import EnhancedTyper, pysae_ai_tools_env_resolver

app = EnhancedTyper(resolvers=[pysae_ai_tools_env_resolver])

@app.command()
def restore(mongo_uri: str = typer.Argument(envvar="MONGO_URI")): ...
```

It takes the same `environments` knob as `resolve_env` plus all standard
`typer.Typer` options. Use `resolve_env` directly when you don't own the app (a
plain `typer.Typer`); both forms are supported.

Unless `show_env=False`, every invocation also echoes the **Typer-managed** env
vars — those declared via `Argument/Option(envvar=…)` across the commands — to
stderr, both locally and at pod startup (the pod runs the same app). Values are
shown in clear except that a **URL password is always masked**. Flag opaque
secrets (tokens) via `sensitive=[…]` to hide their whole value; a flagged URL
still shows its structure (only the password is masked):

```python
app = EnhancedTyper(
    resolvers=[pysae_ai_tools_env_resolver],
    sensitive=["API_TOKEN"],
)
```

```
[env] Typer-managed environment variables:
  MONGO_URI=mongodb+srv://user:***@dev.host/db
  API_TOKEN=***
  LOG_LEVEL=debug
```

`sensitive` is honoured wherever a value is printed (this echo and the
`resolve_env` conflict prompt).

Inside the ephemeral pod, resolution is **skipped** (everything is already
injected by the dispatcher, so re-resolving would be redundant and could raise
on a TTY-less conflict), and the echo additionally masks every secret-sourced
var — inline bindings *and* `secretKeyRef` mounts — so a mounted secret is never
printed in clear, even without listing it in `sensitive`.

Both forms detect the environment from `sys.argv` (a bare token matching an enum
value, else the first member) and seed on every invocation but `--help`. The
seeded values feed a required `Argument(envvar=…)` at parse time and, for a
`@k8s_support` command, the pod forwarding — but resolution itself no longer
depends on the `--k8s` flags.

#### 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.

##### Env-resolution traces

Each variable's origin and merge is traced to stderr as `[env] …`, so the
operator can see where every value comes from and how it was reconciled — the
source reference and strategy are shown, **never the resolved value**:

```
[env] DD_API_KEY: resolved from the k8s secret prod/prod-secrets:dd-api-key (before argv parse)
[env] MONGO_URI: resolved locally by pysae_ai_tools_env_resolver (env=prod)
[env] MONGO_URI: local value merged with the k8s mount prod-secrets:api-mongo-uri in the pod (prefer_local_auth_only)
[env] DD_API_KEY: using the environment value, the k8s secret prod/prod-secrets:dd-api-key not re-fetched (prefer_local)
[env] CACHE_URL: injected from the k8s mount prod-secrets:cache-url via secretKeyRef
```

Inline `k8s:secret:`/`aws:secret:` bindings are fetched **before Typer parses**
(to seed a required `Argument(envvar=…)`) — their origin is traced there with a
`(before argv parse)` suffix; the pod-env build then just forwards the value.

See [`pysae_cli_tools/k8s/config.py`](pysae_cli_tools/k8s/config.py) for the
complete `K8sConfig` reference.

## Development

```bash
uv sync
uv run pre-commit install
uv run pytest
```

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