Metadata-Version: 2.5
Name: consul-http
Version: 0.1.0
Summary: Lightweight httpx Consul KV client (sync + async)
Author-email: consul-http contributors <hhs66317@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: async,consul,distributed-lock,httpx,kv
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx<1,>=0.27
Description-Content-Type: text/markdown

# consul-http

Lightweight [Consul](https://developer.hashicorp.com/consul) **KV** client built on [httpx](https://www.python-httpx.org/).

中文说明见下方。

## Why this package?

Unlike full-featured clients such as [`py-consul`](https://pypi.org/project/py-consul/) / `python-consul`, this library focuses on:

- **In scope:** KV (`get` / `put` / `delete` / prefix), Session, distributed lock, blocking `watch_kv`
- **Out of scope (for now):** Agent / Catalog / Health / Txn / Connect — use a full Consul SDK if you need those
- First-class **sync + async** with the same API shape
- Async-friendly `async with client.lock(...)` for leader election
- Optional injection of an existing `httpx.Client` / `AsyncClient` (shared pools, proxies, OTel)
- Small dependency surface (`httpx` only)
- Explicit typing (`py.typed`)

## Compatibility

Targeted at **Consul 1.10+** HTTP API (KV + Session).

**URL / port rules** (intentional):

| Address | Port used |
|---------|-----------|
| `http://127.0.0.1` / `localhost` (no port) | **8500** |
| `http://consul.example.com` (no port) | scheme default **80/443** (reverse-proxy friendly) |
| `http://host:8500` | explicit **8500** |
| `http://host:8500/prefix` | path prefix kept → `.../prefix/v1` |

## Install

```bash
pip install consul-http
# or
uv add consul-http
```

Editable (from this monorepo):

```bash
uv add --editable ./packages/consul_http
```

Requires **Python 3.10+**.

## Tests

```bash
# unit (default; skips live Consul)
uv run pytest

# live agent — needs write ACL; loads CONSUL_HTTP_* from repo .env when present
$env:CONSUL_INTEGRATION = "1"   # PowerShell
uv run pytest -m integration
```

## Quick start

```python
from consul_http import ConsulClient

# Uses CONSUL_HTTP_ADDR / CONSUL_HTTP_TOKEN when set
with ConsulClient() as c:
    c.put_kv("app/demo/key", "hello")
    res = c.get_kv("app/demo/key")
    if res is None:
        print("missing")
    else:
        print(res.raw_value, res.index)
    c.delete_kv("app/demo/key")
```

### Async + blocking query

```python
import asyncio
from consul_http import AsyncConsulClient

async def watch(key: str) -> None:
    async with AsyncConsulClient() as c:
        res = await c.get_kv(key)
        assert res is not None
        index = res.index
        while True:
            res = await c.get_kv(key, index=index, wait="30s")
            if res is None:
                continue
            if res.index != index:
                print("changed:", res.raw_value)
                index = res.index

# asyncio.run(watch("app/demo/key"))
```

Prefer the built-in watcher (handles **index regression** after leader change /
snapshot restore — waiting on a stale index would hang forever otherwise):

```python
async def watch(key: str) -> None:
    async with AsyncConsulClient() as c:
        async for res in c.watch_kv(key, wait="30s"):
            if res is None:
                print("missing")
                continue
            print("changed:", res.raw_value)
```

### Session + distributed lock

```python
from consul_http import AsyncConsulClient

async def on_lost() -> None:
    print("lock lost — stop critical work")

async def run_leader() -> None:
    async with AsyncConsulClient() as c:
        lock = c.lock(
            "service/demo/leader",
            ttl="15s",
            value="instance-1",
            on_lost=on_lost,  # sync or async callable
        )
        async with lock as acquired:
            if not acquired:
                return
            while lock.is_held:
                # do leader work; exit early if renew failed
                ...

# Low-level: create_session / put_kv(..., acquire=...) / renew_session / destroy_session
# Read lock holder: c.get_kv("service/demo/leader", raw=False) → res.session, res.lock_index
```

Sync mirror: `with client.lock(...) as acquired:` (`on_lost` must be sync there).

Exit skips KV `release` when the lock was already lost (`lock.lost is True`).

### Injected httpx client / per-request token

```python
import httpx
from consul_http import ConsulClient

# Share pools / proxies / custom transport; ConsulClient will NOT close `http`.
# Default headers (incl. X-Consul-Token) and connect settings stay on `http` —
# constructing ConsulClient(client=...) does not merge token into that client.
# Blocking / watch reads still apply ConsulClient.timeout via per-request timeout.
with httpx.Client(
    base_url="http://127.0.0.1:8500/v1",
    headers={"X-Consul-Token": "default-tok"},
    timeout=30.0,
) as http:
    with ConsulClient(client=http, timeout=10.0) as c:
        c.get_kv("app/demo/key", token="one-shot-acl-token")
```

### Binary KV

```python
with ConsulClient() as c:
    c.put_kv("app/demo/blob", b"\x00\xffprotobuf-or-msgpack")
    res = c.get_kv("app/demo/blob", raw=False)
    assert res is not None
    data = res.raw_bytes  # exact bytes; raw_value may use U+FFFD on bad UTF-8
```

### Retry / backoff

By default clients retry transient failures up to **3 attempts** with exponential
backoff + jitter (`RetryConfig`).

| What | Retried? |
|------|----------|
| Transport / timeout (`httpx.RequestError`) | Yes |
| HTTP `429` / `502` / `503` / `504` | Yes (except CAS / acquire / release / `create_session` — see below) |
| `404`, other 4xx, CAS body `false` | No |

**CAS / lock / session-create caveat:** For `put_kv` / `delete_kv` with `cas=` /
`acquire=` / `release=`, and for `create_session`, status-code retries are
**disabled** so a lost `502`/`504` after a successful apply is not re-issued
(which could look like a false conflict / failed acquire, or orphan extra
sessions). Transport-error retries still run; if a request may have been applied
but the response was lost, a later retry can still return `false` /
`ConsulCASConflictError`, or create an extra session (TTL eventually reaps
orphans). When that happens and you suspect a retry race, re-`get_kv` (prefer
`raw=False`) and compare value / `session` before treating it as a real
concurrent write.

```python
from consul_http import ConsulClient, RetryConfig

# Custom policy
with ConsulClient(retry=RetryConfig(max_attempts=5, backoff_factor=0.2)) as c:
    c.get_kv("app/demo/key")

# Disable retries
with ConsulClient(retry=RetryConfig(max_attempts=1)) as c:
    c.get_kv("app/demo/key")
# or: ConsulClient(retry=None)
```

Writes do not retry transport errors by default because the server may have
applied a write before the response was lost. Such cases raise
`ConsulRequestOutcomeUnknown`; reconcile the key before deciding whether to
retry. `RetryConfig(respect_retry_after=True)` honors a server `Retry-After`
header for retryable status codes.

### Production watch lifecycle

```python
from threading import Event

stop = Event()
with ConsulClient() as c:
    for update in c.watch_kv(
        "app/config",
        stop_event=stop,
        max_consecutive_failures=5,
    ):
        if update is not None:
            apply_config(update.value)
```

Use `raise_if_lost()` during long lock-protected work. Lock loss cannot fence
an already-running process from writing an external system; use a fencing
token at the downstream storage layer when that guarantee is required.

For observability, pass an implementation of `EventSink` to the client. Hook
exceptions are isolated from request execution and events never contain ACL
tokens or KV values.

### Environment variables

| Variable | Meaning |
|----------|---------|
| `CONSUL_HTTP_ADDR` | e.g. `http://127.0.0.1:8500` |
| `CONSUL_HTTP_TOKEN` | ACL token |

### Errors

- `ConsulConnectionError` — transport / timeout failures  
- `ConsulPermissionError` — HTTP 401/403 (invalid token / ACL deny); subclass of `ConsulAPIError`  
- `ConsulAPIError` — other non-success HTTP (body truncated in message)  
- `ConsulCASConflictError` — `put_kv`/`delete_kv` with `cas=` returned `false` when `raise_on_cas_conflict=True`

---

## 中文

轻量 Consul KV 客户端（httpx Sync/Async），含 Session / 分布式锁。默认指数退避重试；`cas`/`acquire`/`release`/`create_session` 不重试 HTTP 状态码。目标 API：**Consul 1.10+**。

```python
from consul_http import ConsulClient

with ConsulClient.from_base_url("http://127.0.0.1:8500") as c:
    print(c.get_kv("app/demo/key"))
```
