Python

Sync and async clients over httpx, with typed errors and cursor auto-pagination.

Install

Or uv add billkit-eu. Requires Python 3.11 or newer, and pulls in httpx.

Initialise

Two clients, same surface. BillKit is synchronous; AsyncBillKit is the same methods with await and an async context manager. Omit api_key and it reads BILLKIT_API_KEY.

from billkit import BillKit

client = BillKit(
  api_key="bk_test_...",
  base_url="https://api.billkit.eu",
  timeout=30.0,
)

from billkit import AsyncBillKit

async with AsyncBillKit(api_key="bk_test_...") as client:
  customer = await client.customers.create(email="ada@example.com")

Making calls

One attribute per resource family: client.customers, products, prices, checkout_sessions, one_shot_payments, subscriptions, refunds, disputes, webhook_endpoints, events, tenant, coupons, tax_rates, invoices, audit_logs, payments, billing_portal_sessions.

Arguments are keyword-only and named exactly as the API names them. Every method returns the decoded JSON body as a dict, so nothing is lost in translation on its way to your own models.

price = client.prices.create(
  product_id="prod_9XKp2vQ1",
  amount_cents=999,
  currency="EUR",
  interval="month",
  trial_days=14,
)
print(price["id"])

Auto-pagination

iter() walks the has_more and starting_after cursor protocol. Paging is forward-only.

for customer in client.customers.iter():
  print(customer["id"])

for event in client.events.iter(type="subscription.created", page_size=100):
  handle(event)

Idempotency and retries

Connection errors, 5xx and short Retry-After 429s are retried with jittered backoff. Every mutating call carries an auto-generated Idempotency-Key, so a retried write replays the first response instead of repeating it. Pass idempotency_key= to make that hold across process restarts.

from billkit import BillKit, RetryPolicy

client = BillKit(
  api_key="bk_test_...",
  retry_policy=RetryPolicy(max_attempts=5, max_retry_after_seconds=10.0),
)

Errors

Everything inherits from BillKitError: APIConnectionError, APIError, ServerError, AuthenticationError, PermissionError, ResourceMissingError, InvalidRequestError, ConflictError, RateLimitError. The class is picked from the error envelope’s type, falling back to the HTTP status when the envelope does not name one. Each carries status_code, message, code, param and request_id.

from billkit import BillKitError, RateLimitError, ResourceMissingError

try:
  client.customers.retrieve("cus_gone")
except ResourceMissingError:
  return None
except RateLimitError as exc:
  retry_in(exc.retry_after)
except BillKitError as exc:
  raise RuntimeError(f"BillKit {exc.status_code}: {exc.message}") from exc

Webhooks

from billkit import WebhookSignature, WebhookVerificationError

try:
  event = WebhookSignature.verify(
      payload=request.body,  # raw bytes, never the parsed dict
      signature_header=request.headers.get("BillKit-Signature"),
      secret=os.environ["BILLKIT_WEBHOOK_SECRET"],
  )
except WebhookVerificationError:
  return Response(status_code=400)

if event["type"] == "subscription.created":
  provision(event["data"])

Constant-time HMAC compare and a five-minute timestamp tolerance for replay protection. Pass tolerance_seconds= to change it.

Logging

The SDK owns one logger, logging.getLogger("billkit"), with a NullHandler on it. It never calls basicConfig, never sets a level, and never touches a logger it does not own.

import logging

logging.basicConfig()
logging.getLogger("billkit").setLevel(logging.DEBUG)

DEBUG gives one line per attempt and per response; WARNING one per retry. Your API key, request and response bodies and the query string are never logged.