Metadata-Version: 2.4
Name: anaconda-cli-base
Version: 0.9.1
Summary: A base CLI entrypoint supporting Anaconda CLI plugins
License: BSD-3-Clause
License-File: LICENSE
Requires-Python: >=3.10
Requires-Dist: click
Requires-Dist: packaging>=23.0
Requires-Dist: pydantic-settings>=2.3
Requires-Dist: pydantic>=2.12
Requires-Dist: readchar
Requires-Dist: rich
Requires-Dist: tomli
Requires-Dist: tomlkit
Requires-Dist: typer>=0.17
Provides-Extra: dev
Requires-Dist: mypy; extra == 'dev'
Requires-Dist: oteltest; extra == 'dev'
Requires-Dist: pytest; extra == 'dev'
Requires-Dist: pytest-cov; extra == 'dev'
Requires-Dist: pytest-mock; extra == 'dev'
Requires-Dist: tox; extra == 'dev'
Requires-Dist: types-requests; extra == 'dev'
Provides-Extra: publish
Requires-Dist: build; extra == 'publish'
Requires-Dist: twine; extra == 'publish'
Requires-Dist: wheel; extra == 'publish'
Provides-Extra: telemetry
Requires-Dist: anaconda-opentelemetry>=1.2.0; extra == 'telemetry'
Description-Content-Type: text/markdown

# anaconda-cli-base

A base CLI entrypoint supporting Anaconda CLI plugins using [Typer](https://github.com/fastapi/typer).

## Telemetry

The CLI automatically reports command execution metrics for all registered plugins via
[anaconda-opentelemetry](https://github.com/anaconda/anaconda-otel-python). Every command invocation
records the command name, execution duration, and success/failure status. No plugin changes are required.

### Disabling telemetry

Telemetry is enabled by default. To disable:

```bash
export ANACONDA_TELEMETRY_ENABLED=false
```

Or in `~/.anaconda/config.toml`:

```toml
[telemetry]
enabled = false
```

### Configuration

Telemetry settings live in the `[telemetry]` section of `~/.anaconda/config.toml` or
as environment variables with the `ANACONDA_TELEMETRY_` prefix.

| Setting | Env Variable | Default | Description |
|---------|-------------|---------|-------------|
| `enabled` | `ANACONDA_TELEMETRY_ENABLED` | `true` | Enable or disable all CLI telemetry |
| `endpoint` | `ANACONDA_TELEMETRY_ENDPOINT` | `None` | Set a custom OTEL endpoint ur. If `None` uses anaconda.com endpoint |
| `share_session_identity` | `ANACONDA_TELEMETRY_SHARE_SESSION_IDENTITY` | `true` | Include anonymous session tokens for usage correlation |
| `proxy_url` | `ANACONDA_TELEMETRY_PROXY_URL` | None | HTTP proxy for telemetry export (for corporate networks) |
| `flush_timeout_ms` | `ANACONDA_TELEMETRY_FLUSH_TIMEOUT_MS` | `500` | Max milliseconds to wait for telemetry flush on CLI exit |
| `export_interval_ms` | `ANACONDA_TELEMETRY_EXPORT_INTERVAL_MS` | `60000` | Millisecond frequency over which data is exported for long-running tasks |

When `share_session_identity` is `true`, hashed machine and session tokens are included with telemetry
data. These allow Anaconda to correlate usage patterns across CLI sessions without identifying you personally.
Set to `false` to send only standalone metrics with no session linking.

## Registering plugins

To develop a subcommand in a third-party package, first create a `typer.Typer()` app with one or more commands.
See [this example](https://typer.tiangolo.com/#example-upgrade). The commands defined in your package will be prefixed
with the *subcommand* you define when you register the plugin.

In your `pyproject.toml` subcommands can be registered as follows:

```toml
# In pyproject.toml

[project.entry-points."anaconda_cli.subcommand"]
auth = "anaconda_auth.cli:app"
```

In the example above:

* `"anaconda_cli.subcommand"` is the required string to use for registration. The quotes are important.
* `auth` is the name of the new subcommand, i.e. `anaconda auth`
  * All `typer.Typer` commands you define in your package are accessible the registered subcommand
  * i.e. `anaconda auth <command>`.
* `anaconda_auth.cli:app` signifies the object named `app` in the `anaconda_auth.cli` module is the entry point for the subcommand.

### Error handling

By default any exception raised during CLI execution in your registered plugin will be caught and only a minimal
message will be displayed to the user.

You can define a custom callback for individual exceptions that may be thrown from your subcommand. You can
register handlers for standard library exceptions or custom defined exceptions. It may be best to use custom
exceptions to avoid unintended consequences for other plugins.

To register the callback decorate a function that takes an exception as input, and return an integer error code.
The error code will be sent back through the CLI and your subcommand will exit with that error code.

```python
from typing import Type
from anaconda_cli_base.exceptions import register_error_handler

@register_error_handler(MyCustomException)
def better_exception_handling(e: Type[Exception]) -> int:
    # do something or print useful information
    return 1

@register_error_handler(AnotherException)
def just_ignore_it(e: Type[Exception])
    # ignore the error and let the CLI exit successfully
    return 0


@register_error_handler(YetAnotherException)
def fix_the_error_and_try_again(e: Type[Exception]) -> int:
    # do something and retry the CLI command
    return -1
```

In the second example the handler returns `-1`. This means that the handler has attempted to correct the error
and the CLI subcommand should be re-tried. The handler could call another interactive command, like a login action,
before attempting the CLI subcommand again.

See the [anaconda-auth](https://github.com/anaconda/anaconda-auth/blob/main/src/anaconda_auth/cli.py) plugin for an example custom handler.

### Config file

If your plugin wants to utilize the Anaconda config file, default location `~/.anaconda/config.toml`, to read configuration
parameters you can derive from `anaconda_cli_base.config.AnacondaBaseSettings` to add a section in the config file for
your plugin.
 Each subclass of `AnacondaBaseSettings`
defines the section header. The base class is configured so that parameters defined in subclasses can be read in the
following priority from lowest to highest.

1. default value in the subclass of `AnacondaBaseSettings`
1. Global config file at ~/.anaconda/config.toml
1. `ANACONDA_<PLUGIN-NAME>_<FIELD>` variables defined in the .env file in your working directory
1. A file named `/run/secrets/anaconda_<plugin-name>_<field>`, usually populated by a mounted
   [Docker secret](https://docs.docker.com/engine/swarm/secrets/)
1. `ANACONDA_<PLUGIN-NAME>_<FIELD>` env variables set in your shell or on command invocation
1. value passed as kwarg when using the config subclass directly

Notes:

* `AnacondaBaseSettings` is a subclass of `BaseSettings` from [pydantic-settings](https://docs.pydantic.dev/latest/concepts/pydantic_settings/#usage).
* Nested pydantic models are also supported.
* Per pydantic defaults, both secret filenames and environment variables
  may be uppercase or lowercase.

Here's an example subclass:

```python
from anaconda_cli_base.config import AnacondaBaseSettings

class MyPluginConfig(AnacondaBaseSettings, plugin_name="my_plugin"):
    foo: str = "bar"
```

To read the config value in your plugin according to the above
priority:

```python
config = MyPluginConfig()
assert config.foo == "bar"
```

Since there is no value of `foo` in the config file it assumes the default value from the subclass definition.

The value of `foo` can now be written to the config file under the section `my_plugin`

```toml
# ~/.anaconda/config.toml
[plugin.my_plugin]
foo = "baz"
```

Now that the config file has been written, the value of `foo` is read from the
config.toml file:

```python
config = MyPluginConfig()
assert config.foo == "baz"
```

### Nested tables

The AnacondaBaseSettings supports nested Pydantic models.

```python
from anaconda_cli_base.config import AnacondaBaseSettings
from pydantic import BaseModel

class Nested(BaseModel):
    n1: int = 0
    n2: int = 0

class MyPluginConfig(AnacondaBaseSettings, plugin_name="my_plugin"):
    foo: str = "bar"
    nested: Nested = Nested()
```

In the `~/.anaconda/config.toml` you can set values of nested fields as an in-line table

```toml
# ~/.anaconda/config.toml
[plugin.my_plugin]
foo = "baz"
nested = { n1 = 1, n2 = 2}
```

Or as a separate table entry

```toml
# ~/.anaconda/config.toml
[plugin.my_plugin]
foo = "baz"

[plugin.my_plugin.nested]
n1 = 1
n2 = 2
```

To set environment variables use the `__` delimiter

```bash
ANACONDA_MY_PLUGIN_NESTED__N1=1
ANACONDA_MY_PLUGIN_NESTED__N2=2
```

### Nested plugins

You can pass a tuple to `plugin_name=` in subclasses of `AnacondaBaseSettings` to nest whole plugins,
which may be defined in separate packages.

```python
class Nested(BaseModel):
    n1: int = 0
    n2: int = 0
class MyPluginConfig(AnacondaBaseSettings, plugin_name="my_plugin"):
    foo: str = "bar"
    nested: Nested = Nested()
```

Then in another package you can nest a new config into `my_plugin`.

```python
class MyPluginExtrasConfig(AnacondaBaseSettings, plugin_name=("my_plugin", "extras")):
    field: str = "default"
```

The new config table is now nested in the config.toml

```toml
# ~/.anaconda/config.toml
[plugin.my_plugin]
foo = "baz"
nested = { n1 = 1, n2 = 2}
[plugin.my_plugin.extras]
field = "value"
```

And can be set by env variable using the concatenation of `plugin_name`

```bash
ANACONDA_MY_PLUGIN_EXTRAS_FIELD="value"
```

### Writing configuration

Plugin configurations can be written directly from subclasses of `AnacondaBaseSettings` with the
`.write_config()` member method. This method takes two arguments

* `preserve_existing_keys`:
  * If True (default) updates to existing keys in the
    config.toml file, will not remove the key if set to the default
    value. If False fields set to default value are removed from the file
* `dry_run`:
  * If True, displays a diff of proposed changes without writing
    to the file. If False (default), writes changes to config.toml.

Here are some key aspects of writing configuration

* `.write_config()` will only update changed lines in the config.toml preserving all existing configuration and comments
* toml does not support `None` or `null`, any field set to the value `None` will not be written to the config.toml
* fields set to their default value are not written to the config.toml
  * Except when an existing key in the config.toml is updated to its default value. The key will still be written
  * This is disabled with `preserve_existing_keys=False`

Let's start with the plugin defined earlier and an instance of the config object with all default values

```python
from anaconda_cli_base.config import AnacondaBaseSettings
from pydantic import BaseModel

class Nested(BaseModel):
    n1: int = 0
    n2: int = 0

class MyPluginConfig(AnacondaBaseSettings, plugin_name="my_plugin"):
    foo: str = "bar"
    nested: Nested = Nested()


config = MyPluginConfig()
```

If there is either no config.toml or the existing file does not have the `[plugin.my_plugin]` table attempting
to write the current state of the config will just add the table header since all values are default. Here is an
example of the `dry_run` output in the case where the config.toml file did not exist

```text
>>> config.write_config(dry_run=True)
--- ~/.anaconda/config.toml
+++ ~/.anaconda/config.toml 01-06-26 09:40
@@ -0,0 +1 @@
+[plugin.my_plugin]
```

You can change the configuration either by passing kwargs to the initialization or by directly updating attributes.

```python
config.foo = "baz"
config.nested.n1 = 1
config.nested.n2 = 2
```

this will now write the configuration equivalent to what you saw above

```text
>>> config.write_config(dry_run=True)
--- ~/.anaconda/config.toml
+++ ~/.anaconda/config.toml 01-06-26 09:44
@@ -0,0 +1,6 @@
+[plugin.my_plugin]
+foo = "baz"
+
+[plugin.my_plugin.nested]
+n1 = 1
+n2 = 2
```

Now with that configuration written to disk (using `dry_run=False`) we can re-read the configuration to confirm
the change.

```text
>>> config = MyPluginConfig()
>>> print(config)
foo='baz' nested=Nested(n1=1, n2=2)
```

Let's change `foo` back to its default value. We can do that either by setting the attribute `config.foo = "bar"` or
by passing a kwarg to override the config.toml.

The dry-run output now only changes the `foo` key in the config.toml leaving all other lines unchanged

```text
>>> config = MyPluginConfig(foo="bar")
>>> config.write_config(dry_run=True)
--- ~/.anaconda/config.toml 01-06-26 09:53
+++ ~/.anaconda/config.toml 01-06-26 09:56
@@ -1,5 +1,5 @@
 [plugin.my_plugin]
-foo = "baz"
+foo = "bar"

 [plugin.my_plugin.nested]
 n1 = 1
```

If instead we wish to remove keys when set to their default value pass the `preserve_existing_keys=False` argument

```text
>>> config.write_config(dry_run=True, preserve_existing_keys=False)
--- ~/.anaconda/config.toml 01-06-26 09:53
+++ ~/.anaconda/config.toml 01-06-26 09:57
@@ -1,5 +1,4 @@
 [plugin.my_plugin]
-foo = "baz"

 [plugin.my_plugin.nested]
 n1 = 1
```

See the [tests](https://github.com/anaconda/anaconda-cli-base/blob/main/tests/test_config.py) for more examples of reading and writing plugin configuration.

### Plugin telemetry

Plugins get baseline command metrics for free. To add custom instrumentation:

```python
from anaconda_cli_base.telemetry import traced, count, histogram, log_event

@app.command()
def download(model: str):
    with traced("models_download", plugin_name="ai", attributes={"model": model}) as span:
        result = do_download(model)
        span.add_event("download_complete", {"size_bytes": result.size})
    count("models_downloaded", plugin_name="ai")
    histogram("download_size_bytes", plugin_name="ai", value=result.size)
    log_event("user downloaded a model", event_name="model_downloaded", plugin_name="ai", attributes={"model": model})
```

The `plugin_name` should match your registered subcommand name (e.g., `"ai"` for `anaconda ai`).
This ensures custom telemetry correlates with the automatic command metrics in dashboards.

All functions are no-ops when telemetry is disabled — they will never raise or affect CLI behavior.

### Logging handler

For error and warning capture via Python's standard `logging` module, attach the OTel handler
to your plugin's logger. By default log records at WARNING and above are exported to the telemetry backend
while still flowing to any other handlers (stderr, file) you have configured.

```python
import logging
from anaconda_cli_base.telemetry import get_otel_handler

log = logging.getLogger("anaconda_ai")
log.addHandler(get_otel_handler())

# WARNING+ goes to OTel; all levels still go to other handlers
log.warning("retry attempt", extra={"attempt": 3, "endpoint": url})
log.error("download failed", extra={"model": model, "error.type": "TimeoutError"})
```

Pass a custom level to change the threshold:

```python
log.addHandler(get_otel_handler(level=logging.ERROR))  # Only errors
```

When telemetry is disabled or `anaconda-opentelemetry` is not installed, `get_otel_handler()`
returns a `NullHandler` — safe to call unconditionally with zero overhead.

Use `get_otel_handler()` for structured errors/warnings. Use `log_event()` for business events
that shouldn't appear in developer console output (e.g., `"model_downloaded"`, `"session_started"`).

### Long-running commands

Commands that block indefinitely (e.g., servers) need explicit lifecycle management
so the process exits cleanly on SIGTERM/SIGINT without hanging on telemetry flush.

```python
from anaconda_cli_base.lifecycle import long_running

@app.command()
@long_running
def serve():
    asyncio.run(run_server())
```

The `@long_running` decorator:
- Installs SIGTERM/SIGINT handlers that trigger a bounded shutdown sequence
- Starts a watchdog timer (`WATCHDOG_DEADLINE_SECS = 10`) that force-exits the process if shutdown stalls
- Calls `shutdown_telemetry(timeout_seconds=2.0)` to flush telemetry within a hard time bound

Shutdown hooks are optional, but a blocking server usually wants one. On shutdown the
decorator flushes telemetry and arms the watchdog, but it does not stop your loop: SIGINT
unblocks `asyncio.run(...)` on its own (it raises `KeyboardInterrupt`), whereas SIGTERM does
not. Without a hook to stop the loop, a SIGTERM'd server keeps running until the 10s watchdog
force-exits it — no hang, but neither prompt nor graceful. Register a hook to unblock the loop
(close a connection, unblock a stdin reader, signal the loop to stop); hooks run before the
telemetry flush:

```python
from anaconda_cli_base.lifecycle import long_running, register_shutdown_hook

@app.command()
@long_running
def serve():
    register_shutdown_hook(cleanup_resources)
    asyncio.run(run_server())
```

For signal handlers or manual shutdown paths, use `shutdown_telemetry` directly:

```python
from anaconda_cli_base.telemetry import shutdown_telemetry

# In a signal handler or cleanup path:
shutdown_telemetry(timeout_seconds=2.0)
```

Short-lived commands (the common case) need no changes — telemetry is flushed
automatically via `_after_command` on every normal CLI exit. The `flush_timeout_ms`
config (default 500ms) controls the per-command flush bound.

## Setup for development

Ensure you have `conda` installed.
Then run:

```shell
make setup
```

### Run the unit tests

```shell
make test
```

### Run the unit tests across isolated environments with tox

```shell
make tox
```
