Stripe

本指南将通过一个最小化的示例，演示如何使用 Stripe 的一次性支付链接和 webhook 来安全地核对支付。

首先，我们可以导入 stripe 库，并使用您可以在 Stripe 网页界面获得的 Stripe API 密钥 进行身份验证。

导出的源代码 
from fasthtml.common import *
import os

Stripe 身份验证

您可以直接从 PyPI 安装 stripe python sdk

pip install stripe

此外，您还需要安装 stripe cli。您可以在他们的文档这里找到如何在您的特定系统上安装它。

# uncomment and execute if needed
#!pip install stripe
导出的源代码 
import stripe
导出的源代码 
stripe.api_key = os.environ.get("STRIPE_SECRET_KEY")
DOMAIN_URL = os.environ.get("DOMAIN_URL", "https://:5001")

您可以通过访问此网址从 Stripe 控制面板获取此 API 密钥。

注意

注意：确保您已在控制面板中开启了测试模式。

确保在本教程中使用的是测试密钥。

assert 'test_' in stripe.api_key

应用程序前置设置

提示

“应用程序前置设置”部分的所有内容都只需运行一次，不应包含在您的 Web 应用程序中。

创建产品

您可以运行此代码以编程方式创建一个带有价格的 Stripe 产品。通常，这并不是您在 FastHTML 应用程序中动态执行的操作，而是一次性设置的内容。您也可以选择在 Stripe 控制面板界面上完成此操作。

导出的源代码 
def _search_app(app_nm:str, limit=1): 
    "Checks for product based on app_nm and returns the product if it exists"
    return stripe.Product.search(query=f"name:'{app_nm}' AND active:'True'", limit=limit).data

def create_price(app_nm:str, amt:int, currency="usd") -> list[stripe.Price]:
    "Create a product and bind it to a price object. If product already exist just return the price list."
    existing_product = _search_app(app_nm)
    if existing_product: 
        return stripe.Price.list(product=existing_product[0].id).data
    else:
        product = stripe.Product.create(name=f"{app_nm}")
        return [stripe.Price.create(product=product.id, unit_amount=amt, currency=currency)]

def archive_price(app_nm:str):
    "Archive a price - useful for cleanup if testing."
    existing_products = _search_app(app_nm, limit=50)
    for product in existing_products:
        for price in stripe.Price.list(product=product.id).data: 
            stripe.Price.modify(price.id, active=False)
        stripe.Product.modify(product.id, active=False)
提示

要进行定期付款，您可以在创建 stripe 价格时使用 recurring={"interval": "year"} 或 recurring={"interval": "month"}。

导出的源代码 
app_nm = "[FastHTML Docs] Demo Product"
price_list = create_price(app_nm, amt=1999)
assert len(price_list) == 1, 'For this tutorial, we only have one price bound to our product.'
price = price_list[0]
print(f"Price ID = {price.id}")
Price ID = price_1R1ZzcFrdmWPkpOp9M28ykjy

创建 webhook

webhook 只是一个 URL，您的应用程序通过它来监听来自 Stripe 的消息。它为 Stripe（支付处理器）提供了一种方式，在支付发生某些事件时通知您的应用程序。可以把它想象成一个送货通知：当客户完成支付时，Stripe 需要告知您的应用程序，以便您可以更新记录、发送确认邮件或提供对所购内容的访问权限。它只是一个 URL，

但您的应用程序需要确保每个 webhook 事件都确实来自 Stripe。也就是说，它需要对通知进行身份验证。为此，您的应用程序将需要一个webhook 签名密钥，用它来确认这些通知是由 Stripe 签名的。

这个密钥不同于您的 Stripe API 密钥。Stripe API 密钥让您向 Stripe 证明您的身份。而 webhook 签名密钥则让您确信来自 Stripe 的消息确实源自 Stripe。

无论您的应用程序是在本地以测试模式运行，还是在服务器上运行的真实生产应用，您都需要一个 webhook 签名密钥。以下是在这两种情况下获取 webhook 签名密钥的方法。

本地 Webhook

当您的应用程序在开发期间本地运行时，只能从您的计算机访问，因此 Stripe 无法向该 webhook 发出 HTTP 请求。为了在开发中解决这个问题，Stripe CLI 工具创建了一个安全隧道，将这些 webhook 通知从 Stripe 的服务器转发到您的本地应用程序。

运行此命令以启动该隧道。

stripe listen --forward-to https://:5001/webhook

成功后，该命令还会告诉您 webhook 签名密钥。获取它给您的密钥并将其设置为环境变量。

export STRIPE_LOCAL_TEST_WEBHOOK_SECRET=<your-secret>

生产环境 Webhook

对于已部署的应用程序，您需要在 Stripe 控制面板中配置一个永久的 webhook 连接。这会建立一个官方的通知渠道，Stripe 将通过该渠道将关于支付的实时更新发送到您应用程序的 /webhook URL。

在控制面板上，您可以配置哪些特定的支付事件通知将发送到此 webhook（例如，完成的结账、成功的支付、失败的支付等）。您的应用程序向 Stripe 库提供 webhook 签名密钥，以验证这些通知来自 Stripe 服务。这对于需要您的应用程序在没有人工干预的情况下自动响应支付活动的生产环境至关重要。

要配置永久的 webhook 连接，您需要执行以下步骤：

  1. 确保您像之前一样处于测试模式。

  2. 前往 https://dashboard.stripe.com/test/webhooks

  3. 点击“+ 添加端点”以创建一个新的 webhook（或者，如果没有此选项，请点击“创建事件目标”）。

  4. 在如下所示的主屏幕“监听 Stripe 事件”中，填写详细信息。您的端点 URL 将是 https://您的域名/webhook

  5. 保存您的 webhook 签名密钥。在“监听 Stripe 事件”屏幕上，您可以在右侧的应用程序示例代码中找到它，即“端点密钥”。您也可以稍后从控制面板中检索它。

您还需要配置哪些事件应生成 webhook 通知。

  1. 点击“+ 选择事件”以打开如下所示的辅助控制屏幕“选择要发送的事件”。在我们的案例中，我们希望监听 checkout.session.completed。

  2. 点击“添加事件”按钮，以确认要发送的事件。

提示

对于订阅，您可能还需要为您的 webhook 启用其他事件，例如：customer.subscription.created、customer.subscription.deleted 以及其他基于您用例的事件。

最后，点击“添加端点”以完成端点的配置。

应用程序

提示

此后的所有内容都将包含在您的实际应用程序中。本教程中创建的应用程序可以在这里找到。

设置以获取正确信息

为了接受付款，您需要知道是谁在付款。

有多种方法可以实现这一点，例如使用 oauth 或表单。在本例中，我们将首先将一个电子邮件地址硬编码到会话中，以模拟使用 oauth 时的情景。

我们将电子邮件地址保存到会话对象中，键为 auth。通过将此逻辑放入在每个请求处理之前运行的 beforeware 中，我们确保每个路由处理程序都能从会话对象中读取该地址。

导出的源代码 
def before(sess): sess['auth'] = '[email protected]'
bware = Beforeware(before, skip=['/webhook'])
app, rt = fast_app(before=bware)

我们将需要我们创建的 webhook 密钥。在本教程中，我们将使用上面创建的本地开发环境变量。对于您部署的生产环境，您需要从 Stripe 控制面板获取您的 webhook 密钥。

导出的源代码 
WEBHOOK_SECRET = os.getenv("STRIPE_LOCAL_TEST_WEBHOOK_SECRET")

支付设置

我们首先需要两样东西：

  1. 一个供用户点击支付的按钮
  2. 一个向 stripe 提供处理支付所需信息的路由
导出的源代码 
@rt("/")
def home(sess):
    auth = sess['auth']
    return Titled(
        "Buy Now", 
        Div(H2("Demo Product - $19.99"),
            P(f"Welcome, {auth}"),
            Button("Buy Now", hx_post="/create-checkout-session", hx_swap="none"),
            A("View Account", href="/account")))

我们只允许银行卡支付（payment_method_types=['card']）。有关其他选项，请参阅 Stripe 文档。

导出的源代码 
@rt("/create-checkout-session", methods=["POST"])
async def create_checkout_session(sess):
    checkout_session = stripe.checkout.Session.create(
        line_items=[{'price': price.id, 'quantity': 1}],
        mode='payment',
        payment_method_types=['card'],
        customer_email=sess['auth'],
        metadata={'app_name': app_nm, 
                  'AnyOther': 'Metadata',},
        # CHECKOUT_SESSION_ID is a special variable Stripe fills in for you
        success_url=DOMAIN_URL + '/success?checkout_sid={CHECKOUT_SESSION_ID}',
        cancel_url=DOMAIN_URL + '/cancel')
    return Redirect(checkout_session.url)
提示

对于订阅，模式通常是 subscription 而不是 payment。

本节创建了两个关键组件：一个带有“立即购买”按钮的简单网页，以及一个处理该按钮被点击时发生的事情的函数。

当客户点击“立即购买”时，应用程序会创建一个 Stripe 结账会话（实质上是一个支付页面），其中包含产品详情、价格和客户信息。然后，Stripe 接管支付过程，向客户显示一个安全的支付表单。支付完成或取消后，Stripe 会使用您指定的成功或取消 URL 将客户重定向回您的应用程序。这种方法将敏感的支付信息保留在您的服务器之外，因为实际的交易由 Stripe 处理。

支付后处理

在客户发起支付后，有两个并行的流程：

  1. 用户体验流程：客户被重定向到 Stripe 的结账页面，完成支付，然后被重定向回您的应用程序（成功或取消页面）。

  2. 后端处理流程：Stripe 向您的服务器发送关于支付事件的 webhook 通知，允许您的应用程序更新记录、提供访问权限或触发其他业务逻辑。

这种双轨方法确保了流畅的用户体验和可靠的支付处理。

webhook 通知至关重要，因为它是确认支付完成的可靠方式。

后端处理流程

创建一个包含您想要存储的信息的数据库模式。

导出的源代码 
# Database Table
class Payment:
    checkout_session_id: str  # Stripe checkout session ID (primary key)
    email: str
    amount: int  # Amount paid in cents
    payment_status: str  # paid, pending, failed
    created_at: int # Unix timestamp
    metadata: str  # Additional payment metadata as JSON

连接到数据库

导出的源代码 
db = Database("stripe_payments.db")
payments = db.create(Payment, pk='checkout_session_id', transform=True)

在我们的 webhook 中，我们可以执行任何我们需要的业务逻辑和数据库更新。

导出的源代码 
@rt("/webhook")
async def post(req):
    payload = await req.body()
    # Verify the event came from Stripe
    try:
        event = stripe.Webhook.construct_event(
            payload, req.headers.get("stripe-signature"), WEBHOOK_SECRET)
    except Exception as e:
        print(f"Webhook error: {e}")
        return
    if event and event.type == "checkout.session.completed":
        event_data = event.data.object
        if event_data.metadata.get('app_name') == app_nm:
            payment = Payment(
                checkout_session_id=event_data.id,
                email=event_data.customer_email,
                amount=event_data.amount_total,
                payment_status=event_data.payment_status,
                created_at=event_data.created,
                metadata=str(event_data.metadata))
            payments.insert(payment)
            print(f"Payment recorded for user: {event_data.customer_email}")
            
    # Do not worry about refunds yet, we will cover how to do this later in the tutorial
    elif event and event.type == "charge.refunded":
        event_data = event.data.object
        payment_intent_id = event_data.payment_intent
        sessions = stripe.checkout.Session.list(payment_intent=payment_intent_id)
        if sessions and sessions.data:
            checkout_sid = sessions.data[0].id
            payments.update(Payment(checkout_session_id= checkout_sid, payment_status="refunded"))
            print(f"Refund recorded for payment: {checkout_sid}")

webhook 路由是 Stripe 发送关于支付事件的自动通知的地方。当支付完成时，Stripe 会向此端点发送一个安全的通知。代码使用 webhook 密钥验证此通知的合法性，然后处理事件数据——提取诸如客户电子邮件和支付状态等信息。这使您的应用程序能够自动更新用户帐户、触发履行流程或记录交易详情，而无需人工干预。

请注意，在此路由中，我们的代码从 Stripe 事件中提取用户的电子邮件地址，而不是从会话对象中。这是因为此路由将由来自 Stripe 服务器的请求命中，而不是来自用户的浏览器。

提示

在处理订阅时，您通常会在一个 if 语句中添加额外的事件类型，以根据订阅状态适当地更新您的数据库。

if event.type == "payment_intent.succeeded":
    ...
elif event.type == "customer.subscription.created":
    ...
elif event.type == "customer.subscription.deleted":
    ...

用户体验流程

/success 路由是 Stripe 在支付成功完成后将用户重定向到的地方，这也会在 Stripe 调用 webhook 通知您的应用程序该交易之后发生。

Stripe 知道将用户发送到这里，因为您在创建结账会话时向 Stripe 提供了此路由。

但您需要验证情况是否如此。因此，在此路由中，您应该通过检查您的数据库中由您的应用程序在收到 webhook 通知时保存的条目，来验证用户的支付状态。

导出的源代码 
@rt("/success")
def success(sess, checkout_sid:str):    
    # Get payment record from database (saved in the webhook)
    payment = payments[checkout_sid]

    if not payment or payment.payment_status != 'paid': 
        return Titled("Error", P("Payment not found"))

    return Titled(
        "Success",
        Div(H2("Payment Successful!"),
            P(f"Thank you for your purchase, {sess['auth']}"),
            P(f"Amount Paid: ${payment.amount / 100:.2f}"),
            P(f"Status: {payment.payment_status}"),
            P(f"Transaction ID: {payment.checkout_session_id}"),
            A("Back to Home", href="/")))

还有一个 /cancel 路由，如果用户取消了结账，Stripe 会将用户重定向到那里。

导出的源代码 
@rt("/cancel")
def cancel():
    return Titled(
        "Cancelled",
        Div(H2("Payment Cancelled"),
            P("Your payment was cancelled."),
            A("Back to Home", href="/")))

这张图片显示了客户点击“立即购买”按钮后看到的 Stripe 支付页面。当您的应用重定向到 Stripe 结账 URL 时，Stripe 会显示这个安全的支付表单，客户在此输入他们的银行卡详细信息。出于测试目的，您可以使用 Stripe 的测试卡号（4242 4242 4242 4242），并配以任何未来的到期日期和任何 3 位数的 CVC 码。这张测试卡将在测试模式下成功处理支付，而不会收取真实费用。该表单显示了在您的 Stripe 会话中配置的产品名称和价格，提供了从您的应用到支付处理器，再到支付完成后返回的无缝过渡。

处理完支付后，您可以在 sqlite 数据库中看到存储在 webhook 中的每条记录。

接下来，我们可以看看如何添加退款路由。

为了使用退款功能，我们需要一个账户管理页面，用户可以在该页面上为他们的付款申请退款。

当您发起退款时，您可以在您的 Stripe 控制面板的 https://dashboard.stripe.com/payments 查看退款状态，如果您处于测试模式，则在 https://dashboard.stripe.com/test/payments 查看。

它会像这样带有一个已退款图标。

回顾

在本教程中，我们学习了如何实现和测试一个完整的 Stripe 支付流程，包括：

  1. 创建测试产品和价格
  2. 设置支付页面和结账会话
  3. 用于安全支付验证的 Webhook 处理
  4. 为用户体验构建成功/取消页面
  5. 添加退款功能
  6. 创建用于查看支付历史的账户管理页面

将此支付系统迁移到生产环境时，您需要在 Stripe 控制面板中创建实际的产品、价格和 webhook，而不是测试用的。您还需要将您的测试 API 密钥替换为生产环境的 Stripe API 密钥。