core

Async Stripe API client generated from Stripe’s OpenAPI spec, with object conversion, pagination helpers, and webhook verification.

env_fn = Path('../.env')
if env_fn.exists(): os.environ.update(parse_env(fn=env_fn))

---

source

stripe_group

def stripe_group(
    oid, path, verb, ptags, optags
):

Call self as a function.

pspec

SpecParser(base_url='https://api.stripe.com/', ops=619)

---

source

StripeSignatureError

def StripeSignatureError(
    *args, **kwargs
):

Common base class for all non-exit exceptions.

---

source

StripeError

def StripeError(
    e
):

Common base class for all non-exit exceptions.

Stripe’s API returns JSON dictionaries, but faststripe converts those responses into Python objects. That means you can write customer.email instead of customer['email'], while still keeping normal dictionary behavior.

The object classes are created dynamically from Stripe’s OpenAPI spec. Each Stripe schema declares an object value, and faststripe uses that to build a matching Python class. Those classes inherit from StripeObject, which inherits from AttrDict, so nested responses get convenient dot access without losing their original structure.

---

source

s2obj

def s2obj(
    x
):

Call self as a function.

---

source

StripeObject

def StripeObject(
    *args, **kwargs
):

dict subclass that also provides access to keys as attrs, and has a pretty markdown repr

---

source

camel

def camel(
    s
):

Call self as a function.

StripeApi

---

source

StripeApi

def StripeApi(
    api_key:NoneType=None, webhook_key:NoneType=None, publishable_key:NoneType=None, service_name:str='faststripe',
    headers:NoneType=None, timeout:float=60.0
):

Client built from OpenAPI operation metadata; async by default, blocking with sync=True.

---

source

StripeTransport

def StripeTransport(
    timeout:float=60.0, client:NoneType=None, base_headers:NoneType=None, follow_redirects:bool=True,
    verify:bool=True
):

Thin async transport over httpx2. By default each request gets a fresh client, so nothing is tied to an event loop; pass client= to use a persistent client that you manage (and close via aclose).

Create a StripeApi instance to get a generated async client for the Stripe API. By default it reads credentials from the environment, builds auth headers, creates a transport, and attaches endpoint groups such as v1 and v2.

sapi = StripeApi()
sapi.groups

{'v1': <fastspec.oapi.OpGroup>,
 'v2': <fastspec.oapi.OpGroup>}

The generated client mirrors Stripe’s URL structure with attributes. For example, sapi.v1.account.get() calls GET /v1/account, and the response comes back as a StripeObject with dot access.

acct = await sapi.v1.account.get()
acct

Account(id=acct_1Q4H44KGhqIw9PXm)

You also get nice error messages when things go wrong.

try:
    await sapi.v1.prices.post(product='abc')
    assert False, 'Should raise StripeError'
except StripeError as e: print(e.msg, e.param, e.type)

No such product: 'abc' product invalid_request_error

The next few cells walk through a small real Stripe flow: create a test product, update it, attach a price, and use that price in a Checkout Session. These examples exercise both request encoding and response object conversion.

prod = await sapi.v1.products.post(name='Test Product')
prod.id, prod.name

('prod_UbdvjLmvG6gLqD', 'Test Product')

prod = await sapi.v1.products.id.post(id=prod.id, name='New')    # update
prod.name

'New'

price = await sapi.v1.prices.post(product=prod.id, unit_amount=10_00, currency='usd')
price.id, price.unit_amount, price.currency

('price_1TcQddKGhqIw9PXmRQn6J4Kb', 1000, 'usd')

Now we can create our checkout session with a mode of payment which means that it will only happen once and is not part of any sort of subscription.

checkout = await sapi.v1.checkout.sessions.post(mode='payment', line_items=[dict(price=price.id, quantity=1)],
                                                success_url='https://localhost:5001/success', cancel_url='https://localhost:5001/cancel')
print(f'Payment link: {checkout.url[:64]}...')

Payment link: https://billing.answer.ai/c/pay/cs_test_a1covGnywAwhXKG7huU7H0Ud...

Pagination

The paged function is the low-level pagination helper. It calls a Stripe list operation, yields each response page, then keeps requesting the next page with starting_after until Stripe reports that there are no more results.

Because it yields whole pages, callers still have access to page metadata such as has_more, along with the page’s data list.

---

source

paged

def paged(
    oper, *args, **kwargs
):

Iterate through all pages of a Stripe API operation.

ps = L()
async for p in paged(sapi.v1.customers.get, limit=2):
    ps.append(p)
    if len(ps) == 2: break

cs = L(c for p in ps for c in p.data)
test_eq(len(ps), 2)
test_eq(len(cs), 4)

The pages function is the convenience wrapper for cases where you want the items, not the page objects. It consumes paged, pulls each item out of each page’s data list, and returns one flat L of Stripe objects.

Use this when you expect the result set to be reasonably small. For very large collections, iterate with paged instead so you can process one page at a time.

---

source

pages

async def pages(
    oper, *args, **kwargs
):

Retrieve all items from all pages of a Stripe API operation.

Let’s test the pagination with the coupons endpoint.

coupons = await pages(sapi.v1.coupons.get, limit=100)
len(coupons), coupons[0].keys()

(593,
 dict_keys(['id', 'object', 'amount_off', 'created', 'currency', 'duration', 'duration_in_months', 'livemode', 'max_redemptions', 'metadata', 'name', 'percent_off', 'redeem_by', 'times_redeemed', 'valid']))

Webhooks

The webhook helpers handle the other direction of the Stripe integration. API calls start in Python and go out to Stripe; webhooks start at Stripe and arrive as signed HTTP requests. The main job here is to verify that the payload really came from Stripe, then convert the event body into the same StripeObject style used by normal API responses.

For the example below, we create a small FastHTML app and run it locally in the notebook. Stripe can then forward webhook events to /webhook, which lets the notebook test the full request path instead of only testing helper functions in isolation.

from fasthtml.common import *
from fasthtml.jupyter import *

app,rt = fast_app()
if 'server' in globals(): server.stop()
server = JupyUvi(app,port=8000)
Show = partial(HTMX, app=app)

This creates a real customer and then asks Stripe for the matching customer.created event. The resulting payload is used as a realistic webhook body for the signature tests, rather than a hand-written toy dictionary.

email = '[email protected]'
c = await sapi.v1.customers.post(email=email)
evts = await sapi.v1.events.get(limit=10, type='customer.created')
payload = obj2dict(first(evts.data, lambda e: e.data.id == c.id))
payload.keys()

dict_keys(['id', 'object', 'api_version', 'created', 'data', 'livemode', 'pending_webhooks', 'request', 'type'])

verify_webhook checks the Stripe signature header. It rebuilds Stripe’s signed message from the timestamp and raw payload, computes the expected HMAC with the webhook secret, and compares it with the v1 signature. It also rejects old timestamps so a captured webhook cannot be replayed much later.

---

source

verify_webhook

def verify_webhook(
    payload, sig_header, secret, tolerance:int=300
):

Verify a Stripe webhook signature, accepting any of multiple v1 entries

mk_webhook is a test helper. It serializes the payload the same compact way Stripe does, signs it with the webhook secret, and returns both the body and the Stripe-Signature header. The following checks cover the happy path, an expired timestamp, and an invalid signature.

def mk_webhook(payload, secret, t=None):
    "Create Stripe-style webhook payload and signature"
    if t is None: t = int(time.time())
    payload = json.dumps(payload, separators=(',', ':')) if isinstance(payload, dict) else payload
    if hasattr(payload, 'decode'): payload = payload.decode()
    sig = hmac.new(secret.encode(), f'{t}.{payload}'.encode(), hashlib.sha256).hexdigest()
    return payload, f't={t},v1={sig}'

payload_s, sig = mk_webhook(payload, os.environ['STRIPE_WEBHOOK_SECRET'])
verify_webhook(payload_s, sig, os.environ['STRIPE_WEBHOOK_SECRET'])
with ExceptionExpected(StripeSignatureError): verify_webhook(payload_s, sig, os.environ['STRIPE_WEBHOOK_SECRET'], tolerance=0)
with ExceptionExpected(StripeSignatureError): verify_webhook(payload_s, 't=123,v1=abc', os.environ['STRIPE_WEBHOOK_SECRET'], tolerance=1)

The Stripe CLI forwards real webhook events to the local FastHTML route. This is useful while developing because the notebook can receive the same signed requests that a deployed app would receive.

%%bash
stripe listen --forward-to http://localhost:8000/webhook > /tmp/stripe.log 2>&1 &

parse_webhook is the method an app route normally calls. It reads the raw request body, verifies the signature header against self.webhook_key, parses the JSON, and converts the event into StripeObject form. Keeping this on StripeApi means webhook handling uses the same credentials and object conversion as the rest of the client.

---

source

StripeApi.parse_webhook

async def parse_webhook(
    req
):

Call self as a function.

Some applications share one Stripe account across more than one service. filter_evt gives each StripeApi instance a simple routing check: if service_name is set, the event must carry matching STRIPE_SERVICE_NAME metadata. If service_name is falsey, every event is accepted.

---

source

StripeApi.filter_evt

def filter_evt(
    evt
):

Return True if evt belongs to this StripeApi service.

The route itself stays small: parse and verify the webhook, then handle the resulting event object. The final payment intent call gives Stripe something real to send back through the listener, so the printed output should show both the event and the payment intent object.

@rt
async def webhook(req):
    'Handle incoming webhooks from stripe'
    evt = await sapi.parse_webhook(req)
    print(evt, evt.data)

Should print the event and payment intent objects.

pi = await sapi.v1.payment_intents.post(customer=c.id, amount=10_00, currency='usd')
pi

PaymentIntent(id=pi_3TcQdzKGhqIw9PXm1M6IXQEq)

Event(id=evt_3TcQdzKGhqIw9PXm1cJa318w) PaymentIntent(id=pi_3TcQdzKGhqIw9PXm1M6IXQEq)