Python SDK — async vs sync

paylera.PayleraClient and paylera.AsyncPayleraClient are twins. Same constructor signature. Same method names. Same return shapes. The only difference is whether each method is callable or awaitable — and which httpx transport sits underneath (sync httpx.Client vs httpx.AsyncClient).

That sounds like a coin flip, but it isn’t. The choice is dictated by the runtime your code already runs in.

Rules of thumb

Your runtime	Use
FastAPI / Starlette / any ASGI app	AsyncPayleraClient
Django ASGI (Daphne / Uvicorn) inside an async def view	AsyncPayleraClient
Django WSGI / DRF view / Celery / RQ worker / cron script / one-shot CLI	PayleraClient
Inside a webhook handler from paylera_webhooks(...)	AsyncPayleraClient (handlers run on an event loop)
Inside a webhook handler from paylera_webhook_urls(...) (Django)	Either — handlers can be async def or def. Match the surrounding code.
Jupyter notebooks	PayleraClient — notebooks block on each cell anyway
Pytest test functions	Either; match the function shape (async def test_x(): → async client)

The single tie-breaker: don’t mix. Calling sync requests-style code from inside an event loop blocks the loop and silently destroys concurrency. Using asyncio.run() from inside a synchronous Django view spawns a fresh loop per request and burns CPU.

Sync example

from paylera import PayleraClient

with PayleraClient(api_token="pl_test_...") as client:
    ent = client.get_entitlements("sub_a7e2...")
    print(ent["features"])

with is the lifecycle: the SDK opens the underlying httpx.Client on entry and closes it on exit. Reusing one client across many calls is faster than opening a fresh one per call — keep the with block as wide as your unit of work.

Async example

import asyncio
from paylera import AsyncPayleraClient


async def main() -> None:
    async with AsyncPayleraClient(api_token="pl_live_...") as client:
        ent = await client.get_entitlements("sub_a7e2...")
        print(ent["features"])


asyncio.run(main())

async with is the async lifecycle. Always close with await client.aclose() if you’re not using async with.

Inside FastAPI

from fastapi import FastAPI
from paylera import AsyncPayleraClient

app = FastAPI()
client: AsyncPayleraClient | None = None


@app.on_event("startup")
async def boot() -> None:
    global client
    client = AsyncPayleraClient(api_token=os.environ["PAYLERA_API_TOKEN"])


@app.on_event("shutdown")
async def shutdown() -> None:
    assert client is not None
    await client.aclose()


@app.post("/usage/{customer_id}")
async def track(customer_id: str, payload: dict[str, Any]) -> dict[str, Any]:
    assert client is not None
    return await client.track(
        customer_id=customer_id,
        feature_code=payload["feature_code"],
        value=payload["value"],
    ) or {}

One client lives for the lifetime of the process. FastAPI handlers reuse it — no per-request connection setup.

Inside Django (WSGI)

views.py

from django.http import HttpRequest, JsonResponse
from paylera import PayleraClient

_client = PayleraClient(api_token=os.environ["PAYLERA_API_TOKEN"])


def track(request: HttpRequest) -> JsonResponse:
    result = _client.track(
        customer_id=request.GET["customer_id"],
        feature_code=request.GET["feature_code"],
        value=int(request.GET.get("value", 1)),
    )
    return JsonResponse(result or {})

A module-level sync client is fine under WSGI — each worker process owns its own. Don’t share it across process boundaries; httpx.Client is thread-safe but not fork-safe.

Inside a Celery worker

from celery import Celery
from paylera import PayleraClient

celery = Celery(...)
_client = PayleraClient(api_token=os.environ["PAYLERA_API_TOKEN"])


@celery.task
def report_usage(customer_id: str, feature_code: str, value: int) -> None:
    _client.track(customer_id=customer_id, feature_code=feature_code, value=value)

Celery prefork-by-default forks after the task module imports. Construct the client at module scope and Celery will give each worker its own copy.

Performance characteristics

Both clients use the same retry + idempotency + telemetry machinery. The async client wins when:

• You’re already on an event loop (no asyncio.run overhead).
• You’re fanning out many requests in parallel — asyncio.gather over AsyncPayleraClient calls is dramatically faster than the sync equivalent with a thread pool.

The sync client wins when:

• The surrounding code is sync. Don’t pay the loop bring-up cost.
• You’re issuing one call per request and the request handler is sync — the wall-clock difference is rounding-error.

Don’t do this

# BAD: spawning a loop per request inside a sync view.
def track(request):
    return asyncio.run(_async_client.track(...))     # creates + tears down a loop every call

# BAD: blocking the loop with a sync client inside async code.
async def track(request):
    return _sync_client.track(...)                   # blocks the loop until httpx returns

If your surrounding code is async, the client is async. If your surrounding code is sync, the client is sync. There’s no third option.