如何将Stripe Webhook与付费获取测验结果的用户的Flask Session信息关联？

直接让Stripe Webhook访问用户的Flask Session是不可行的——原因很简单：Webhook请求是从Stripe的服务器发起的，完全没有用户浏览器携带的Cookie，自然无法关联到用户的Flask Session实例。不过这个问题有成熟的解决方案，核心思路是把用户的测验答案和Stripe的支付会话绑定起来，下面是具体的实现方案：

这是最可靠、扩展性最强的方案，适合大多数场景：

步骤说明 #

1. 用户完成测验后，先持久化答案：将用户存在Flask Session里的测验答案存入数据库，生成一条唯一的测验记录（包含答案、状态等字段）。
2. 创建Stripe会话时绑定记录ID：在生成Stripe Checkout Session时，把测验记录的ID存入Stripe会话的metadata字段——Stripe会在Webhook事件中带回这个元数据。
3. Webhook触发时关联记录处理：当收到Stripe的checkout.session.completed事件时，从事件对象中取出测验记录ID，再从数据库中找到对应答案进行分析。

代码示例 #

生成Stripe Checkout Session的路由

from flask import Flask, session, redirect, request
import stripe
from your_db_module import db, QuizRecord  # 假设用SQLAlchemy作为ORM

app = Flask(__name__)
app.secret_key = "your-flask-secret-key"
stripe.api_key = "your-stripe-secret-key"

@app.route("/create-checkout-session", methods=["POST"])
def create_checkout_session():
    # 从用户Session中取出测验答案
    quiz_answers = session.get("quiz_answers")
    if not quiz_answers:
        return "No quiz answers found", 400

    # 存入数据库，生成测验记录
    quiz_record = QuizRecord(answers=quiz_answers, status="pending")
    db.session.add(quiz_record)
    db.session.commit()

    # 创建Stripe会话，绑定测验记录ID到metadata
    checkout_session = stripe.checkout.Session.create(
        payment_method_types=["card"],
        line_items=[
            {
                "price_data": {
                    "currency": "usd",
                    "product_data": {"name": "Quiz Result Analysis"},
                    "unit_amount": 500,  # 5美元，单位为分
                },
                "quantity": 1,
            }
        ],
        mode="payment",
        success_url="https://your-domain.com/thank-you",
        cancel_url="https://your-domain.com/cancel",
        metadata={"quiz_record_id": str(quiz_record.id)}
    )

    return redirect(checkout_session.url, code=303)

Webhook处理路由

@app.route("/stripe-webhook", methods=["POST"])
def stripe_webhook():
    payload = request.get_data(as_text=True)
    sig_header = request.headers.get("Stripe-Signature")
    webhook_secret = "your-stripe-webhook-secret"

    # 验证Stripe签名，防止恶意请求
    try:
        event = stripe.Webhook.construct_event(payload, sig_header, webhook_secret)
    except ValueError:
        return "Invalid payload", 400
    except stripe.error.SignatureVerificationError:
        return "Invalid signature", 400

    # 处理支付完成事件
    if event["type"] == "checkout.session.completed":
        session_obj = event["data"]["object"]
        quiz_record_id = session_obj["metadata"].get("quiz_record_id")
        
        if not quiz_record_id:
            return "Missing quiz record ID", 400
        
        # 找到对应的测验记录
        quiz_record = QuizRecord.query.get(quiz_record_id)
        if not quiz_record:
            return "Quiz record not found", 404
        
        # 幂等性处理：避免重复分析
        if quiz_record.status == "completed":
            return "Already processed", 200
        
        # 执行你的答案分析逻辑
        analyze_quiz_answers(quiz_record.answers)
        
        # 更新记录状态为已完成
        quiz_record.status = "completed"
        db.session.commit()

    return "Success", 200

如果你的测验答案数据量很小（比如只有几道选择题的答案），也可以跳过数据库，把答案加密后直接存入Stripe会话的metadata字段。但要注意Stripe对单个对象的metadata有500KB的大小限制，且必须保证加密安全。

代码示例 #

from cryptography.fernet import Fernet

# 生成并保存加密密钥（不要硬编码到代码中，用环境变量存储）
cipher = Fernet(b"your-encryption-key")

@app.route("/create-checkout-session", methods=["POST"])
def create_checkout_session():
    quiz_answers = session.get("quiz_answers")
    if not quiz_answers:
        return "No quiz answers found", 400

    # 加密答案
    encrypted_answers = cipher.encrypt(str(quiz_answers).encode())

    checkout_session = stripe.checkout.Session.create(
        # 其他参数同方案一
        metadata={"encrypted_answers": encrypted_answers.decode()}
    )

    return redirect(checkout_session.url, code=303)

# Webhook处理时解密
@app.route("/stripe-webhook", methods=["POST"])
def stripe_webhook():
    # 签名验证步骤省略...
    if event["type"] == "checkout.session.completed":
        session_obj = event["data"]["object"]
        encrypted_answers = session_obj["metadata"].get("encrypted_answers")
        
        if not encrypted_answers:
            return "Missing encrypted answers", 400
        
        # 解密答案并转换为原格式
        quiz_answers = eval(cipher.decrypt(encrypted_answers.encode()).decode())
        # 执行分析逻辑
        analyze_quiz_answers(quiz_answers)

    return "Success", 200

• 必须验证Webhook签名：这是防止恶意请求的核心步骤，绝对不能省略。
• 保证幂等性：Stripe可能会因网络问题重试Webhook事件，所以要确保你的分析逻辑不会重复执行（比如检查记录状态）。
• 优先选择数据库方案：方案一更适合长期维护和扩展，方案二仅作为小数据量场景的临时替代。
• 清理过期记录：定期清理数据库中状态为pending的旧测验记录（比如用户取消支付的情况）。

内容的提问来源于stack exchange，提问作者Anonymous12332313