}
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`](https://AnswerDotAI.github.io/faststripe/core.html#stripeobject)
with dot access.
``` python
acct = await sapi.v1.account.get()
acct
```
``` python
Account(id=acct_1Q4H44KGhqIw9PXm)
```
You also get nice error messages when things go wrong.
``` python
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.
``` python
prod = await sapi.v1.products.post(name='Test Product')
prod.id, prod.name
```
('prod_UbdvjLmvG6gLqD', 'Test Product')
``` python
prod = await sapi.v1.products.id.post(id=prod.id, name='New') # update
prod.name
```
'New'
``` python
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.
``` python
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`](https://AnswerDotAI.github.io/faststripe/core.html#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
``` python
def paged(
oper, *args, **kwargs
):
```
*Iterate through all pages of a Stripe API operation.*
``` python
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`](https://AnswerDotAI.github.io/faststripe/core.html#pages)
function is the convenience wrapper for cases where you want the items,
not the page objects. It consumes
[`paged`](https://AnswerDotAI.github.io/faststripe/core.html#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`](https://AnswerDotAI.github.io/faststripe/core.html#paged)
instead so you can process one page at a time.
------------------------------------------------------------------------
source
### pages
``` python
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.
``` python
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`](https://AnswerDotAI.github.io/faststripe/core.html#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.
``` python
from fasthtml.common import *
from fasthtml.jupyter import *
```
``` python
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.
``` python
email = 'test@example.com'
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`](https://AnswerDotAI.github.io/faststripe/core.html#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
``` python
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.
``` python
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
%%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`](https://AnswerDotAI.github.io/faststripe/core.html#stripeobject)
form. Keeping this on
[`StripeApi`](https://AnswerDotAI.github.io/faststripe/core.html#stripeapi)
means webhook handling uses the same credentials and object conversion
as the rest of the client.
------------------------------------------------------------------------
source
### StripeApi.parse_webhook
``` python
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`](https://AnswerDotAI.github.io/faststripe/core.html#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
``` python
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.
``` python
@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.
``` python
pi = await sapi.v1.payment_intents.post(customer=c.id, amount=10_00, currency='usd')
pi
```
``` python
PaymentIntent(id=pi_3TcQdzKGhqIw9PXm1M6IXQEq)
```
Event(id=evt_3TcQdzKGhqIw9PXm1cJa318w) PaymentIntent(id=pi_3TcQdzKGhqIw9PXm1M6IXQEq)