如何在 python appengine 应用中集成支付和购物车？

2026-10-24支付老李

结论

Python App Engine 应用可以接入支付，但不建议把 Google Wallet 当成在线商店的通用收款接口。Google Wallet 主要用于保存和展示支付凭证、票券等内容。它能否直接处理商品收款，还要看所在地区、商户资质以及当前开放的产品能力。

更常见且便于维护的做法是：

App Engine 管理商品、购物车和订单。
Stripe、PayPal 或当地合规的支付服务商负责收款。
后端创建支付会话，前端跳转到服务商托管的结账页。
支付服务商通过 Webhook 将最终结果通知后端。
购物车直接用 Python 数据模型实现，通常不需要专门的”购物车 API”。

下面用 Flask 和 Stripe Checkout 说明整体结构。代码仅用于展示集成思路。SDK 接口、支付方式和地区支持可能发生变化，接入前应查看服务商当前的官方文档。

在 App Engine 中创建支付会话

先把密钥保存为环境变量或 Secret Manager 中的机密，不要写进源代码：

# app.yaml
runtime: python312

env_variables:
  APP_BASE_URL: "https://your-project.example.com"

支付密钥更适合通过 Secret Manager 读取。以下示例使用环境变量，只是为了清楚展示支付流程：

import os

import stripe
from flask import Flask, jsonify, request

app = Flask(__name__)

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
APP_BASE_URL = os.environ["APP_BASE_URL"]

PRODUCTS = {
    "sku_keyboard": {
        "name": "Mechanical Keyboard",
        "unit_amount": 59900,
        "currency": "cny",
    },
    "sku_mouse": {
        "name": "Wireless Mouse",
        "unit_amount": 19900,
        "currency": "cny",
    },
}


@app.post("/api/checkout")
def create_checkout():
    body = request.get_json(silent=True) or {}
    cart_items = body.get("items", [])

    if not cart_items:
        return jsonify({"error": "购物车为空"}), 400

    line_items = []

    for item in cart_items:
        product_id = item.get("product_id")
        quantity = item.get("quantity")

        product = PRODUCTS.get(product_id)
        if product is None:
            return jsonify({"error": f"商品不存在：{product_id}"}), 400

        if not isinstance(quantity, int) or quantity < 1 or quantity > 20:
            return jsonify({"error": "商品数量无效"}), 400

        line_items.append({
            "price_data": {
                "currency": product["currency"],
                "product_data": {
                    "name": product["name"],
                },
                "unit_amount": product["unit_amount"],
            },
            "quantity": quantity,
        })

    # 实际项目应先把订单写入数据库，再将订单 ID 放入 metadata。
    order_id = "replace-with-database-order-id"

    session = stripe.checkout.Session.create(
        mode="payment",
        line_items=line_items,
        success_url=f"{APP_BASE_URL}/payment/success?session_id={{CHECKOUT_SESSION_ID}}",
        cancel_url=f"{APP_BASE_URL}/cart",
        metadata={"order_id": order_id},
    )

    return jsonify({"checkout_url": session.url})

浏览器收到 checkout_url 后，将用户跳转到托管结账页：

async function checkout(items) {
  const response = await fetch("/api/checkout", {
    method: "POST",
    headers: {
      "Content-Type": "application/json"
    },
    body: JSON.stringify({ items })
  });

  const result = await response.json();

  if (!response.ok) {
    throw new Error(result.error || "无法创建支付");
  }

  window.location.assign(result.checkout_url);
}

这里需要注意，前端只提交 product_id 和数量，不提交可被信任的价格。商品单价、币种、折扣和库存都必须由服务端重新计算，否则攻击者可以篡改请求中的金额。

使用 Webhook 确认付款

Webhook 端点必须使用原始请求体验证签名：

@app.post("/webhooks/stripe")
def stripe_webhook():
    payload = request.get_data()
    signature = request.headers.get("Stripe-Signature", "")
    webhook_secret = os.environ["STRIPE_WEBHOOK_SECRET"]

    try:
        event = stripe.Webhook.construct_event(
            payload,
            signature,
            webhook_secret,
        )
    except ValueError:
        return "Invalid payload", 400
    except stripe.error.SignatureVerificationError:
        return "Invalid signature", 400

    if event["type"] == "checkout.session.completed":
        session = event["data"]["object"]
        order_id = session.get("metadata", {}).get("order_id")

        if order_id:
            mark_order_as_paid_idempotently(
                order_id=order_id,
                provider_session_id=session["id"],
            )

    return "", 200


def mark_order_as_paid_idempotently(order_id, provider_session_id):
    """
    在实际项目中通过数据库事务完成以下操作：

    1. 查询订单。
    2. 如果订单已经是 paid，直接返回。
    3. 核对支付会话、币种和金额。
    4. 将订单更新为 paid。
    5. 保存 provider_session_id，防止重复处理。
    """
    pass

支付服务商可能重复发送同一个事件，所以 Webhook 处理必须具备幂等性。可以在数据库中为支付事件 ID 或支付会话 ID 设置唯一约束。遇到已经处理过的事件时，直接返回成功，避免重复扣减库存或发货。

对于支持延迟确认的支付方式，创建 Checkout Session 并不一定表示资金已经最终到账。服务端应根据具体支付方式处理成功、失败、取消和退款等事件。

购物车不需要依赖支付服务商

购物车和支付是两个独立模块：

购物车记录用户打算购买的商品。
订单保存结账时确定的商品、价格和收货信息。
支付记录第三方支付会话、交易状态和退款情况。

匿名用户的购物车可以暂存在浏览器 Cookie 或服务端 Session 中。登录用户的购物车通常保存在数据库里。Firestore、采用 Cloud Datastore 模式的 Firestore，或者项目现有的关系型数据库都可以使用。

简单的数据结构可以这样设计：

from dataclasses import dataclass


@dataclass
class CartItem:
    product_id: str
    quantity: int


@dataclass
class OrderLine:
    product_id: str
    product_name: str
    unit_amount: int
    quantity: int
    currency: str

订单行应保存结账时的商品名称和单价快照，不能只保存商品 ID。否则商品之后改价或改名，历史订单也会随之变化。

金额应使用最小货币单位的整数。例如，59900 表示 599.00 CNY。不要用二进制浮点数计算货币：

total_amount = sum(
    line.unit_amount * line.quantity
    for line in order_lines
)

如果业务包含税费、运费、优惠券或多币种，最好由服务端统一计算，并把明细写入订单。

选择支付服务商时要确认什么

只要支付 SDK 能在普通 Python Web 应用中运行，通常也能部署到现代 App Engine Python 运行时，因为支付调用本质上是 HTTPS 请求。不过，接入前仍需确认：

服务商是否支持商户所在的国家或地区。
是否支持所需的币种、银行卡和本地支付方式。
Python SDK 是否兼容项目使用的 Python 版本。
所用 App Engine 运行时是否允许发起外部 HTTPS 请求。
Webhook 是否能访问公开的 HTTPS 地址。
退款、争议处理、税务和对账能力是否符合业务需求。
是否需要额外的商户备案、实名验证或行业许可。

如果服务商没有合适的 Python SDK，也可以直接调用其 REST API，但认证、超时、重试和错误响应都需要自行处理。只要官方协议允许，编程语言本身通常不是限制。

上线前的检查

将密钥存入 Secret Manager，不要提交到 Git。
为结账接口增加登录校验、CSRF 防护和请求频率限制。
由服务端重新验证商品状态、库存、价格、币种和购买数量。
创建订单和扣减库存时使用事务或可靠的库存预留机制。
为支付 API 设置合理的连接超时和读取超时。
创建支付会话时使用幂等键，防止网络重试产生重复交易。
Webhook 必须验证签名并记录事件 ID。
不要在日志中输出卡号、密钥、完整地址等敏感数据。
在支付服务商的测试环境中覆盖成功、失败、取消、重复通知和退款流程。
不要自行接收或保存银行卡信息，优先使用托管结账页或服务商提供的安全组件。

更稳妥的方案并不是寻找一个同时处理购物车和支付的 Python API，而是由 App Engine 管理购物车与订单，让专业支付服务商处理敏感的收款流程。这样分工更清楚，安全、合规、退款和后续维护也更容易处理。