如何在FastAPI中通过PayPal IPN和Webhooks关联订阅与用户信息

一、通过PayPal官方字段传递用户标识 #

不要直接修改PayPal按钮添加自定义数据——PayPal会忽略非官方预留字段。正确的做法是在创建订阅的API请求中，使用PayPal支持的custom_id或invoice_id字段存储你的用户唯一ID（比如数据库里的user_id）。

示例代码（FastAPI中调用PayPal创建订阅）：

import requests
from fastapi import HTTPException

def create_paypal_subscription(user_id: str, paypal_access_token: str, plan_id: str):
    paypal_api_url = "https://api-m.paypal.com/v1/billing/subscriptions"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {paypal_access_token}"
    }
    payload = {
        "plan_id": plan_id,
        "custom_id": user_id,  # 存入你的用户唯一标识
        "application_context": {
            "return_url": "https://your-domain.com/subscribe-success",
            "cancel_url": "https://your-domain.com/subscribe-cancel"
        }
    }
    response = requests.post(paypal_api_url, json=payload, headers=headers)
    if response.status_code != 201:
        raise HTTPException(status_code=400, detail="创建PayPal订阅失败")
    return response.json()

二、接收PayPal通知并关联用户 #

无论是Webhooks还是IPN，PayPal都会在通知的payload中返回你之前传入的custom_id/invoice_id，以此关联到具体用户。

Webhooks处理示例（FastAPI端点）

from fastapi import FastAPI, Request
from paypalcheckoutsdk.webhooks import WebhookEventValidator
import os
from datetime import datetime

app = FastAPI()

# 从PayPal后台获取的Webhook ID和签名密钥
PAYPAL_WEBHOOK_ID = os.getenv("PAYPAL_WEBHOOK_ID")
PAYPAL_SIGNING_SECRET = os.getenv("PAYPAL_SIGNING_SECRET")

@app.post("/paypal-webhook")
async def handle_paypal_webhook(request: Request):
    # 1. 验证Webhook签名（必须做，防止伪造请求）
    request_body = await request.body()
    auth_header = request.headers.get("Paypal-Transmission-Sig")
    transmission_id = request.headers.get("Paypal-Transmission-Id")
    transmission_time = request.headers.get("Paypal-Transmission-Time")
    cert_url = request.headers.get("Paypal-Cert-Url")
    webhook_id = PAYPAL_WEBHOOK_ID

    validator = WebhookEventValidator()
    try:
        validator.validate(request_body, auth_header, transmission_id, transmission_time, cert_url, webhook_id)
    except Exception as e:
        return {"status": "error", "detail": "签名验证失败"}, 403

    # 2. 解析通知内容，提取用户标识
    payload = await request.json()
    event_type = payload["event_type"]
    
    # 处理订阅激活/续费成功等关键事件
    target_events = [
        "BILLING.SUBSCRIPTION.CREATED",
        "BILLING.SUBSCRIPTION.ACTIVATED",
        "BILLING.SUBSCRIPTION.PAYMENT.SUCCESS"
    ]
    if event_type in target_events:
        subscription_data = payload["resource"]
        user_id = subscription_data["custom_id"]  # 拿到关联的用户ID
        subscription_id = subscription_data["id"]
        subscription_status = subscription_data["status"]
        
        # 3. 更新数据库中的用户订阅信息
        update_user_subscription(user_id, subscription_id, subscription_status)
    
    return {"status": "success"}

# 数据库更新逻辑示例（以SQLAlchemy为例）
def update_user_subscription(user_id: str, subscription_id: str, status: str):
    from sqlalchemy.orm import Session
    from your_models import User  # 导入你的用户模型
    
    db: Session = get_db()  # 替换为你的数据库会话获取方式
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        user.subscription_id = subscription_id
        user.subscription_status = status
        user.subscription_updated_at = datetime.utcnow()
        db.commit()

IPN处理注意事项

如果用IPN，确保你的IPN端点是公网可访问的HTTPS地址，且在PayPal后台配置正确。IPN的POST payload中会包含custom字段（对应你创建订阅时传入的custom_id），提取后同样可以关联用户并更新数据库。

三、关键排查点 #

• 不要尝试给PayPal按钮添加自定义HTML字段，PayPal不会传递这些数据；必须使用custom_id/invoice_id官方字段。
• 测试时使用PayPal沙箱环境，避免产生真实交易。
• 务必验证Webhook/IPN的签名，防止恶意伪造通知修改用户订阅状态。

内容的提问来源于stack exchange，提问作者Yahya Al Hattali