如何在Flask项目中通过Stripe判断用户订阅是否激活？

核心思路是将用户订阅状态本地化存储（数据库/缓存），通过Stripe Webhook同步状态变更，页面访问时直接读取本地数据判断，彻底避免每次请求调用Stripe API。以下是具体实现步骤：

1. 本地存储订阅状态 #

首先在用户模型中添加订阅状态字段，关联Stripe客户ID：

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    stripe_customer_id = db.Column(db.String(100), unique=True)
    subscription_status = db.Column(db.String(20), default="inactive")
    # 其他用户字段...

2. 配置Stripe Webhook同步状态 #

监听Stripe订阅相关事件，实时更新本地用户的订阅状态，同时必须验证Stripe签名确保请求合法：

import stripe
from flask import request, jsonify

# 替换为你的Stripe Webhook密钥
STRIPE_WEBHOOK_SECRET = "your_webhook_secret"

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

    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, STRIPE_WEBHOOK_SECRET
        )
    except ValueError:
        # 无效payload
        return jsonify(success=False), 400
    except stripe.error.SignatureVerificationError:
        # 无效签名
        return jsonify(success=False), 400

    # 处理核心订阅事件
    if event['type'] in ['customer.subscription.created', 'customer.subscription.updated', 'customer.subscription.deleted']:
        subscription = event['data']['object']
        customer_id = subscription['customer']
        
        # 更新本地用户状态
        user = User.query.filter_by(stripe_customer_id=customer_id).first()
        if user:
            user.subscription_status = subscription['status']
            db.session.commit()

    return jsonify(success=True), 200

需要在Stripe后台将该Webhook地址（如https://your-domain.com/stripe-webhook）配置为监听上述三个订阅事件。

3. 改造订阅状态校验逻辑 #

替换原有的check_subs函数，优先读取本地数据库状态，仅在必要时兜底调用Stripe API：

def check_subs(user_id):
    user = User.query.get(user_id)
    if not user:
        return 0
    
    # 首次访问或本地状态异常时，同步一次Stripe数据
    if user.subscription_status == "inactive":
        subs = stripe.Subscription.list(customer=user.stripe_customer_id)
        if subs['data']:
            latest_status = subs['data'][0]['status']
            user.subscription_status = latest_status
            db.session.commit()
            return 1 if latest_status == 'active' else 0
        return 0
    
    return 1 if user.subscription_status == 'active' else 0

4. 更新页面路由 #

路由逻辑无需大幅改动，直接复用改造后的校验函数即可：

@app.route('/home')
def home_page():
    if check_subs(session['user_id']) == 0:  # 确保session['user_id']对应用户模型的主键ID
         flash("用户订阅已过期")
         return redirect(url_for('payment_page'))
    return render_template('home.html')

@app.route('/dashboard')
def dashboard_page():
    if check_subs(session['user_id']) == 0:
         flash("用户订阅已过期")
         return redirect(url_for('payment_page'))
    return render_template('dashboard.html')

额外优化建议 #

• 高并发场景下，可将订阅状态存入Redis等缓存，进一步提升读取速度，Webhook更新时同步更新缓存与数据库。
• 每日执行一次全量用户订阅状态同步，避免Webhook漏接事件导致的状态不一致。

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