Metadata-Version: 2.5
Name: adrian-utils
Version: 0.9.1
Summary: Generic Python utilities — KEV environment manager, XDG directories, persistent dedup filter, structured logger (lines and spans as one JSON or human event stream, shared with go-utils and ts-utils), number/percentage/currency formatting, terminal presentation (ANSI-aware tables, semantic palette, live regions with spinners and progress, ported from ts-utils), TTS text normalization (say). Python 3.12+.
Author-email: Adrian Galilea <adriangalilea@gmail.com>
Requires-Python: >=3.12
Provides-Extra: verbalize
Requires-Dist: num2words>=0.5.14; extra == 'verbalize'
Provides-Extra: verbalize-spanish
Requires-Dist: num2words>=0.5.14; extra == 'verbalize-spanish'
Requires-Dist: spacy<4,>=3.7; extra == 'verbalize-spanish'
Description-Content-Type: text/markdown

# adrian-utils

Generic Python utilities for DX — structured logger, typed boundary errors, env var manager (KEV), XDG paths, formatting helpers. Python 3.12+.

The sibling of [go-utils](https://github.com/adriangalilea/go-utils) and [ts-utils](https://github.com/adriangalilea/ts-utils). The logging contract the three share, and the harness that proves they write the same event stream, live in [utils](https://github.com/adriangalilea/utils).

## Install

```bash
pip install adrian-utils
# or with uv
uv add adrian-utils
```

## Install (local dev)

```bash
uv add -e .
```

## Usage

```python
from py_utils import log, kev, xdg, boundary, SourcedError, must, usd, percentage

# Logger: a message, then fields
log.info("connected", host="postgres", port=5432)
with log.span("build assets") as s:
    s.step("transpiling")
    s.step("bundling")
log.success("deployed")

# Typed boundary errors — every external call wears its source
@boundary("stripe")
def charge(customer_id: str, amount: int):
    return stripe.Charge.create(customer=customer_id, amount=amount)

try:
    charge("cus_ABC", 500)
except SourcedError as e:
    log.error("charge failed", err=e)  # source, operation, status, context, cause: all in the event

# Optional[T] → T with context
user = must(users.get(user_id), "user vanished", user_id=user_id)

# KEV — Redis-style env vars with source fallbacks
api_key = kev.must_get("API_KEY")
port = kev.int("PORT", 8080)
kev.set("DEBUG", "true")

# XDG paths
config_file = xdg.config / "myapp" / "config.toml"
state_dir = xdg.state / "myapp"

# Formatting
usd(1234.56)          # "+$1,234.56"
percentage(15.234)    # "+15.2%"
```

### Logger

A program's telemetry is one stream of events: a log line is an event at an instant, a span is a piece of work written as a start and an end event, and every line written inside a span names it. go-utils and ts-utils write the same stream byte for byte; the contract and the harness that checks it live in [utils](https://github.com/adriangalilea/utils).

Every verb takes a message, then fields. The message is positional-only, so any field name works:

```python
from py_utils import log

log.info("ready")
log.warn("retrying", attempt=2)
log.error("charge failed", err=e)
```

| verb | level | human |
|---|---|---|
| `error`, `fail` | error | `⨯` |
| `warn`, `warn_once` (once per message and scope) | warn | `⚠` |
| `info` | info | |
| `success` | info | `✓` |
| `wait` | info | `○` |
| `ready` | info | `▶` |
| `step` | info | `•`, indented |
| `debug` | debug | `◦` |
| `trace` | trace | `»` |

**Scopes** name subsystems: `log.scope("api").scope("auth")` writes `"scope":["api","auth"]`.

**Spans** are loggers. What is written through the span belongs to it; there is no ambient context, so a function that logs inside a span takes the span:

```python
with log.span("charge", amount=a) as s:
    s.warn("retrying", attempt=2)
    with s.span("stripe") as c:
        c.info("posted")
    s.set(receipt=r)          # carried on the end event
```

An exception escaping the block ends the span failed (`status: error`, the exception in `fields.err`) and propagates. Work that does not fit a block opens and ends the span by hand, once (a second `end` raises):

```python
s = log.begin("upload", size=n)
...
s.end()                       # or s.end(err)
```

**Traces cross processes** explicitly. `s.traceparent()` is the W3C value to send in an HTTP `traceparent` header or a child's `TRACEPARENT`; `log.join(value)` returns a logger whose spans are children of that remote span (`None` returns `log`, a malformed value is ignored with a warn once per distinct value). A process started with `TRACEPARENT` set joins it for its root spans with no code.

**Output** goes to stderr. `LOG_FORMAT=json` writes one event per line:

```json
{"time":"2026-10-12T12:20:39.123Z","level":"warn","verb":"warn","msg":"retrying","service":"checkout","scope":["payments"],"trace_id":"4bf92f3577b34da6a3ce929d0e0e4736","span_id":"00f067aa0ba902b7","fields":{"attempt":2}}
```

`LOG_FORMAT=human` writes symbols, colour (`NO_COLOR`, `FORCE_COLOR`), dim `key=value` fields, a raised exception's traceback under its line (once: a span that fails with the error a span inside it already failed with shows the message only), and spans with what they hold indented:

```
○ release version=0.7.0
    • building wheel
  ○ checks
      ruff result=clean
  ✓ checks 41ms
✓ release 254ms published=false version=0.7.0
```

The default is human when stderr is a TTY, json otherwise. Field values follow one set of rules: maps sorted at every depth, exceptions as `type`, `message`, `stack`, `cause` (and a `SourcedError`'s `source`, `operation`, `status`, `context`), `datetime` as a UTC millisecond string, `timedelta` as milliseconds, bytes (base64), `Decimal` and integers beyond 2^53 as strings, NaN and infinities as strings, dataclasses as objects, sets as arrays (sorted when their items compare), non-string map keys as their string form, a value that contains itself as `"[circular]"`, anything else as `str()`.

**Configuration.** Each variable has a setter that sets the program's default; the environment overrides it. Values match without regard to case (`LOG_LEVEL=DEBUG` is `debug`), an empty variable is unset, and an unknown value raises `ValueError`.

| variable | setter | values |
|---|---|---|
| `LOG_LEVEL`, `{SCOPE}_LOG_LEVEL` | `log.set_log_level` | `silent` `error` `warn` `info` `debug` `trace`, default `info` |
| `LOG_FORMAT` | `log.set_log_format` | `human` `json` |
| `LOG_TIME` | `log.set_log_time` | `1` `0`: a dim local `HH:MM:SS` before human lines |
| `LOG_SERVICE` | `log.set_log_service` | the program's name, default the script's file name without extension |
| `LOG_SHIP` | `log.set_log_ship` | an http(s) URL: every event that passes its threshold is also POSTed there (Shipping) |
| `LOG_SHIP_TOKEN` | `log.set_log_ship_token` | a token, sent as `Authorization: Bearer` with each batch |

A scoped event reads its level live from the most specific variable set: scope `["api", "auth"]` reads `API_AUTH_LOG_LEVEL`, then `API_LOG_LEVEL`, then `LOG_LEVEL`, then `set_log_level`, then `info`. Whether a span is shown is decided once, when it begins, by its start's level; its end follows, so a hidden span writes no end. A failure is never lost to that: a hidden span that fails writes one `fail` line (level error, its name as the message, its fields plus `err` and `dur_ms`) as a line of its nearest shown span, with no `phase`. A hidden span hands its lines, child spans and `traceparent()` to its nearest shown ancestor (its own id only when it has none), so a viewer never meets a parent that is not there.

**Shipping.** With `LOG_SHIP` set, the logger also POSTs its events to that URL, so a script ships its stream with no code:

```sh
LOG_SHIP=https://logs.example.com/ingest LOG_SHIP_TOKEN=… uv run python job.py
```

What ships is the stream's json line, byte for byte, whatever stderr renders. A batch is lines joined by `\n`, `Content-Type: text/plain;charset=utf-8`, at most 64 KB and 50 events. It leaves 2s after its first event, at 50 events, when a span opened outside any span of this process ends (a root span, or a request joined from its caller: its whole trace leaves together), and when the process exits (`atexit`). `log.flush_log()` sends what waits and returns once nothing is in flight, for a process that leaves through `os._exit`. Shipping never blocks a write and never raises: a refused or lost batch is one `warn_once` under scope `ship` (no retry), an event over 64 KB stays on stderr and warns once. The wire and the collector's answers are the contract's, in [utils](https://github.com/adriangalilea/utils#shipping).

The runnable proof of every rule above is `uv run python example_usage.py`.

### Typed errors

> "A confused program SHOULD scream." — John Carmack

Python is already offensive by default: exceptions propagate, uncaught crashes the process, `assert` is a language keyword. `py_utils.offensive` does **not** try to replace `assert`. It fills the one real gap Python has — **typed forensics at external-system boundaries** — and adds a small ergonomic helper for `Optional[T]` plus a typed assertion complement for the narrow case of catch-boundaries that route on bug-class.

#### The real win — `@boundary` + `SourcedError`

Every Python codebase talking to external systems reinvents error wrapping badly:

```python
try:
    return stripe.Charge.create(...)
except stripe.error.StripeError as e:
    raise RuntimeError(f"stripe failed: {e}")   # type lost, status lost, context lost
```

Upstream can't tell which system failed, can't read the HTTP status, can't retry per-source, can't serialize the forensics across a queue. `@boundary` + `SourcedError` provide the missing convention:

```python
from py_utils import boundary, SourcedError, log

@boundary("stripe")
def charge(customer_id: str, amount: int) -> Charge:
    return stripe.Charge.create(customer=customer_id, amount=amount)

@boundary("postgres")
def save(row: dict) -> None:
    db.execute("INSERT INTO charges VALUES (?)", row)

for doc in docs:
    try:
        save({"summary": summarize(doc.text)})
    except SourcedError as e:
        log.error("skipping doc", doc=doc.id, err=e)
        if e.source == "stripe" and e.status == 402:
            continue        # card declined, keep going
        if e.source == "postgres":
            raise           # db down, abort the whole batch
```

Every `SourcedError` carries `source`, `operation`, `status`, the original exception (chained via `__cause__`), arbitrary keyword context, and `.to_dict()` for transport across process boundaries. You can also raise one explicitly when you want finer control than the decorator gives you.

#### The ergonomic win — `must()`

`users.get(user_id)` → `Optional[User]` is everywhere in Python, and the idiomatic unwrap is verbose and untyped:

```python
user = users.get(user_id)
if user is None:
    raise RuntimeError(f"user vanished: {user_id}")
process(user)
```

`must()` is the one-liner. It narrows the type for pyright/mypy and carries structured context on failure:

```python
from py_utils import must

user = must(users.get(user_id), "user vanished", user_id=user_id)
process(user)   # user: User, not User | None
```

#### The narrow case — typed assertions that route on bug-class

**For most code, `assert` is fine.** `assert amount > 0, f"bad amount: {amount}"` does the job in scripts and CLIs.

The typed assertion primitives exist for one specific case: a catch boundary that needs to route on *whose bug it is*. HTTP handlers are the canonical example:

```python
from fastapi import FastAPI, HTTPException
from py_utils import require, ensure, boundary, SourcedError, PreconditionError, ContractError

app = FastAPI()

@boundary("stripe")
def stripe_charge(customer_id: str, amount: int) -> Charge:
    return stripe.Charge.create(customer=customer_id, amount=amount)

def do_charge(customer_id: str, amount: int) -> Charge:
    require(amount > 0, "amount must be positive", amount=amount)                 # caller's bug → 400
    result = stripe_charge(customer_id, amount)                                    # stripe's bug → 502
    ensure(result.id is not None, "stripe returned no id", customer_id=customer_id)  # our bug  → 500
    return result

@app.post("/charge")
def charge_endpoint(customer_id: str, amount: int):
    try:
        return do_charge(customer_id, amount)
    except PreconditionError as e:
        raise HTTPException(status_code=400, detail={"error": str(e), **e.context})
    except SourcedError as e:
        raise HTTPException(status_code=502, detail=e.to_dict())
    except ContractError as e:
        raise HTTPException(status_code=500, detail={"error": str(e)})
```

The design-by-contract vocabulary (Eiffel's lineage) names *whose bug* each check catches:

- `require(cond, msg, **ctx)` — precondition: **caller** is wrong → `PreconditionError`
- `invariant(cond, msg, **ctx)` — internal state: **we** are wrong → `InvariantError`
- `ensure(cond, msg, **ctx)` — postcondition: **we broke our promise** → `PostconditionError`

All three subclass `ContractError` → `AssertionError`, so existing `except AssertionError` still catches. **In a script with no such routing boundary, these are overkill — use `assert`.**

#### When to reach for what

| Situation | Use |
|---|---|
| Script or CLI | `assert x, f"..."` |
| Calling an external system (HTTP, db, subprocess) | `@boundary("source")` |
| Unwrapping `Optional[T]` where `None` is a bug | `must(value, ...)` |
| Handler that maps exceptions to HTTP status | `require` / `invariant` / `ensure` |
| Anywhere else | `assert` is fine |

#### Exception hierarchy

```
AssertionError
└── ContractError
    ├── PreconditionError   (require)
    ├── InvariantError      (invariant, must)
    └── PostconditionError  (ensure)

Exception
└── SourcedError            (boundary failures)
```

`@boundary` wraps raw exceptions into `SourcedError` but **lets `ContractError` pass through unwrapped** — contract failures are bugs in *us*, not failures of the external source, and should not be mislabeled.

Full rationale at the top of `src/py_utils/offensive.py`. Runnable demo: `uv run python example_offensive.py`.

### KEV

Redis-style KV store for environment variables. Searches memory → `os.environ` → `.env` files, caches results.

```python
from py_utils import kev

# Basics
api_key = kev.must_get("API_KEY")       # Raises if not found
api_key = kev.get("API_KEY", "dev")     # With default
port = kev.int("PORT", 8080)            # Type conversion
debug = kev.bool("DEBUG", False)        # true/1/yes/on → True
rate = kev.float("RATE", 0.5)           # Float conversion
kev.set("APP_NAME", "myapp")            # Memory only (fast)

# Namespaced access (skip the search chain)
kev.get("os:PATH")                      # ONLY from OS
kev.get(".env:SECRET")                  # ONLY from .env file
kev.set("os:DEBUG", "true")             # Write directly to OS
kev.set(".env:API_KEY", "secret")       # Update .env file

# Source tracking
value, source = kev.get_with_source("API_KEY")  # ("secret", ".env")
kev.source_of("API_KEY")                        # ".env" or "os" or "default"

# Customize search order
kev.source.remove("os")                 # Ignore OS env (perfect for tests!)
kev.source.add(".env.local")            # Add more fallbacks
kev.source.set(".env.test")             # Replace entirely

# Pattern matching
kev.keys("API_*")                       # Find all API_ keys
kev.has("API_KEY")                      # Check if configured
kev.clear("TEMP_*")                     # Clear from memory

# Lookup trace: KEV_LOG_LEVEL=debug logs each source checked (never the value)
kev.get("DATABASE_URL")
```

### XDG

XDG Base Directory paths — reads env vars set by [xdg-dirs](https://github.com/adriangalilea/xdg-dirs), falls back to spec defaults:

```python
from py_utils import xdg

xdg.config / "myapp" / "config.toml"    # ~/.config/myapp/config.toml
xdg.data / "myapp" / "data.db"          # ~/.local/share/myapp/data.db
xdg.state / "notify"                    # ~/.local/state/notify
xdg.cache / "myapp"                     # ~/.cache/myapp
xdg.runtime / "myapp"                   # $XDG_RUNTIME_DIR/myapp
```

### Unseen

"What's new since last time?" — filters a sequence of dicts to only the ones you haven't seen before. Remembers across runs.

```python
from py_utils import unseen

messages = fetch_messages()
new_messages = unseen("messages", messages, "id")
```

1st run:
```
messages     = [{"id": "1", "from": "alice", "text": "hi"}]
new_messages = [{"id": "1", "from": "alice", "text": "hi"}]
```

2nd run, no new message:
```
new_messages = []
```

3rd run, bob replied:
```
messages     = [{"id": "1", ...}, {"id": "2", "from": "bob", "text": "hey"}]
new_messages = [{"id": "2", "from": "bob", "text": "hey"}]
```

Saves state to: `$XDG_STATE_HOME/unseen/{name}.json`

### Colors

`format` and `currency` return plain strings, so a value reads the same in a terminal, a file or a JSON field. Color is the caller's: `cli.ui.ok(percentage(x))`, gated on TTY / `NO_COLOR` / `FORCE_COLOR` by `cli`.

#### Migrating from 0.8

`set_color_enabled`, `color_enabled` and `color_by_sign` are gone. They emitted rich markup (`[green]+5.0%[/]`), not escapes, so a terminal printed the brackets. Wrap the value with `cli.ui` instead.

### ANSI primitives (`ansi`)

Raw escape codes, ALWAYS ON: the vocabulary under the TTY-aware `cli` palette, for output whose consumer renders ANSI regardless of what stdout is (Claude Code status lines, tmux status strings). 256-color first; `rgb` for known-truecolor consumers. Out-of-range values raise `PreconditionError`. Same names as ts-utils `ansi`.

```python
from py_utils.ansi import fg, bg, rgb, BOLD, RESET, strip_ansi

f"{BOLD}{fg(171)}Fable 5{RESET}"   # bold bright violet
strip_ansi(styled)                 # the string minus its escapes
```

### CLI presentation (`cli`)

Aligned tables, key/value blocks, trees and a semantic palette, the port of ts-utils `cli`. Colors require stdout and stderr to be TTYs (`NO_COLOR` / `FORCE_COLOR` override), so the same `table()` is colored in a terminal and plain aligned text in a pipe. Alignment uses visible width in terminal cells: escapes take none, a ZWJ emoji or flag is one grapheme, CJK takes two cells.

```python
from py_utils.cli import table, kv, tree, indent, clip, ui

table(
    [["Ada", ui.accent("ada@x.com"), ui.ref("42")],
     ["Bo",  ui.accent("bo@y.com"),  ui.ref("7")]],
    head=["name", "email", "msgs"], align=["l", "l", "r"],
)
kv([("name", "Ada"), ("email", "ada@x.com")], indent=2)
clip("a very long value", 10)   # "a very lo…", styles survive
```

### Live output (`cli`): pinned region, spinner, progress

A region at the bottom of the terminal repaints `render()` (a string, so `table()` and the widgets `spin()`, `bar()`, `elapsed()` compose inside it). While it is active, `sys.stdout` and `sys.stderr` print each complete line above it, so the logger and any `print` never tear it. Workers in threads mutate the state and log freely. Non-TTY: nothing animates, `done()` prints the final frame once.

```python
from py_utils.cli import live, spinner, spin, bar, table, ui

with spinner("connecting 0/4") as set_label:   # ✓ connecting 4/4 1.2s
    ...
    set_label("connecting 4/4")

with live(lambda: table([[ui.ok("✓") if d == t else spin(), name, bar(d, t, 18)]
                         for name, (d, t) in jobs.items()])):
    run_workers()                               # final frame persists
```

One region at a time (`PreconditionError` otherwise); a `spinner()` under an active region prints its final line above it. A render that raises stops the repainting and `done()` raises it. Run `uv run python example_cli.py` (and `| cat` for the pipe rendering).

## Run the demos

```bash
uv add -e .
uv run python example_usage.py       # logger contract, asserted; human rendering
uv run python example_offensive.py   # offensive primitives
```

## Linting / formatting

```bash
uv run ruff check src
uv run ruff format src
```

Part of the utils suite by Adrian Galilea: **[go-utils](https://github.com/adriangalilea/go-utils)**, **[ts-utils](https://github.com/adriangalilea/ts-utils)**, **py-utils**.
