python-config-client 0.1.2

Python client SDK for Config Service

python-config-client

Python SDK for Config Service — fetch, decrypt, and hot-reload service configurations with AES-256-GCM encryption.

Features

• AES-256-GCM decryption — compatible with Go SDK's wire format
• Sync HTTP client via httpx with configurable retry + exponential backoff
• SSE watch mode with auto-reconnect and backoff
• Thread-safe Snapshot[T] for zero-lock hot-reload reads
• Flexible deserialization — dataclass, Pydantic v2 BaseModel, or dict
• Full mypy --strict typing — no Any leaks in the public API

Requirements

• Python ≥ 3.10
• A running Config Service with a valid service token and encryption key

Installation

pip install python-config-client

# With Pydantic v2 support
pip install "python-config-client[pydantic]"

Quick Start

import os
from dataclasses import dataclass
from config_client import ConfigClient, Options

@dataclass
class AppConfig:
    log_level: str
    debug: bool

with ConfigClient(Options(
    host=os.environ["CONFIG_SERVICE_HOST"],
    service_token=os.environ["CONFIG_SERVICE_TOKEN"],
    encryption_key=os.environ["CONFIG_SERVICE_KEY"],
)) as client:
    cfg = client.get("my-service", AppConfig)
    print(cfg.log_level)

Configuration

Options

Field	Type	Default	Description
host	str	required	Base URL of Config Service, e.g. https://config.example.com
service_token	str	required	Plain-text service token (sent as X-Service-Token)
encryption_key	str	required	64-character hex string representing a 32-byte AES-256 key
request_timeout	float	10.0	Per-request timeout in seconds
retry_count	int	3	Number of retry attempts for 5xx / network errors
retry_delay	float	1.0	Base delay for exponential backoff (seconds)
on_error	Callable[[Exception], None] | None	None	Invoked on watch errors; does not stop the loop
on_change	Callable[[str], None] | None	None	Invoked with config name on each received change event
http_client	httpx.Client | None	None	BYO client (caller owns lifecycle)

Environment Variables

Load all three required options from the environment with from_env():

from config_client import ConfigClient

client = ConfigClient.from_env()

Variable	Description
CONFIG_SERVICE_HOST	Base URL of Config Service
CONFIG_SERVICE_TOKEN	Plain-text service token
CONFIG_SERVICE_KEY	AES-256 encryption key (64 hex chars)

GetOptions

Per-request overrides for get(), get_raw(), get_bytes():

from config_client import GetOptions

cfg = client.get("my-service", AppConfig, GetOptions(environment="production", version=5))

Field	Type	Default	Description
environment	str	""	Environment override (e.g. "production")
version	int	0	Specific version to fetch (0 = latest)

API Reference

ConfigClient

get(config_name, target_type, opts=None) → T

Fetch, decrypt (AES-256-GCM), and deserialize a configuration. Dispatches to:

• pydantic.BaseModel.model_validate(data) — for Pydantic v2 models
• Recursive dataclass construction — for @dataclass types
• Identity passthrough — for dict

cfg = client.get("my-service", AppConfig)

get_raw(config_name, opts=None) → dict[str, object]

Decrypt and return the raw JSON as a dict.

data = client.get_raw("my-service")

get_bytes(config_name, opts=None) → bytes

Decrypt and return raw JSON bytes (no parsing).

raw = client.get_bytes("my-service")

get_formatted(config_name, fmt) → bytes

Fetch a config in plaintext via the /formatted endpoint. No decryption.

from config_client import Format

yaml_bytes = client.get_formatted("my-service", Format.YAML)
env_bytes  = client.get_formatted("my-service", Format.ENV)
json_bytes = client.get_formatted("my-service", Format.JSON)

list() → list[ConfigInfo]

Return metadata for all configurations accessible to this token.

for info in client.list():
    print(info.name, info.is_valid, info.updated_at)

ConfigInfo fields: name: str, is_valid: bool, valid_from: datetime, updated_at: datetime.

watch(config_name, callback)

Subscribe to SSE change events. Blocks the calling thread. Auto-reconnects.

watch_and_decode(config_name, target_type, snapshot, opts=None)

Watch for changes and update a Snapshot[T] on every change. Blocks.

from_env() → ConfigClient

Class method. Creates a client from CONFIG_SERVICE_HOST, CONFIG_SERVICE_TOKEN, CONFIG_SERVICE_KEY.

close()

Stop all watch loops and close the HTTP transport. Also called by __exit__.

Watch & Hot-Reload

Use watch_and_decode inside a daemon thread to keep a Snapshot up to date without blocking the main thread:

import threading
from dataclasses import dataclass
from config_client import ConfigClient, Options, Snapshot

@dataclass
class FeatureFlags:
    dark_mode: bool
    max_connections: int

snapshot: Snapshot[FeatureFlags] = Snapshot()
client = ConfigClient.from_env()

def _watcher() -> None:
    client.watch_and_decode("feature-flags", FeatureFlags, snapshot)

t = threading.Thread(target=_watcher, daemon=True)
t.start()

# At any point — zero-lock read:
flags = snapshot.load()
if flags is not None and flags.dark_mode:
    print("dark mode enabled")

Call client.close() to stop the watcher thread gracefully.

Error Handling

Exception	HTTP status / cause
UnauthorizedError	HTTP 401
ForbiddenError	HTTP 403
NotFoundError	HTTP 404
InvalidResponseError	Any other 4xx
ConnectionError	Network error or 5xx after all retries exhausted
DecryptionError	AES-GCM authentication failure or malformed ciphertext
UnmarshalError	JSON → target type deserialization failed
ConfigClientError	Base class for all SDK exceptions; also raised for invalid options

from config_client import ConfigClient, NotFoundError, DecryptionError

try:
    cfg = client.get("my-service", AppConfig)
except NotFoundError:
    print("config not found")
except DecryptionError:
    print("wrong encryption key or corrupted payload")

Development

Requirements

• uv ≥ 0.4

Setup

git clone https://github.com/holdemlab/python-config-client
cd python-config-client
make install
cp .env.example .env  # fill in real values for integration tests

Makefile Commands

Command	Description
make install	Install all dependencies (including dev)
make test	Run unit + integration tests with coverage
make unit	Unit tests only
make lint	ruff check + ruff format --check
make fmt	Auto-format with ruff format
make typecheck	mypy config_client --strict
make security	bandit -r config_client + uv audit
make build	Build sdist + wheel (uv build)
make docker-up	Start postgres + app via docker-compose
make docker-down	Stop containers

Running Tests

# All unit tests (fast, no external services):
make unit

# Integration tests (requires CONFIG_SERVICE_* env vars):
make integration

# Full suite with coverage report:
make test

Project Structure

config_client/
  __init__.py        # public API surface
  client.py          # ConfigClient main class
  options.py         # Options / GetOptions dataclasses
  types.py           # ConfigInfo, ConfigChangeEvent, Format
  errors.py          # exception hierarchy
  crypto.py          # AES-256-GCM decrypt + parse_encryption_key
  transport.py       # httpx transport with retry + backoff
  _sse.py            # internal SSE reader (private)
  snapshot.py        # thread-safe Snapshot[T]

tests/               # pytest suite (97% coverage)
examples/            # runnable usage examples

License

MIT — see LICENSE for details.