PROJECT
OVERVIEW
Used FastAPI to asynchronously process high-concurrency third-party Webhooks (Stripe), and strictly verified digital signatures to defend against forged requests.
Rotate device for better view ↺
Used FastAPI to asynchronously process high-concurrency third-party Webhooks (Stripe), and strictly verified digital signatures to defend against forged requests.
# _*_coding : utf-8 _*_
from fastapi import FastAPI, Request, Header, HTTPException
import stripe
import logging
app = FastAPI()
stripe.api_key = "YOUR_STRIPE_API_KEY"
logger = logging.getLogger("webhook_server")
@app.post("/YOUR_WEBHOOK_SECRET_PATH/stripe-notify")
async def handle_stripe_webhook(request: Request, stripe_signature: str | None = Header(None)):
'''处理 Stripe 支付回调，验证签名并发放权益'''
payload = await request.body()
try:
event = stripe.Webhook.construct_event(
payload=payload, sig_header=stripe_signature, secret="YOUR_WEBHOOK_SECRET"
)
except Exception as e:
logger.error(f"Stripe Webhook 签名验证失败: {e}")
raise HTTPException(status_code=400, detail="Invalid signature")
if event['type'] == 'checkout.session.completed':
session = event['data']['object']
telegram_user_id = session.get('client_reference_id')
if telegram_user_id:
# 更新共享数据库，完成发货
database.grant_membership(int(telegram_user_id), days=30, tier=1)
logger.info(f"Stripe 支付成功！已为 User {telegram_user_id} 发货。")
return {"status": "received"}