如何在Python/Flask中整合Stripe与Firebase实现订阅功能？

1. 安装依赖包 #

先安装所需的Python库：

pip install firebase-admin stripe flask  # 示例用Flask框架，也可替换为FastAPI等

2. 初始化客户端 #

在后端代码中完成Firebase Admin和Stripe的初始化：

import firebase_admin
from firebase_admin import credentials, auth
import stripe

# 初始化Firebase Admin（替换为你的服务账号文件路径）
cred = credentials.Certificate("path/to/your-firebase-service-account.json")
firebase_admin.initialize_app(cred)

# 初始化Stripe（替换为你的Stripe秘钥）
stripe.api_key = "sk_test_your_stripe_secret_key"

3. 关联Firebase用户与Stripe客户 #

每个Firebase认证用户需要对应一个Stripe Customer对象，写工具函数实现创建或获取逻辑：

def get_or_create_stripe_customer(firebase_uid):
    # 获取Firebase用户信息
    firebase_user = auth.get_user(firebase_uid)
    
    # 检查用户是否已关联Stripe客户（存在Firebase自定义声明中）
    if "stripe_customer_id" in firebase_user.custom_claims:
        return firebase_user.custom_claims["stripe_customer_id"]
    
    # 创建新的Stripe客户
    customer = stripe.Customer.create(
        email=firebase_user.email,
        metadata={"firebase_uid": firebase_uid}
    )
    
    # 将Stripe客户ID存入Firebase用户自定义声明
    auth.set_custom_user_claims(firebase_uid, {"stripe_customer_id": customer.id})
    
    return customer.id

4. 创建支付Intent（一次性支付场景） #

处理前端支付请求，生成Stripe的PaymentIntent并返回客户端所需密钥：

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/create-payment-intent", methods=["POST"])
def create_payment_intent():
    data = request.get_json()
    firebase_uid = data.get("firebase_uid")
    amount = data.get("amount")  # 单位为分，如1000代表10美元
    currency = data.get("currency", "usd")
    
    try:
        customer_id = get_or_create_stripe_customer(firebase_uid)
        
        payment_intent = stripe.PaymentIntent.create(
            amount=amount,
            currency=currency,
            customer=customer_id,
            metadata={"firebase_uid": firebase_uid}
        )
        
        return jsonify({"clientSecret": payment_intent.client_secret})
    except Exception as e:
        return jsonify(error=str(e)), 403

5. 处理Stripe Webhooks #

支付/订阅状态变更后，Stripe会发送Webhook通知，后端需验证并处理：

@app.route("/stripe-webhook", methods=["POST"])
def stripe_webhook():
    payload = request.data
    sig_header = request.headers.get("Stripe-Signature")
    webhook_secret = "whsec_your_stripe_webhook_secret"  # 替换为你的Webhook秘钥
    
    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, webhook_secret
        )
    except ValueError as e:
        # 无效payload
        return jsonify(error=str(e)), 400
    except stripe.error.SignatureVerificationError as e:
        # 无效签名
        return jsonify(error=str(e)), 400
    
    # 处理支付成功事件
    if event["type"] == "payment_intent.succeeded":
        payment_intent = event["data"]["object"]
        firebase_uid = payment_intent["metadata"]["firebase_uid"]
        # 更新Firebase用户状态，比如标记已支付
        auth.set_custom_user_claims(firebase_uid, {"has_paid": True})
        
    # 处理订阅创建事件（如需支持订阅）
    elif event["type"] == "customer.subscription.created":
        subscription = event["data"]["object"]
        firebase_uid = subscription["metadata"]["firebase_uid"]
        auth.set_custom_user_claims(firebase_uid, {"is_subscribed": True})
    
    return jsonify(success=True)

6. 订阅场景补充（可选） #

如果需要实现订阅模式，创建订阅的函数示例：

@app.route("/create-subscription", methods=["POST"])
def create_subscription():
    data = request.get_json()
    firebase_uid = data.get("firebase_uid")
    price_id = data.get("price_id")  # Stripe后台创建的价格ID
    
    try:
        customer_id = get_or_create_stripe_customer(firebase_uid)
        
        subscription = stripe.Subscription.create(
            customer=customer_id,
            items=[{"price": price_id}],
            payment_behavior="default_incomplete",
            expand=["latest_invoice.payment_intent"],
            metadata={"firebase_uid": firebase_uid}
        )
        
        return jsonify({
            "subscriptionId": subscription.id,
            "clientSecret": subscription["latest_invoice"]["payment_intent"]["client_secret"]
        })
    except Exception as e:
        return jsonify(error=str(e)), 403

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