Python

pip install asyncbase
# or: uv add, poetry add

import os
from asyncbase import Queue

with Queue(os.environ["ASYNCBASE_KEY"]) as q:
    r = q.send("emails", {"to": "a@b.com"}, delay="30s")
    print(r.id)

    for msg in q.pull("emails", group="workers", limit=10, wait_seconds=5):
        try:
            handle(msg.payload)
            msg.ack()
        except Exception:
            msg.nack()

import os, asyncio
from asyncbase import AsyncQueue

async def main():
    aq = AsyncQueue(os.environ["ASYNCBASE_KEY"])
    async for msg in aq.consume("emails", group="workers", visibility_seconds=60):
        try:
            await handle(msg.payload)
            await msg.ack()
        except Exception:
            await msg.nack()

asyncio.run(main())

from contextlib import asynccontextmanager
from fastapi import FastAPI
from asyncbase import AsyncQueue
import asyncio, os

_queue: AsyncQueue | None = None
_task: asyncio.Task | None = None

async def _consume():
    async for msg in _queue.consume("emails", group="fastapi-workers"):
        try:
            await handle(msg.payload); await msg.ack()
        except Exception:
            await msg.nack()

@asynccontextmanager
async def lifespan(_: FastAPI):
    global _queue, _task
    _queue = AsyncQueue(os.environ["ASYNCBASE_KEY"])
    _task = asyncio.create_task(_consume())
    try: yield
    finally:
        _task.cancel()
        try: await _task
        except asyncio.CancelledError: pass
        await _queue.aclose()

app = FastAPI(lifespan=lifespan)

from asyncbase import AsyncBaseError

try:
    q.send("emails", payload)
except AsyncBaseError as e:
    if e.code == "RATE_LIMITED":
        time.sleep(1)
    else:
        raise