Skip to content
useEventStack 0.3.0 is live.Read the quickstart

Reference

Sync and async Python client

Use the useeventstack package to emit events, manage workflows, replay, and query projections from Python 3.10+.

Installation

pip install useeventstack

Version 0.3.0. Requires Python 3.10+ and httpx.

To pin it in a project, add useeventstack>=0.3,<0.4 to your requirements. The SDK is MIT licensed; the platform itself is not open source.

Synchronous client

import os
from useeventstack import UseEventStackClient

with UseEventStackClient(
    api_key=os.environ["USEEVENTSTACK_API_KEY"],
    project_id=os.environ["USEEVENTSTACK_PROJECT_ID"],
    base_url="http://localhost:3001",
    organization_id=os.environ.get("USEEVENTSTACK_ORGANIZATION_ID"),
    environment="production",
) as client:
    result = client.events.emit(
        "deployment.completed",
        payload={
            "service": "api",
            "version": "2.1.0",
            "environment": "production",
            "status": "healthy",
        },
        idempotency_key="deploy-api-2.1.0",
    )
    print(f"Event: {result['event']['id']}")

    # Query events
    events = client.events.query(event_type="deployment.completed")

    # Replay
    client.replay.single("event-uuid", skip_side_effects=True)

Asynchronous client

from useeventstack import AsyncUseEventStackClient

async with AsyncUseEventStackClient(
    api_key="useeventstack_key",
    project_id="project-uuid",
) as client:
    result = await client.events.emit(
        "deployment.completed",
        payload={"service": "worker", "version": "1.5.0",
                 "environment": "sandbox", "status": "healthy"},
        environment="sandbox",
    )
    events = await client.events.query(event_type="deployment.completed")
    workflows = await client.workflows.list()

Available methods

Both sync and async clients expose identical namespaces:

NamespaceMethods
eventsemit, query, get, trace
workflowsquery, list, get, create, update, delete, enable, disable
custom_actionslist, get, delete, simulate
replaysingle, bulk
dlqlist, reprocess
api_keyslist, create, delete
projectionsquery, service_status

Responses are Python dictionaries. Non-2xx responses raise UseEventStackApiError.

Error handling

from useeventstack.errors import UseEventStackApiError

try:
    client.events.emit("bad.event", payload={})
except UseEventStackApiError as e:
    print(f"[{e.status}] {e.code}: {e.message}")
    print(e.details)  # Optional dict