asyncbase — Python SDK

AsyncBase Python SDK. Queue primitive client with both sync and async APIs.

Step 1 scaffolding. Implementation in Step 6 (post core API).

Install

pip install asyncbase

Quickstart (once implemented)

Sync

from asyncbase import Queue

q = Queue("sk_live_xxx")

q.send("emails", {"to": "user@example.com"})

for msg in q.consume("emails", group="worker-1"):
    send_email(msg.payload)
    msg.ack()

Async (FastAPI / asyncio)

from asyncbase import AsyncQueue

q = AsyncQueue("sk_live_xxx")

await q.send("emails", {"to": "user@example.com"})

async for msg in q.consume("emails", group="worker-1"):
    await send_email(msg.payload)
    await msg.ack()

Development

python -m venv .venv
source .venv/bin/activate
pip install -e '.[dev]'

pytest
ruff check .
mypy src