首页
/ 30分钟搞定支付集成:FastAPI全栈项目接入Stripe与PayPal实战指南

30分钟搞定支付集成:FastAPI全栈项目接入Stripe与PayPal实战指南

2026-02-05 05:17:37作者:裴麒琰

你是否还在为全栈项目的支付功能开发头疼?从支付网关选型到接口联调,从订单状态同步到异常处理,每个环节都可能踩坑。本文将以 full-stack-fastapi-postgresql 框架为基础,带你一步步实现 Stripe 与 PayPal 双支付渠道的无缝集成,解决真实业务场景中的支付流程痛点。

读完本文你将掌握:

  • 支付网关在 FastAPI 项目中的架构设计
  • Stripe 支付意图(Payment Intent)的完整实现
  • PayPal Smart Payment Buttons 前端集成方案
  • 订单状态同步与异步通知处理
  • 生产环境部署的安全最佳实践

项目基础架构概览

full-stack-fastapi-postgresql 提供了完善的全栈开发框架,基于 FastAPI 后端与现代前端组件。支付功能需在现有架构中新增以下模块:

backend/
├── app/api/routes/payments.py       # 支付相关API路由
├── app/core/payment_config.py       # 支付网关配置
├── app/services/stripe_service.py   # Stripe服务实现
├── app/services/paypal_service.py   # PayPal服务实现
└── app/models.py                    # 新增订单与支付记录模型

前端需在商品管理页面添加支付按钮,通过 API 与后端交互:

数据库模型设计

首先扩展数据模型以支持支付功能,需新增 OrderPayment 两个核心模型。修改 backend/app/models.py 文件,添加以下内容:

class Order(Base):
    __tablename__ = "orders"
    
    id = Column(Uuid, primary_key=True, default=uuid.uuid4)
    user_id = Column(Uuid, ForeignKey("users.id"))
    total_amount = Column(Numeric(10, 2), nullable=False)
    status = Column(Enum(OrderStatus), default=OrderStatus.PENDING)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
    
    # 关联字段
    user = relationship("User", back_populates="orders")
    payments = relationship("Payment", back_populates="order", cascade="all, delete-orphan")

class Payment(Base):
    __tablename__ = "payments"
    
    id = Column(Uuid, primary_key=True, default=uuid.uuid4)
    order_id = Column(Uuid, ForeignKey("orders.id"))
    provider = Column(Enum(PaymentProvider), nullable=False)  # STRIPE/PAYPAL
    provider_payment_id = Column(String, index=True)  # 支付平台交易ID
    amount = Column(Numeric(10, 2), nullable=False)
    status = Column(Enum(PaymentStatus), default=PaymentStatus.PENDING)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    
    # 关联字段
    order = relationship("Order", back_populates="payments")

支付网关配置

创建支付配置文件 backend/app/core/payment_config.py,统一管理支付网关参数:

from pydantic_settings import BaseSettings

class PaymentSettings(BaseSettings):
    STRIPE_API_KEY: str
    STRIPE_WEBHOOK_SECRET: str
    PAYPAL_CLIENT_ID: str
    PAYPAL_CLIENT_SECRET: str
    PAYPAL_WEBHOOK_ID: str
    PAYMENT_SUCCESS_URL: str = "http://localhost:3000/payment/success"
    PAYMENT_CANCEL_URL: str = "http://localhost:3000/payment/cancel"

payment_settings = PaymentSettings()

配置参数需添加到环境变量文件,参考项目现有配置格式:backend/app/core/config.py

Stripe支付集成实现

后端服务实现

创建 Stripe 服务文件 backend/app/services/stripe_service.py:

import stripe
from app.core.payment_config import payment_settings
from app.models import Order, Payment, PaymentProvider, PaymentStatus

stripe.api_key = payment_settings.STRIPE_API_KEY

class StripeService:
    @staticmethod
    def create_payment_intent(order: Order):
        """创建支付意图"""
        intent = stripe.PaymentIntent.create(
            amount=int(order.total_amount * 100),  # 金额以分为单位
            currency="usd",
            metadata={"order_id": str(order.id)},
            receipt_email=order.user.email
        )
        return intent
    
    @staticmethod
    def handle_webhook(payload: bytes, sig_header: str):
        """处理Stripe异步通知"""
        event = stripe.Webhook.construct_event(
            payload, sig_header, payment_settings.STRIPE_WEBHOOK_SECRET
        )
        
        if event.type == "payment_intent.succeeded":
            payment_intent = event.data.object
            return {
                "order_id": payment_intent.metadata.order_id,
                "provider_payment_id": payment_intent.id,
                "status": PaymentStatus.COMPLETED
            }
        return None

支付路由实现

新增支付路由文件 backend/app/api/routes/payments.py:

from fastapi import APIRouter, Depends, Request
from app.deps import CurrentUser, SessionDep
from app.services.stripe_service import StripeService
from app.models import Order

router = APIRouter(prefix="/payments", tags=["payments"])

@router.post("/stripe/create-intent/{order_id}")
def create_stripe_payment_intent(
    session: SessionDep, 
    current_user: CurrentUser, 
    order_id: uuid.UUID
):
    order = session.get(Order, order_id)
    # 验证订单归属权...
    intent = StripeService.create_payment_intent(order)
    return {"client_secret": intent.client_secret}

@router.post("/stripe/webhook")
async def stripe_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("Stripe-Signature")
    result = StripeService.handle_webhook(payload, sig_header)
    # 更新订单和支付记录状态...
    return {"status": "success"}

前端支付组件

修改商品详情页面,添加 Stripe 支付按钮 frontend/src/components/Items/AddItem.tsx

import { loadStripe } from '@stripe/stripe-js';
import { Elements, PaymentElement, useStripe, useElements } from '@stripe/react-stripe-js';
import { useEffect, useState } from 'react';
import { paymentsApi } from '../../client';

const stripePromise = loadStripe('pk_test_your_stripe_publishable_key');

function CheckoutForm({ orderId }) {
  const stripe = useStripe();
  const elements = useElements();
  const [clientSecret, setClientSecret] = useState('');

  useEffect(() => {
    // 创建PaymentIntent
    paymentsApi.createStripePaymentIntent(orderId)
      .then(data => setClientSecret(data.client_secret));
  }, [orderId]);

  const handleSubmit = async (e) => {
    e.preventDefault();
    const { error } = await stripe.confirmPayment({
      elements,
      confirmParams: { return_url: 'http://localhost:3000/payment/success' },
    });
    // 处理支付结果...
  };

  return (
    <form onSubmit={handleSubmit}>
      <PaymentElement />
      <button disabled={!stripe || !clientSecret}>支付</button>
    </form>
  );
}

export default function StripePayment({ orderId }) {
  return (
    <Elements stripe={stripePromise} options={{ clientSecret }}>
      <CheckoutForm orderId={orderId} />
    </Elements>
  );
}

PayPal支付集成实现

后端服务实现

创建 PayPal 服务文件 backend/app/services/paypal_service.py:

import paypalrestsdk
from app.core.payment_config import payment_settings

paypalrestsdk.configure({
  "mode": "sandbox",  # 生产环境改为 "live"
  "client_id": payment_settings.PAYPAL_CLIENT_ID,
  "client_secret": payment_settings.PAYPAL_CLIENT_SECRET
})

class PaypalService:
    @staticmethod
    def create_order(order):
        """创建PayPal订单"""
        paypal_order = paypalrestsdk.Order({
            "intent": "CAPTURE",
            "purchase_units": [{
                "amount": {
                    "currency": "USD",
                    "total": str(order.total_amount)
                },
                "custom_id": str(order.id)
            }],
            "application_context": {
                "return_url": payment_settings.PAYMENT_SUCCESS_URL,
                "cancel_url": payment_settings.PAYMENT_CANCEL_URL
            }
        })
        
        if paypal_order.create():
            return paypal_order
        raise Exception(f"PayPal订单创建失败: {paypal_order.error}")

支付流程整合

在支付路由中添加 PayPal 相关接口 backend/app/api/routes/payments.py:

@router.post("/paypal/create-order/{order_id}")
def create_paypal_order(
    session: SessionDep, 
    current_user: CurrentUser, 
    order_id: uuid.UUID
):
    order = session.get(Order, order_id)
    # 验证订单归属权...
    paypal_order = PaypalService.create_order(order)
    return {"order_id": paypal_order.id, "links": paypal_order.links}

@router.post("/paypal/capture-order/{paypal_order_id}")
def capture_paypal_order(
    session: SessionDep, 
    current_user: CurrentUser, 
    paypal_order_id: str
):
    order = paypalrestsdk.Order.find(paypal_order_id)
    if order.capture():
        # 处理支付成功逻辑
        return {"success": True, "order": order}
    raise Exception(f"支付捕获失败: {order.error}")

支付状态管理与前端展示

订单状态同步

支付完成后需同步订单状态,典型场景包括:

  1. 前端跳转同步通知
  2. 后端异步Webhook通知

实现订单状态更新逻辑,修改 backend/app/crud.py 添加:

def update_order_status(session, order_id: uuid.UUID, status: OrderStatus):
    order = session.get(Order, order_id)
    if not order:
        raise HTTPException(status_code=404, detail="Order not found")
    order.status = status
    session.add(order)
    session.commit()
    session.refresh(order)
    return order

前端订单页面

用户订单页面需展示支付状态与历史记录,修改 frontend/src/routes/_layout/items.tsx

import { useEffect, useState } from 'react';
import { ordersApi } from '../../client';

export default function UserOrders() {
  const [orders, setOrders] = useState([]);
  
  useEffect(() => {
    ordersApi.getUserOrders().then(data => setOrders(data));
  }, []);
  
  return (
    <div className="orders-container">
      <h2>我的订单</h2>
      <div className="orders-list">
        {orders.map(order => (
          <div key={order.id} className="order-card">
            <div className="order-header">
              <span>订单 #{order.id}</span>
              <span className={`status ${order.status}`}>{order.status}</span>
            </div>
            <div className="order-details">
              <p>金额: ${order.total_amount}</p>
              <p>创建时间: {new Date(order.created_at).toLocaleString()}</p>
            </div>
            {order.status === 'PENDING' && (
              <div className="payment-actions">
                <button onClick={() => startStripePayment(order.id)}>
                  Stripe支付
                </button>
                <button onClick={() => startPaypalPayment(order.id)}>
                  PayPal支付
                </button>
              </div>
            )}
          </div>
        ))}
      </div>
    </div>
  );
}

部署与安全最佳实践

环境配置与密钥管理

生产环境需使用环境变量管理敏感信息,参考项目现有配置方式:backend/app/core/config.py

Webhook安全验证

所有支付平台的Webhook必须验证签名,防止伪造请求:

  • Stripe: 使用签名头验证
  • PayPal: 使用证书验证与Webhook ID

测试与部署检查清单

部署前完成以下检查:

  • 使用支付平台测试环境完成端到端测试
  • 验证Webhook端点可公开访问
  • 测试异常场景(支付超时、取消、失败)
  • 配置日志记录支付流程关键节点

总结与扩展方向

本文实现了 full-stack-fastapi-postgresql 项目的双支付网关集成,涵盖从订单创建到支付完成的完整流程。实际业务中还可扩展以下功能:

  • 订阅支付与定期 billing
  • 优惠券与折扣系统
  • 退款处理流程
  • 多币种支持

项目官方文档:README.md 部署指南:deployment.md 开发文档:development.md

支付功能是电商类应用的核心模块,合理的架构设计能有效降低后续维护成本。建议在实际项目中根据业务复杂度选择合适的支付方案,中小规模项目可使用本文方案,大规模项目可考虑引入专业支付中台。

登录后查看全文
热门项目推荐
相关项目推荐