FastAPI Stripe Webhook Template

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