Python SDK

bachs is the official Bachs library for Python servers. It sends authenticated requests, checks the responses, retries reads safely, and verifies webhook signatures. It has a sync client, Bachs, and an async client, AsyncBachs, with the same methods. It runs on your server only. Keep your secret API key out of browser and mobile code.

Install

You need Python 3.10 or later.

pip install bachs

uv add bachs

poetry add bachs

The package is built on httpx and ships type hints for mypy and Pyright.

Create a client

Create one client and reuse it. Pass your secret key and the environment it belongs to.

import os
from bachs import Bachs

bachs = Bachs(api_key=os.environ["BACHS_API_KEY"], environment="sandbox")

import os
from bachs import AsyncBachs

bachs = AsyncBachs(api_key=os.environ["BACHS_API_KEY"], environment="sandbox")

environment is required. "sandbox" sends requests to https://sandbox-api.bachs.io and "production" sends them to https://api.bachs.io. Use a sandbox key with "sandbox" and a production key with "production". See Authentication. Creating a client sends no request, so a wrong key shows up on your first call.

Option	Type	Default	Description
`api_key`	`str`	Required	Your secret API key.
`environment`	`"sandbox"` \| `"production"`	Required	Which Bachs environment to call.
`base_url`	`str`	Set by `environment`	Send requests to another host, such as a proxy or a local test server.
`timeout`	`float`	`30.0`	Seconds allowed for a whole operation, including retries and reading the response.
`max_retries`	`int`	`2`	Extra attempts for reads. Writes are never retried.
`http_client`	`httpx.Client` or `httpx.AsyncClient`	Created for you	Your own `httpx` client, for example to use a proxy. You close it yourself.

To release connections when you are done, use the client as a context manager (with Bachs(...) as bachs: or async with AsyncBachs(...) as bachs:), or call bachs.close() or await bachs.aclose().

Send requests

Each resource on the client maps to a group of API endpoints. Methods follow one pattern:

Create: create(params, *, idempotency_key=None, timeout=None)
Retrieve: get(id, *, timeout=None, max_retries=None)
List: list(params=None, *, timeout=None, max_retries=None)

product = bachs.products.create(
    {
        "name": "Pro Plan",
        "price": {"currency": "USD", "amount": "12.00"},
        "billing_cycle": {"interval": "month", "frequency": 1},
    },
    idempotency_key="product-pro-monthly",
)

checkout = bachs.checkout_sessions.create(
    {
        "product_cart": [{"product_id": product["id"]}],
        "customer": {"email": "ada@example.com", "name": "Ada Obi"},
        "success_url": "https://your-app.example/success",
    },
    idempotency_key=f"checkout-{order.id}",
)

# Redirect the customer to checkout["checkout_url"].

With AsyncBachs, every method is a coroutine: await bachs.products.create(...). Requests and responses are dictionaries with the same field names as the API, in snake_case, so the API reference describes them exactly. Money amounts are decimal strings, such as "12.00", never numbers. Leaving a key out and sending None are different, as in the API.

Resources

Resource	Methods	API reference
`products`	`create`, `get`, `list`	Products
`checkout_sessions`	`create`, `get`	Checkout sessions
`payments`	`get`, `list`	Payments
`subscriptions`	`get`, `list`, `cancel`	Subscriptions
`customer_portal`	`create_session(customer_id)`	Customer portal sessions
`balances`	`get`	Balance
`banks`	`list`, `resolve_account`	Payouts
`payout_destinations`	`create`, `get`, `list`	Payout destinations
`payout_quotes`	`create`	Payouts
`payouts`	`create`, `get`, `list`	Payouts
`webhooks`	`construct_event`	Verify webhooks

subscriptions.cancel(id, params=None, ...) takes {"cancel_at_period_end": True} to cancel at the end of the paid period, or False to cancel now. See Manage subscriptions.

Not in the SDK yet

Use the API directly for these. You can call it alongside the SDK with the same key.

Connect: connected accounts, onboarding, the X-Account-Id header and split payments
Customers, refunds, disputes and conversions
Updating, archiving or unarchiving products, and changing a subscription’s seats or plan
Virtual accounts, uploads, product groups and account settings

Your type checker reports fields the SDK does not describe. Do not pass them anyway: the SDK sends the dictionary you give it, and the API may reject or ignore unknown fields.

Lists and pagination

List methods return one page, in the same shape as the API response. The SDK does not fetch further pages for you. Products page by cursor. Pass pagination["next_cursor"] as cursor to get the next page:

from bachs.types import ProductListParams

params: ProductListParams = {"limit": 50}

while True:
    page = bachs.products.list(params)
    for product in page["items"]:
        print(product["id"], product["name"])
    next_cursor = page["pagination"].get("next_cursor")
    if not next_cursor:
        break
    params["cursor"] = next_cursor

Payments, subscriptions, payouts and payout destinations page by limit and offset. Payout destinations are in destinations; the others are in items.

page = bachs.payouts.list({"limit": 20, "offset": 40, "status_filter": "completed"})
print(page["total"], len(page["items"]))

See Pagination.

Request options

Every method takes these as keyword arguments after its other arguments.

Option	Applies to	Description
`idempotency_key`	Writes	Sent as the `Idempotency-Key` header. Store it with the operation so you can repeat the same write safely. See Idempotency.
`timeout`	All	Overrides the client’s `timeout` for this call, in seconds.
`max_retries`	Reads	Overrides the client’s `max_retries` for this call.

To cancel an AsyncBachs call, cancel its task, for example with asyncio.timeout() or when your framework drops a disconnected request. The cancellation exception reaches your code as usual. Cancelling stops waiting; it cannot undo a write Bachs already received.

Retries and uncertain writes

Reads are retried up to max_retries times after a network failure or an HTTP 429, 500, 502, 503 or 504. The SDK waits about half a second, then a second, and follows any Retry-After header. It stops when the next wait would pass timeout. Writes are sent once, even with an idempotency key. A write can reach Bachs and still fail on the way back, for example when the connection drops or the API returns a 5xx. In those cases the error has outcome_unknown set to True: the payment, payout or product may exist. When a write’s outcome is unknown, do not send it again with a new key. Look it up first, for example in your payout history or by retrieving the checkout session. If it does not exist, repeat the write with the same idempotency key and the same request.

from bachs import BachsApiError, BachsNetworkError, BachsResponseError

try:
    bachs.payouts.create(params, idempotency_key=operation.key)
except (BachsApiError, BachsNetworkError, BachsResponseError) as err:
    if not err.outcome_unknown:
        raise
    mark_for_reconciliation(operation)

If you cancel an async write after it was sent, treat its outcome as unknown in the same way.

AsyncBachs stops exactly at timeout. The sync client can run past it in one case: when the server sends part of a response and then stops, the last network read waits for its own full timeout, so a call can take up to about twice timeout. If you need a hard limit, use AsyncBachs or set timeout to half of your limit.

Errors

Every SDK error extends BachsError.

Error	When	Useful fields
`BachsApiError`	The API returned an error status.	`status`, `error_code`, `field_errors`, `request_id`, `retry_after`, `outcome_unknown`, `body`
`BachsNetworkError`	The request or response failed in transit. The original exception is `__cause__`.	`outcome_unknown`, `response_status`
`BachsTimeoutError`	The call ran past `timeout`. Extends `BachsNetworkError`.	Same as `BachsNetworkError`
`BachsResponseError`	The API returned success, but the body was not the expected shape.	`status`, `request_id`, `outcome_unknown`, `body`
`BachsConfigError`	The client or a request was set up wrongly, such as a missing `environment`. Nothing was sent.	`str(err)`
`BachsWebhookError`	A webhook failed signature or format checks.	`str(err)`

from bachs import BachsApiError

try:
    session = bachs.checkout_sessions.get(checkout_id)
except BachsApiError as err:
    if err.status == 404:
        session = None
    else:
        raise

error_code is the error_code from the response body, and field_errors lists any problems with individual fields. See Errors. Include request_id when you contact support. Error bodies and customer portal URLs can contain personal data or credentials. Do not log them in full.

Verify webhooks

webhooks.construct_event checks the signature on a delivery and returns the event. Pass the raw request body, before any JSON parsing, the request headers, and your endpoint’s signing secret. The signing secret is not your API key.

import os
from fastapi import FastAPI, Request, Response
from bachs import BachsWebhookError, webhooks

app = FastAPI()

@app.post("/webhooks/bachs")
async def bachs_webhook(request: Request) -> Response:
    body = await request.body()
    try:
        event = webhooks.construct_event(body, request.headers, os.environ["BACHS_WEBHOOK_SECRET"])
    except BachsWebhookError:
        return Response("Invalid signature", status_code=400)

    # Record event["id"] durably before acting, so a redelivery is not processed twice.
    await handle_event(event)
    return Response(status_code=200)

import os
from flask import Flask, request
from bachs import BachsWebhookError, webhooks

app = Flask(__name__)

@app.post("/webhooks/bachs")
def bachs_webhook():
    try:
        event = webhooks.construct_event(
            request.get_data(), request.headers, os.environ["BACHS_WEBHOOK_SECRET"]
        )
    except BachsWebhookError:
        return "Invalid signature", 400

    # Record event["id"] durably before acting, so a redelivery is not processed twice.
    handle_event(event)
    return "", 200

import os
from django.http import HttpResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from bachs import BachsWebhookError, webhooks

@csrf_exempt
@require_POST
def bachs_webhook(request):
    try:
        event = webhooks.construct_event(
            request.body, request.headers, os.environ["BACHS_WEBHOOK_SECRET"]
        )
    except BachsWebhookError:
        return HttpResponseBadRequest("Invalid signature")

    # Record event["id"] durably before acting, so a redelivery is not processed twice.
    handle_event(event)
    return HttpResponse(status=200)

It checks X-Bachs-Signature-V2, accepts any matching signature while you rotate a secret, and rejects deliveries more than 300 seconds old or ahead. Change the window with tolerance=, in seconds. You can call it without a client, or as bachs.webhooks.construct_event.

Flask support for request.headers needs bachs 1.0.1 or later. On 1.0.0, pass dict(request.headers).

It does not deduplicate events or run your handlers. Your app stores processed event IDs and returns a 5xx when handling fails, so Bachs retries. See Webhooks.

Types

Request and response types are TypedDicts in bachs.types:

from bachs.types import CreateProductRequest, ProductResponse, WebhookEvent

Responses are checked at runtime against the fields the types promise, then returned as plain dictionaries. Fields the API adds later stay in the dictionary, so a new field never breaks your code.

Source code

The SDK is open source under the MIT license.

GitHub repository: source, issues and the full README
PyPI package
Changelog

Get Started

Payouts

Virtual accounts

Products & pricing

Subscriptions

Customer portal

Payments

Dashboard

Developer Portal

SDKs

CLI

For You and Your Customers

Install

Create a client

Send requests

Resources

Not in the SDK yet

Request options

Retries and uncertain writes

Errors

Verify webhooks

Types

Source code

​Install

​Create a client

​Send requests

​Resources

​Not in the SDK yet

​Lists and pagination

​Request options

​Retries and uncertain writes

​Errors

​Verify webhooks

​Types

​Source code

Install

Create a client

Send requests

Resources

Not in the SDK yet

Lists and pagination

Request options

Retries and uncertain writes

Errors

Verify webhooks

Types

Source code