Python SDK · Sextant docs

Requires Python 3.10 or later. The package is fully typed.

Shell

pip install sextant-client

Client

client.pyPython

import os
from sextant import Sextant, NotFoundError

client = Sextant(url=os.environ["SEXTANT_URL"], token=os.environ["SEXTANT_TOKEN"], timeout=10)

job = client.jobs.create(
    queue="emails",
    handler="send-receipt",
    payload={"order_id": "ord_4412"},
    idempotency_key="receipt-ord_4412",
)

try:
    client.jobs.cancel(job.id)
except NotFoundError:
    pass

Arguments are keyword-only and use snake_case. Times are timezone-aware datetime objects.

Async

AsyncSextant has the same methods as coroutines:

async_client.pyPython

import asyncio
from sextant import AsyncSextant

async def main():
    async with AsyncSextant() as client:
        async for job in client.jobs.list(state="dead"):
            await client.jobs.retry(job.id)

asyncio.run(main())

Paging

list returns an iterator (or async iterator) that follows the cursor. Call .page() for a single page with next_cursor.

Handlers

worker.pyPython

from sextant import Sextant, handler, PermanentError

client = Sextant()
worker = client.worker(queues=["emails"], concurrency=20)

@worker.handler("send-receipt")
def send_receipt(job):
    order = orders.find(job.payload["order_id"])
    if order is None:
        raise PermanentError("order not found")
    mailer.send(order.email, receipt(order))

worker.run()  # blocks; stops gracefully on SIGTERM

Handlers may be async def when you use AsyncSextant. For long-running work, check job.cancelled inside loops:

Python

for chunk in chunks:
    if job.cancelled:
        return
    process(chunk)

Errors

Exceptions mirror the HTTP status: ValidationError (422), PermissionError (401 and 403), NotFoundError (404), ConflictError (409), RateLimitError (429, with retry_after) and ServerError (5xx). All carry request_id.