A FastAPI equivalent for verify-events-came-from-stripe.
import os
from http import HTTPStatus
from typing import Annotated
import stripe
from fastapi import Depends, FastAPI, Header, HTTPException, Request
app = FastAPI()
async def get_body(request: Request) -> bytes:
return await request.body()
@app.post("/webhook", status_code=HTTPStatus.NO_CONTENT)
def post_report(
stripe_signature: Annotated[str, Header(alias="stripe-signature")],
body: bytes = Depends(get_body),
) -> None:
endpoint_secret = os.environ["ENDPOINT_SECRET"]
try:
# signature validation
event = stripe.Webhook.construct_event(body, stripe_signature, endpoint_secret)
except ValueError as e:
# Invalid payload
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST) from e
except stripe.error.SignatureVerificationError as e:
# Invalid signature
raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY) from e
print(event)
return
A possible catch, one that I ran into, is that I had FastAPI convert the request.body
to a dictionary in the function parameters (so body: dict
). Then I serialized it to a string for the validation step… and it failed because it was no longer identical to what came in.