Python

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

Install

Or uv add billkit-eu. Requires Python 3.11 or newer. It depends on httpx (0.28 or later, below 1.0), pydantic 2.7 or later, and anyio, which httpx already installs. AsyncBillKit runs under asyncio or trio.

The package follows semantic versioning from 1.0.0: nothing in a 1.x release breaks code written against an earlier 1.x. Coming from 0.x, read Upgrading from 0.x.

Initialise

Two clients, same surface. BillKit is synchronous; AsyncBillKit has the same methods with await. Omit api_key and it reads BILLKIT_API_KEY; with neither, or with an empty api_key, the constructor raises ValueError. The default timeout is 30 seconds, 10 of them for the connection. Both clients are context managers that close their connection pool on exit, or call close() yourself.

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")

To control TLS, proxies or pooling, pass your own httpx.Client (or httpx.AsyncClient) as httpx_client=. The SDK does not close a client you passed in. A timeout= you pass to the constructor applies to every request through that client too; leave it out and your client’s own default stands. httpx is part of this SDK’s public surface for these two arguments, which is why its major version is capped.

Making calls

One attribute per resource family, each mirroring the routes under /v1/<resource>.

retrieve_pdf and tenant.export() return bytes. expand=[...] is accepted by list (and iter) on customers, products, subscriptions, payments, invoices and events, and by retrieve on products, subscriptions, payments and invoices. The live event stream has no method here; call that route over HTTP.

The object id is positional; every other argument is keyword-only and named exactly as the API names it.

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

Responses

Every method returns a typed object from billkit.types: Customer, Subscription, Invoice and so on, built from the API’s published OpenAPI document. Read a field as an attribute, with your editor and type checker knowing its type, or as a key, exactly as 0.x dicts were read. Both styles are permanent and neither is deprecated.

sub = client.subscriptions.retrieve("sub_9XKp2vQ1")

sub.status            # "active", typed as str
sub["status"]         # the same value, read as a key
sub.get("discount")   # None when the subscription has none
sub.to_dict()         # plain JSON types, all the way down

Responses are never validated, so the API adding a field, or a value to an enum, can never make a call fail. A field this SDK version does not know is still there and kept by to_dict(); read it as obj["new_field"], which a type checker accepts (obj.new_field also works at runtime). Enum-like fields are plain str; the values this version knows are listed as constants you can compare against, such as billkit.types.SubscriptionStatus.PAST_DUE. Treat any other value as “something newer”, not as an error.

An object is a read-only mapping ("discount" in sub, dict(sub), iterating its keys) but not a dict: code that checked isinstance(x, dict) should check collections.abc.Mapping, or call to_dict(). model_dump(), model_dump_json() and FastAPI’s jsonable_encoder return the same as to_dict(): the keys the API sent, with an event’s data and a page’s items intact. Two objects are equal when they are the same class and their to_dict() is equal; an object never equals a dict. Objects, pages and errors pickle.

A field whose name would shadow a model method (items, keys, values, get, to_dict, copy, dict, json, model_*) or is a Python keyword is the attribute obj.items_, and stays obj["items"] as a key.

Auto-pagination

list() returns one page, a ListObject: iterating it yields the items, len(page) counts them and item in page looks for one. A page is not a mapping; the envelope is page.data and page.has_more, and string keys (page["data"], page.get("has_more")) still work. iter() walks the has_more and starting_after cursors and yields one object at a time. Paging is forward-only. iter() takes every filter list() takes, carries them onto every page, and takes limit as the page size and a starting_after to resume from.

for customer in client.customers.iter(provisional=False, expand=["stats"]):
  print(customer.id)

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

Per-request options

Every method takes four more keyword arguments after its own: idempotency_key (on calls that change something), timeout (seconds or an httpx.Timeout, None for no timeout), max_retries (0 sends one attempt) and extra_headers. They apply to that call only. extra_headers cannot set Authorization, Idempotency-Key, Content-Type or User-Agent in any letter case; those raise ValueError, and the key is passed as idempotency_key=.

Two more reach a field this SDK version has no keyword for yet, such as a filter a later 1.x API adds: extra_query (every method) adds query parameters, and extra_body (every POST method) adds top-level body fields. Both are merged over what the keywords built, so a key there wins. A None in extra_query is dropped; in extra_body it is sent as JSON null. The SDK does not check either, so a field the route does not take comes back as an InvalidRequestError with code parameter_unknown.

product = client.products.create(
  name="Pro",
  idempotency_key="product-pro-v1",
  timeout=10.0,
  max_retries=2,
  extra_headers={"X-Request-Source": "admin-panel"},
)

Idempotency and retries

Connection errors, timeouts and 5xx other than 501 are retried with jittered exponential backoff. A 429 is retried only when its Retry-After is within max_retry_after_seconds (30 by default), and a 409 idempotency_in_progress is retried because the first request may already have charged. The default policy makes four attempts in total, starting at 0.5 seconds and doubling up to 8 seconds between them.

Every mutating call carries an Idempotency-Key, generated once per call and reused on every retry, 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

Every error from a request extends BillKitError; an argument mistake caught before anything is sent (a blank id, max_retries=-1, a reserved header in extra_headers) raises ValueError or TypeError. An HTTP error response is an APIError, and the class is picked by status, not by the envelope’s type: 401 is AuthenticationError, 402 PaymentError (a refused charge, code payment_refused, or a suspended account, tenant_suspended), 403 PermissionDeniedError, 404 NotFoundError, 409 ConflictError, 429 RateLimitError, 5xx ServerError, and 400, 405 or any other 4xx InvalidRequestError. A 2xx whose body is not JSON (a proxy answering in the API’s place) is also raised, as a plain APIError. No answer at all is APIConnectionError, and APITimeoutError when it was a timeout. An APIError carries status_code, message, type, code, param, reason (the finer-grained cause some codes have, such as renewal_processing on a 409 operation_pending), request_id, raw_body (the decoded body, or its text when it was not JSON) and headers; RateLimitError adds retry_after in seconds. Every error pickles with its attributes, so it survives a task queue.

from billkit import BillKitError, NotFoundError, RateLimitError

try:
  client.customers.retrieve("cus_gone")
except NotFoundError:
  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

Erasing a customer

customers.purge erases a customer’s personal data for a GDPR request and cannot be undone, so it requires confirmed=True. Without it the SDK raises before sending anything. The answer is a PurgedResource, the deletion marker {"id", "object": "customer", "deleted": True, "purged": True} rather than the customer, and retrieving the customer afterwards is a NotFoundError. delete() is different: it removes the customer from the API and keeps their data.

purged = client.customers.purge(customer.id, confirmed=True)

Webhooks

from billkit import WebhookSignature, WebhookVerificationError, types

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 isinstance(event, types.SubscriptionCreatedEvent):
  provision(event.data)  # a typed Subscription

verify returns the event’s own class when this SDK version knows its type, with data typed to match, and a plain billkit.Event otherwise. Ignore a type you do not handle and answer 2xx; a non-2xx makes BillKit retry a delivery you were never going to process. event.previous_attributes is always present: on an update it holds the changed fields with their values before the change, and it is None when nothing changed, so test it for None rather than for presence.

The signature is compared in constant time, and a timestamp more than five minutes from now is rejected to stop replays. Pass tolerance_seconds= to change the window. The header can carry several v1 signatures while you rotate a secret; any one matching is enough.

Logging

The SDK writes to one logger, logging.getLogger("billkit"), which carries a NullHandler and no level. It never calls basicConfig. Set a level on it to opt in.

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.

Upgrading from 0.x

1.0 is a deliberate break, made once. The changes that need an edit:

• Responses are typed objects rather than dicts. Key access (obj["id"], obj.get("status")) still works; isinstance(x, dict) does not, so check Mapping or call to_dict(), and compare with obj.to_dict() == {...}. A list page is not a mapping at all: iterating it yields items, so use page.to_dict() where you used dict(page).
• APIError is now the parent of every HTTP error, not just 5xx. Code that caught APIError to mean “the server failed” should catch ServerError.
• Renamed: PermissionError is PermissionDeniedError, ResourceMissingError is NotFoundError, and tenant.capabilities(), tenant.portal_branding() and tenant.billing_profile() are retrieve_capabilities(), retrieve_portal_branding() and retrieve_billing_profile().
• customers.purge(id) needs confirmed=True, and returns a PurgedResource marker rather than the customer.
• method is payment_method on checkout sessions and one-off payments (checkout_sessions.create(payment_method=...)), test_clock is test_clock_id on customers, and a dispute’s reversed is reversed_at.
• The status and renewal_state filters on subscriptions, status on disputes and type on events take a list, sent comma-joined: subscriptions.list(status=["active", "past_due"]). One string is still sent as given. percent_off takes up to two decimal places and reads back as the API wrote it: an int for a whole percent (99), a float otherwise (12.5).
• iter(page_size=...) is iter(limit=...), the API’s own name for the page size.
• coupons.create takes percent_off or amount_off_cents instead of discount_type plus discount_value.
• A price tier’s amounts are unit_amount_cents and flat_amount_cents.
• WebhookSignature.verify returns a typed event, and raises WebhookVerificationError (now a BillKitError) for a body that is not a JSON object. An empty secret raises ValueError.
• billkit.WebhookEvent is gone: remove the import and annotate with billkit.Event or the specific event class.

The changelog has every change with before-and-after code.