综合项目实战：电子商务API支付集成与Webhook

1. 介绍

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API，基于 Python 3.6+ 类型提示。在本教程中，我们将通过一个实际项目——电子商务API，学习如何集成支付功能和实现Webhook处理。这个项目将帮助您掌握FastAPI的核心概念，并应用到真实场景中。

本教程适合新手，即使您没有FastAPI经验，也可以跟随步骤一步步学习。

2. 环境设置

首先，确保您安装了Python 3.6+。然后，使用pip安装所需依赖。我们将创建一个新项目目录并设置虚拟环境。

安装依赖

打开终端，运行以下命令：

pip install fastapi uvicorn stripe python-dotenv

• fastapi: FastAPI框架
• uvicorn: ASGI服务器，用于运行FastAPI应用
• stripe: Stripe支付库，用于集成支付功能（您可以选择其他提供商，但本教程以Stripe为例）
• python-dotenv: 管理环境变量

项目结构

创建项目文件夹，例如 ecommerce_api，并添加以下文件：

ecommerce_api/
├── main.py
├── .env
├── requirements.txt
└── README.md

在 requirements.txt 中添加依赖：

fastapi
uvicorn
stripe
python-dotenv

3. 设计电子商务API

我们将设计一个简单的电子商务API，包含产品（Product）和订单（Order）模型。

定义数据模型

在 main.py 中，首先导入必要的模块并定义Pydantic模型：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime

app = FastAPI()

class Product(BaseModel):
    id: int
    name: str
    price: float
    description: Optional[str] = None

class Order(BaseModel):
    id: int
    product_id: int
    quantity: int
    total_amount: float
    status: str = "pending"  # e.g., pending, paid, shipped
    created_at: datetime = datetime.now()

创建CRUD端点

添加一些简单的路由来处理产品和订单：

# 模拟数据库（实际项目中应使用真实数据库如SQLite或PostgreSQL）
db_products = []
db_orders = []

@app.get("/products", response_model=List[Product])
async def get_products():
    return db_products

@app.post("/products")
async def create_product(product: Product):
    db_products.append(product)
    return {"message": "Product created"}

@app.post("/orders")
async def create_order(order: Order):
    db_orders.append(order)
    return {"message": "Order created", "order_id": order.id}

4. 支付集成（以Stripe为例）

Stripe是一个流行的支付处理平台。我们将集成Stripe来创建支付会话并处理支付结果。

配置Stripe API密钥

在 .env 文件中添加您的Stripe API密钥（请从Stripe Dashboard获取测试密钥）：

STRIPE_SECRET_KEY=sk_test_your_key_here
STRIPE_PUBLISHABLE_KEY=pk_test_your_key_here

在 main.py 中加载环境变量并配置Stripe：

import os
import stripe
from dotenv import load_dotenv

load_dotenv()  # 加载.env文件
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")

# 添加支付相关路由
@app.post("/create-payment-intent")
async def create_payment_intent(order_id: int):
    # 查找订单（模拟）
    order = next((o for o in db_orders if o.id == order_id), None)
    if not order:
        raise HTTPException(status_code=404, detail="Order not found")
    
    # 创建Stripe支付意图
    try:
        intent = stripe.PaymentIntent.create(
            amount=int(order.total_amount * 100),  # Stripe以美分为单位
            currency="usd",
            metadata={"order_id": order.id}
        )
        return {"client_secret": intent.client_secret}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

前端集成提示（可选）

前端可以使用Stripe.js或Checkout来处理支付。例如，在前端获取 client_secret 并重定向到Stripe支付页面。

5. Webhook实现

Webhook用于处理异步事件，例如支付成功后的通知。我们将设置一个Webhook端点来接收Stripe事件。

设置Webhook端点

在Stripe Dashboard中配置Webhook URL（例如，本地测试时使用 http://localhost:8000/webhook）。

在 main.py 中添加Webhook路由：

from fastapi import Request
import json

@app.post("/webhook")
async def stripe_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("Stripe-Signature")
    
    try:
        # 验证Webhook签名（确保事件来自Stripe）
        event = stripe.Webhook.construct_event(
            payload, sig_header, os.getenv("STRIPE_WEBHOOK_SECRET")  # 在Stripe Dashboard中获取
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail="Invalid payload")
    except stripe.error.SignatureVerificationError as e:
        raise HTTPException(status_code=400, detail="Invalid signature")
    
    # 处理事件
    if event['type'] == 'payment_intent.succeeded':
        payment_intent = event['data']['object']
        order_id = payment_intent['metadata'].get('order_id')
        # 更新订单状态为paid
        for order in db_orders:
            if order.id == order_id:
                order.status = "paid"
                break
        print(f"Order {order_id} paid successfully.")
    
    return {"status": "success"}

安全最佳实践

• 始终验证Webhook签名以防止伪造请求。
• 使用环境变量存储密钥，不要硬编码在代码中。

6. 安全和最佳实践

• 环境变量管理: 使用 python-dotenv 或类似工具管理敏感信息。
• 错误处理: 在API中添加适当的HTTP异常处理。
• 数据验证: 利用Pydantic的自动验证功能。
• 日志记录: 添加日志以调试Webhook事件。

7. 测试和部署

测试

启动开发服务器：

uvicorn main:app --reload

使用工具如Postman或cURL测试API端点。

• 测试 /products 和 /orders 端点。
• 模拟支付流程，触发Webhook事件（可以使用Stripe CLI进行本地测试）。

部署

项目可以部署到云平台如Heroku、AWS或Docker容器。确保设置环境变量（如Stripe密钥）在部署环境中。

8. 总结

通过本教程，您学习了如何：

1. 使用FastAPI构建电子商务API。
2. 集成Stripe支付功能。
3. 实现Webhook处理异步事件。
4. 应用安全和最佳实践。

这个项目为您提供了一个起点，可以扩展到更复杂的功能，如用户认证、库存管理等。继续探索FastAPI文档以深入学习。

下一步学习建议: 尝试集成其他支付提供商、添加数据库（如SQLAlchemy）、或实现更复杂的电子商务逻辑。

Happy coding!