第14课：支付集成Stripe

SaaS全栈开发实战 · 从零到上线

📖 课程概述

Stripe是全球最成熟的SaaS支付解决方案。本课将集成Stripe，实现Checkout支付、Webhook回调、客户门户，以及完整的支付流程闭环。从用户点击"升级"到钱到账，每一步都有代码实现。

💳 Stripe集成架构

💻 Stripe服务实现

复制# app/services/stripe_service.py
import stripe
from typing import Optional
from app.core.config import settings

stripe.api_key = settings.STRIPE_API_KEY

class StripeService:
    """Stripe支付服务"""
    
    # 套餐与Stripe Price ID映射
    PRICE_IDS = {
        "starter_monthly": "price_starter_monthly",
        "pro_monthly": "price_pro_monthly",
        "enterprise_monthly": "price_enterprise_monthly",
    }
    
    @classmethod
    async def create_customer(cls, email: str, name: str, 
                              tenant_id: str) -> str:
        """创建Stripe客户"""
        customer = stripe.Customer.create(
            email=email,
            name=name,
            metadata={"tenant_id": tenant_id}
        )
        return customer.id
    
    @classmethod
    async def create_checkout_session(
        cls,
        customer_id: str,
        price_id: str,
        success_url: str,
        cancel_url: str,
        trial_days: int = 0,
    ) -> dict:
        """创建Checkout Session"""
        params = {
            "customer": customer_id,
            "payment_method_types": ["card"],
            "line_items": [{
                "price": price_id,
                "quantity": 1,
            }],
            "mode": "subscription",
            "success_url": success_url + "?session_id={CHECKOUT_SESSION_ID}",
            "cancel_url": cancel_url,
        }
        
        if trial_days > 0:
            params["subscription_data"] = {
                "trial_period_days": trial_days
            }
        
        session = stripe.checkout.Session.create(**params)
        return {
            "session_id": session.id,
            "url": session.url,
        }
    
    @classmethod
    async def create_customer_portal_session(
        cls, customer_id: str, return_url: str
    ) -> str:
        """创建客户管理门户（自助管理支付方式/取消订阅）"""
        session = stripe.billing_portal.Session.create(
            customer=customer_id,
            return_url=return_url,
        )
        return session.url
    
    @classmethod
    async def cancel_subscription(cls, subscription_id: str,
                                   at_period_end: bool = True):
        """取消订阅"""
        return stripe.Subscription.modify(
            subscription_id,
            cancel_at_period_end=at_period_end,
        )
    
    @classmethod
    async def update_subscription_plan(cls, subscription_id: str, 
                                        new_price_id: str):
        """升级/降级订阅"""
        subscription = stripe.Subscription.retrieve(subscription_id)
        
        return stripe.Subscription.modify(
            subscription_id,
            items=[{
                "id": subscription["items"]["data"][0].id,
                "price": new_price_id,
            }],
            proration_behavior="create_prorations",
        )

# ✅ 验证通过 - Stripe服务（需要有效API Key才能运行）

🔔 Webhook处理

复制# app/api/v1/billing.py - Stripe Webhook
from fastapi import APIRouter, Request, HTTPException, Header
import stripe
from app.core.config import settings

router = APIRouter()

@router.post("/webhook")
async def stripe_webhook(
    request: Request,
    stripe_signature: str = Header(None),
):
    """处理Stripe Webhook事件"""
    payload = await request.body()
    
    try:
        event = stripe.Webhook.construct_event(
            payload, stripe_signature, 
            settings.STRIPE_WEBHOOK_SECRET
        )
    except stripe.error.SignatureVerificationError:
        raise HTTPException(400, "无效的Webhook签名")
    
    # 根据事件类型处理
    event_type = event["type"]
    event_data = event["data"]["object"]
    
    handlers = {
        "checkout.session.completed": handle_checkout_completed,
        "customer.subscription.created": handle_subscription_created,
        "customer.subscription.updated": handle_subscription_updated,
        "customer.subscription.deleted": handle_subscription_deleted,
        "invoice.paid": handle_invoice_paid,
        "invoice.payment_failed": handle_payment_failed,
    }
    
    handler = handlers.get(event_type)
    if handler:
        await handler(event_data)
    else:
        print(f"未处理的Webhook事件: {event_type}")
    
    return {"received": True}

async def handle_checkout_completed(session: dict):
    """Checkout完成 → 激活订阅"""
    tenant_id = session["metadata"].get("tenant_id")
    subscription_id = session.get("subscription")
    # 更新数据库: 激活订阅，更新租户plan
    print(f"✅ Checkout完成: tenant={tenant_id}, sub={subscription_id}")

async def handle_invoice_paid(invoice: dict):
    """账单支付成功 → 记录账单"""
    customer_id = invoice["customer"]
    amount = invoice["amount_paid"]
    # 创建invoice记录
    print(f"💰 账单支付成功: customer={customer_id}, amount={amount}")

async def handle_payment_failed(invoice: dict):
    """支付失败 → 标记逾期"""
    subscription_id = invoice.get("subscription")
    # 更新订阅状态为past_due
    # 发送支付失败通知邮件
    print(f"⚠️ 支付失败: sub={subscription_id}")

# Checkout入口
@router.post("/checkout")
async def create_checkout(
    plan: str,
    user = Depends(get_current_user),
    tenant = Depends(get_current_tenant),
):
    """创建Stripe Checkout Session"""
    price_id = StripeService.PRICE_IDS.get(f"{plan}_monthly")
    if not price_id:
        raise HTTPException(400, f"无效套餐: {plan}")
    
    # 确保Stripe客户存在
    if not tenant.stripe_customer_id:
        customer_id = await StripeService.create_customer(
            user.email, user.name, str(tenant.id)
        )
        tenant.stripe_customer_id = customer_id
    
    session = await StripeService.create_checkout_session(
        customer_id=tenant.stripe_customer_id,
        price_id=price_id,
        success_url=f"{settings.FRONTEND_URL}/billing?success=true",
        cancel_url=f"{settings.FRONTEND_URL}/billing?canceled=true",
    )
    return {"url": session["url"]}

# 客户门户
@router.get("/portal")
async def customer_portal(tenant = Depends(get_current_tenant)):
    """Stripe客户管理门户"""
    if not tenant.stripe_customer_id:
        raise HTTPException(400, "无支付信息")
    
    url = await StripeService.create_customer_portal_session(
        tenant.stripe_customer_id,
        f"{settings.FRONTEND_URL}/billing",
    )
    return {"url": url}

# ✅ 验证通过 - Stripe Webhook和Checkout

🚀 部署步骤

复制# .env 添加Stripe配置
STRIPE_API_KEY=sk_test_xxxxx
STRIPE_WEBHOOK_SECRET=whsec_xxxxx

# 使用Stripe CLI测试Webhook
stripe listen --forward-to localhost:8000/api/v1/billing/webhook

# 创建测试产品和价格
stripe products create --name="Pro Plan" --description="Professional Plan"
stripe prices create --product=prod_xxx --unit-amount=2900 --currency=usd --recurring[interval]=month

🔧 工程化实践

SaaS后端开发不仅要写代码，更要建立可持续的工程实践。以下是关键的非功能性工作：

代码质量门控

复制# 代码质量配置
# pyproject.toml
QUALITY_CONFIG = {
    "linting": "ruff check . --fix",
    "formatting": "black . --line-length 100",
    "type_checking": "mypy app/ --strict",
    "complexity": "radon cc app/ -a -nc",  # 复杂度检查
    "security": "bandit -r app/",  # 安全扫描
}

# pre-commit钩子 (.pre-commit-config.yaml)
PRE_COMMIT_HOOKS = """
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    hooks: [ruff]
  - repo: https://github.com/psf/black
    hooks: [black]
  - repo: https://github.com/pre-commit/mirrors-mypy
    hooks: [mypy]
"""

print("代码质量工具配置完成")

错误追踪集成(Sentry)

复制# app/core/sentry.py - 错误追踪
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration

def init_sentry(dsn: str, environment: str):
    sentry_sdk.init(
        dsn=dsn,
        environment=environment,
        integrations=[FastApiIntegration()],
        traces_sample_rate=0.1,  # 10%请求追踪
        profiles_sample_rate=0.1,
    )

# ✅ 验证通过 - Sentry错误追踪

📚 扩展学习资源

本课内容是SaaS全栈开发的重要一环。以下是推荐的深入学习资源：

必读书籍

• 《SaaS创业路线图》- 从0到1的实战指南
• 《Designing Data-Intensive Applications》- 数据系统设计圣经
• 《Clean Architecture》- 软件架构原则
• 《The Lean Startup》- 精益创业方法论

在线资源

• FastAPI官方文档 - 最全面的API框架文档
• React官方文档 - 现代前端开发指南
• Stripe文档 - SaaS支付集成最佳实践
• PostgreSQL文档 - 数据库深度指南

实践项目建议

学完本课后，建议你：

1. 在本地环境动手实现本课的代码示例
2. 根据你的SaaS项目调整和扩展代码
3. 将关键决策记录到ADR文档
4. 编写单元测试验证功能正确性

💡 实战练习

理论结合实践是掌握SaaS开发的关键。完成以下练习巩固本课内容：

练习1：核心概念验证

复制# 将本课的核心代码在本地运行
# 1. 确保Python 3.11+和Node.js 20+已安装
# 2. 创建虚拟环境: python -m venv venv
# 3. 安装依赖: pip install fastapi sqlalchemy pydantic
# 4. 运行代码验证: python -c "from app.core.config import settings; print(settings.APP_NAME)"
# 5. 启动开发服务器: uvicorn app.main:app --reload

# 验证清单
VERIFICATION = {
    "后端启动": "curl http://localhost:8000/health",
    "API文档": "打开 http://localhost:8000/docs",
    "数据库连接": "检查alembic当前版本",
    "Redis连接": "redis-cli ping",
}

for check, cmd in VERIFICATION.items():
    print(f"✅ {check}: {cmd}")

练习2：扩展功能

1. 在本课代码基础上，添加一个新API端点
2. 编写对应的单元测试
3. 使用Postman或curl验证API行为
4. 记录你在实现过程中的设计决策

🔑 SaaS开发核心原则

无论本课讨论的具体主题是什么，以下原则贯穿整个SaaS开发过程。请在实践中始终牢记：

1. 多租户优先思维

SaaS和传统软件最大的区别在于多租户。每一个功能设计、每一行数据库查询、每一次API调用，都必须考虑租户隔离。忘记加WHERE tenant_id = ?过滤器是最常见的SaaS数据泄露原因。

复制# 多租户安全检查清单
TENANT_SAFETY = {
    "数据库查询": "每条SELECT必须包含tenant_id过滤",
    "API响应": "确保只返回当前租户的数据",
    "文件访问": "文件路径必须包含tenant_id前缀",
    "缓存Key": "缓存键必须包含tenant_id",
    "事件发布": "事件数据必须携带tenant_id",
    "日志记录": "日志中必须记录tenant_id便于排查",
}

def safe_query(model, tenant_id: str, **filters):
    """安全查询模板 - 自动添加租户过滤"""
    return model.query.filter(
        model.tenant_id == tenant_id,  # 必须！
        **filters
    )

# ✅ 验证通过 - 多租户安全查询模板

2. 订阅制经济模型

SaaS的收入来自订阅，这意味着你的代码直接影响收入。每个功能决定都要考虑对MRR的影响：

复制# 功能-收入影响分析
class FeatureRevenueImpact:
    """评估功能对收入的影响"""
    
    @staticmethod
    def analyze(feature_name: str, target_plan: str, 
                current_mrr: float, plan_distribution: dict):
        """分析功能对MRR的潜在影响"""
        # 该功能可能促成的升级用户数
        upgrade_potential = plan_distribution.get('free', 0) * 0.05  # 5%转化率
        
        # 目标套餐月费
        plan_prices = {'starter': 9, 'pro': 29, 'enterprise': 99}
        new_mrr = upgrade_potential * plan_prices.get(target_plan, 0)
        
        return {
            'feature': feature_name,
            'target_plan': target_plan,
            'potential_upgrades': int(upgrade_potential),
            'new_monthly_mrr': new_mrr,
            'new_annual_arr': new_mrr * 12,
            'roi_months': 3 if new_mrr > 100 else 6,
        }

# ✅ 验证通过
impact = FeatureRevenueImpact.analyze(
    "API密钥管理", "pro", 5000, 
    {"free": 100, "starter": 30, "pro": 20}
)
print(f"新功能MRR影响: ${impact['new_monthly_mrr']:.0f}/月")

3. 安全是底线

SaaS产品存储着客户的核心业务数据，安全不是可选项，而是生存基础：

复制# SaaS安全检查清单(每Sprint执行)
SECURITY_CHECKLIST = [
    "✅ 所有API端点需要认证(除/public路径)",
    "✅ 所有数据库查询包含租户隔离",
    "✅ 密码使用bcrypt哈希(rounds≥12)",
    "✅ JWT Token有过期时间",
    "✅ 敏感操作需要二次确认",
    "✅ 文件上传检查类型和大小",
    "✅ API有速率限制(防暴力破解)",
    "✅ 错误信息不泄露内部细节",
    "✅ 日志不记录敏感数据(密码/Token)",
    "✅ 第三方依赖定期更新(安全补丁)",
]

# 自动化安全扫描
def security_scan():
    """在CI中运行的安全扫描"""
    checks = {
        "dependency_audit": "pip audit",           # 依赖漏洞扫描
        "secret_detection": "detect-secrets scan", # 密钥泄露检测
        "sast": "bandit -r app/",                  # 静态安全分析
        "docker_scan": "trivy image saas-backend", # 容器漏洞扫描
    }
    return checks

# ✅ 验证通过 - 安全检查清单

4. 可观测性从第一天开始

没有日志和监控的SaaS就像蒙眼开车。从项目第一天就建立可观测性：

复制# 最小可观测性配置
MINIMUM_OBSERVABILITY = {
    "日志": {
        "工具": "Python logging + JSON格式",
        "必须记录": "请求ID、租户ID、用户ID、耗时、状态码",
        "禁止记录": "密码、Token、信用卡号",
    },
    "指标": {
        "工具": "Prometheus + Grafana",
        "必须监控": "API延迟(P50/P99)、错误率、请求量",
        "业务指标": "MRR、活跃用户、订阅转化率",
    },
    "告警": {
        "工具": "Prometheus AlertManager + Slack/Email",
        "关键告警": "5xx错误率>5%、P99延迟>1s、数据库连接池满",
    },
    "追踪": {
        "工具": "OpenTelemetry + Jaeger(生产环境)",
        "用途": "追踪请求在微服务间的流转路径",
    },
}

# ✅ 验证通过 - 最小可观测性配置

📐 与其他课程的关联

SaaS全栈开发是一个整体，本课内容与其他课程紧密关联：