confish 0.2.0

Official Python SDK for confish — typed configuration, actions, and webhooks.

confish

Official Python SDK for confish — typed configuration, feeds, actions, and webhook verification.

• One dependency (httpx)
• Sync client with typed exceptions and automatic retry on 429/5xx
• Long-running action consumer with threading.Event cancellation
• HMAC-SHA256 webhook verification (stdlib only)

Install

pip install confish

Requires Python 3.10+.

Quick start

from confish import Confish

client = Confish(
    env_id="a1b2c3d4e5f6",
    api_key="confish_sk_...",
)

config = client.config.fetch()
print(config["site_name"])

The methods return dict[str, Any]. To add static typing, use TypedDict and cast:

from typing import TypedDict, cast

class MyConfig(TypedDict):
    site_name: str
    max_upload_mb: int
    maintenance_mode: bool

config = cast(MyConfig, client.config.fetch())
config["maintenance_mode"]  # type-checked as bool

Or with Pydantic:

from pydantic import BaseModel

class MyConfig(BaseModel):
    site_name: str
    max_upload_mb: int
    maintenance_mode: bool

config = MyConfig.model_validate(client.config.fetch())

Reading and writing config

# GET /c/{env_id}
config = client.config.fetch()

# PATCH — only listed fields change
client.config.update({"maintenance_mode": True})

# PUT — replaces everything; omitted fields reset to defaults
client.config.replace({
    "site_name": "My App",
    "max_upload_mb": 50,
    "maintenance_mode": False,
})

update and replace return the full updated configuration.

Write access must be enabled in environment settings before update and replace will work.

Feeds

Feeds hold living, externally-keyed state — one item per external_id, partitioned per environment. client.feed(slug) returns a bound handle; no HTTP happens until you call a method.

jobs = client.feed("jobs")

# Create or replace an item (PUT). Expires after 24 hours.
jobs.set("sitemap-crawl", {"status": "running", "pages": 1204}, ttl=86400)

# Live items, newest first -> list[FeedItem]
for item in jobs.list():
    print(item.external_id, item.data, item.expires_at)

# Idempotent — deleting a missing item succeeds
jobs.delete("sitemap-crawl")

# Replace the whole feed in one request — built for sync-style cron jobs
# pushing their full dataset. Anything absent is DELETED; [] clears the feed.
result = jobs.replace([
    {"external_id": "sitemap-crawl", "data": {"status": "running"}, "ttl": 86400},
    {"external_id": "price-sync", "data": {"status": "queued"}},
])
print(result.created, result.updated, result.deleted)

set upserts with declarative PUT semantics: the item's data becomes exactly what you pass, and the TTL becomes exactly ttl. Omitting ttl makes the item permanent — it clears any TTL set by a previous set. ttl is in seconds (1 to 2,592,000 — 30 days); external_id is limited to 255 characters.

replace applies the same declarative semantics to the whole partition: it's all-or-nothing (duplicate external IDs, exceeding the plan's item cap, or any schema-invalid item raises ValidationError with nothing written) and returns a FeedReplaceResult with created/updated/deleted counts.

Each FeedItem has id, external_id, data, expires_at (None for permanent items), created_at, and updated_at (ISO 8601 strings). An unknown feed slug raises NotFoundError; a schema mismatch or full feed raises ValidationError.

Logging

client.logs.info("Worker started", {"region": "eu-west-1"})
client.logs.error("Job failed", {"job_id": "abc"})

# Or with an explicit level:
log_id = client.logs.write("info", "User logged in", {"user_id": 123})

Levels: debug, info, notice, warning, error, critical, alert, emergency. They follow RFC 5424 (syslog), so they map 1:1 onto stdlib logging levels.

Actions

The action consumer polls for pending actions, acknowledges them, runs your handler, and reports completion or failure — including idempotent skip if another consumer claimed the action first.

import threading
from confish import Confish, Action, SkipAction

client = Confish(env_id="...", api_key="...")
stop = threading.Event()

def handler(action: Action, ctx) -> dict | None:
    if action.type == "place_order":
        ctx.progress("Submitting order", {"params": action.params})
        # ... do work ...
        return {"order_id": "abc123", "filled_price": 66980.0}
    raise RuntimeError(f"Unknown action type: {action.type}")

client.actions.consume(
    handler=handler,
    poll_interval=15.0,    # base — defaults to 15s
    max_poll_interval=60.0, # adaptive backoff cap
    concurrency=2,
    stop=stop,
    on_error=lambda exc, action: print(f"action {action.id}: {exc}"),
)

# To stop, e.g. on signal:
import signal
signal.signal(signal.SIGTERM, lambda *_: stop.set())

What happens automatically:

• A returned dict becomes the action's result on completion.
• Raising any exception fails the action with {"error": str(exc)}.
• Raising SkipAction leaves the action acknowledged without resolving it.
• A 409 Conflict on ack is silently skipped — safe to run multiple consumers.
• Setting stop halts new work and waits for in-flight handlers to settle.
• After 3 consecutive empty polls the loop doubles its sleep up to max_poll_interval, resetting to poll_interval the moment any action is processed. Idle consumers make ~240 requests/hour by default.

You can also drive the lifecycle manually:

actions = client.actions.list()
client.actions.ack("action_id")
client.actions.progress("action_id", "closing 3 positions", {"step": 2})
client.actions.complete("action_id", {"order_id": "abc"})
client.actions.fail("action_id", {"error": "timeout"})

Webhook verification

verify parses and verifies in one operation: it returns the parsed WebhookPayload on success and raises on failure, so the payload you handle is guaranteed to be the exact bytes the signature covers.

from flask import Flask, request, abort
from confish.webhook import verify, WebhookSignatureError, WebhookTimestampError
import os

app = Flask(__name__)

@app.post("/webhook")
def webhook():
    try:
        payload = verify(
            body=request.data,
            signature=request.headers.get("X-Confish-Signature"),
            secret=os.environ["CONFISH_WEBHOOK_SECRET"],
        )
    except WebhookTimestampError:
        abort(401, "stale timestamp")
    except WebhookSignatureError:
        abort(401, "invalid signature")
    # handle payload.event, payload.changes, payload.values ...
    return "", 200

verify uses constant-time comparison and rejects timestamps older than 5 minutes by default (WebhookTimestampError). Pass tolerance_seconds=0 to disable timestamp checking. Both exceptions subclass WebhookVerificationError (itself a ConfishError) if you don't need to distinguish them. Always pass the raw, unparsed body — re-serializing parsed JSON breaks verification.

Errors

from confish import (
    AuthError,
    ConfishError,
    ConflictError,
    ForbiddenError,
    NetworkError,
    NotFoundError,
    RateLimitError,
    ServerError,
    ValidationError,
)

try:
    client.config.fetch()
except RateLimitError as e:
    print(f"slow down — retry after {e.retry_after}s")
except ValidationError as e:
    for field, msgs in e.errors.items():
        print(f"{field}: {msgs}")
except ConfishError as e:
    print(f"HTTP {e.status_code}: {e.message}")

By default the client retries 429 (honoring Retry-After) and 5xx responses up to twice. Tune with max_retries on the Confish constructor.

Options

client = Confish(
    env_id="a1b2c3d4e5f6",
    api_key="confish_sk_...",
    base_url="https://confi.sh",  # override for self-hosted
    user_agent="my-app/1.0",
    max_retries=2,
    max_retry_delay=30.0,
    http_client=None,             # inject your own httpx.Client
)

Confish is a context manager:

with Confish(env_id="...", api_key="...") as client:
    config = client.config.fetch()

License

MIT