Skip to main content

Python SDK

The zeridion-flare package is published on PyPI and follows the same wire contract as every other Flare SDK.

pip install zeridion-flare

Requires: Python 3.10+ · httpx>=0.27 (declared as a dependency; pip installs it for you).

Quick start​

The client exposes flat methods (one per endpoint) rather than nested namespaces, and returns plain dict payloads matching the wire format.

from zeridion_flare import FlareClient

flare = FlareClient(api_key="zf_live_sk_...")

# Enqueue a job. create_job takes the request body as a dict (the same shape
# as POST /jobs) — including optional fields like "tags".
# Optional keyword-only request_id / idempotency_key are accepted per call:
# - idempotency_key dedupes a replayed write (same key + same body → original response).
# - request_id is a log-correlation hint (your trace ID flows into Zeridion's logs);
#   omit it and the server returns a generated ULID in the response X-Request-Id header.
job = flare.create_job(
    {
        "job_type": "SendWelcomeEmail",
        "payload": {"email": "alice@example.com", "name": "Alice"},
        "queue": "default",
        "max_attempts": 3,
        "tags": {"env": "dev"},
    },
    idempotency_key="welcome-email-alice",
)
print(job["id"], job["state"])  # "job_abc123", "pending"

# Get job status
detail = flare.get_job(job["id"])

# List jobs (returns a dict with "data" + "has_more" + "next_cursor")
page = flare.list_jobs(state="failed", limit=25)
for j in page["data"]:
    print(j["id"], j["state"])

# Cancel / retry
flare.cancel_job(job["id"])
flare.retry_job(job["id"])

Client construction​

KeywordDefaultDescription
api_keyos.environ["FLARE_API_KEY"]Required — pass directly or via env var.
base_urlhttps://api.zeridion.com/flare/v1Override for dev / staging. Unlike most other Flare SDKs, the default includes the /flare/v1 path — an override must include it too.
max_retries3Maximum retry attempts on 429 / 5xx / network errors. Set 0 to disable.
retry_base_delay_ms500Base for the exponential backoff schedule.
retry_max_delay_ms30000Cap on a single backoff wait (and on Retry-After).
max_response_bytes10 * 1024 * 1024 (10 MiB)Hard cap on response body size; protects against an unbounded body. Set 0 to disable.

Type hints​

The SDK is fully typed using Python 3.10+ union syntax (str | None, dict[str, Any]) — no Optional[...] / Union[...] imports needed. Method signatures, dataclass fields, and exception attributes all carry runtime-checkable annotations, so mypy / pyright / your IDE will flag drift without you re-typing anything.

Typed error hierarchy​

The following are importable from zeridion_flare:

ClassHTTPWhen it fires
FlareError—Base class for every SDK exception
AuthError401Missing / invalid / revoked API key
QuotaError402Billing state blocked — subscription past_due / unpaid / cancelled
NotFoundError404Resource doesn't exist (or returns None on GETs)
ConflictError409Idempotency or invalid-state conflict
RateLimitError429Rate-limited; inspect .retry_after, .limit, .remaining

Webhook verification​

verify_webhook(payload, signature_header, secret, *, tolerance_seconds=None) verifies the X-Zeridion-Signature header on a delivery. Pass the raw request body bytes — parsing and re-serializing the JSON invalidates the signature.

from zeridion_flare import verify_webhook

@app.post("/hooks/zeridion")
async def receive(request):
    raw = await request.body()
    header = request.headers.get("x-zeridion-signature", "")
    if not verify_webhook(raw, header, WEBHOOK_SECRET, tolerance_seconds=300):
        return Response(status_code=400)
    ...

Set tolerance_seconds (recommended: 300) to reject replayed deliveries whose signature timestamp is too old. The helper is rotation-safe — it accepts any matching v1= digest in the header.

Tags and progress​

Because create_job takes the raw request body, tags work today — pass them as a key in the body dict (see the example above). Job progress is reported by workers, so reading it is available now via get_job (the response includes progress when set). Reporting progress from Python is available in the worker via ctx.report_progress(...).

Running jobs (worker)​

The package also ships a built-in worker that runs your background jobs — register handlers, long-poll, heartbeat with progress, and drain gracefully on shutdown. It is an opt-in import:

from zeridion_flare.worker import FlareWorker, JobContext

worker = FlareWorker()  # FLARE_API_KEY from env

@worker.job("SendWelcomeEmail", queue="email", max_attempts=5, timeout=60)
def send_welcome(payload, ctx: JobContext) -> None:
    send_email(payload["email"])
    ctx.report_progress(1.0)

if __name__ == "__main__":
    worker.run()  # blocks; SIGTERM / SIGINT → graceful drain

See the worker overview for handlers, recurring jobs, the job context, and configuration.

The threaded worker is what ships today; an asyncio-native worker is a fast-follow (see Asyncio). There is no AsyncFlareClient on the immediate roadmap — if asyncio support is a hard requirement for your stack, wrap FlareClient calls in asyncio.to_thread() for now and tell us via GitHub issues.

See also​

Last updated on