Python - Dodo Payments Documentation

Python SDK 为 Python 应用提供对 Dodo Payments REST API 的类型化访问。它包含同步客户端 DodoPayments 和异步客户端 AsyncDodoPayments，二者都基于 httpx 构建。嵌套请求参数使用类型化字典，响应则使用 Pydantic 模型。

安装

使用 pip 安装 SDK：

pip install dodopayments

要将 aiohttp 用作异步客户端的 HTTP 后端，请安装 aiohttp extra：

pip install "dodopayments[aiohttp]"

要使用 client.webhooks.unwrap() 验证 webhook 签名，还需安装 webhooks extra：pip install "dodopayments[webhooks]"。

SDK 要求 Python 3.9 或更高版本。请使用最新的稳定版 Python，以获取安全更新。

快速开始

同步客户端

创建客户端，然后创建一个结账会话：

import os
from dodopayments import DodoPayments

client = DodoPayments(
    bearer_token=os.environ.get("DODO_PAYMENTS_API_KEY"),  # This is the default and can be omitted
    environment="test_mode",  # defaults to "live_mode"
)

checkout_session_response = client.checkout_sessions.create(
    product_cart=[
        {
            "product_id": "pdt_123",
            "quantity": 1
        }
    ],
)
print(checkout_session_response.session_id)

如果省略 bearer_token，客户端会读取 DODO_PAYMENTS_API_KEY 环境变量。如果省略 environment，客户端会连接到 live mode。test mode API key 仅适用于 environment="test_mode"。

异步客户端

AsyncDodoPayments 与 DodoPayments 具有相同的方法。请等待每次调用完成：

import os
import asyncio
from dodopayments import AsyncDodoPayments

client = AsyncDodoPayments(
    bearer_token=os.environ.get("DODO_PAYMENTS_API_KEY"),
    environment="test_mode",
)

async def main() -> None:
    checkout_session_response = await client.checkout_sessions.create(
        product_cart=[
            {
                "product_id": "pdt_123",
                "quantity": 1,
            }
        ],
    )
    print(checkout_session_response.session_id)

asyncio.run(main())

请将 API keys 保存在环境变量或 secrets manager 中。切勿将其提交到版本控制系统。

核心功能

Pythonic Interface

参数使用关键字参数，嵌套对象使用 TypedDict 类型，响应使用 Pydantic 模型。

Async/Await

用于 asyncio 的 AsyncDodoPayments，并将 aiohttp 作为可选 HTTP 后端。

Type Hints

每个方法都提供类型提示，支持编辑器自动补全以及使用 mypy 进行类型检查。

Auto-Pagination

列表方法返回迭代器，循环时会获取下一页。

配置

环境变量

将 API key 存储在环境变量中：

.env

DODO_PAYMENTS_API_KEY=your_api_key_here

如果未传入对应参数，客户端会读取以下变量：

变量	客户端参数	用途
`DODO_PAYMENTS_API_KEY`	`bearer_token`	API key。如果两者都未设置，客户端会引发错误。
`DODO_PAYMENTS_WEBHOOK_KEY`	`webhook_key`	webhook signing secret，用于验证 webhooks。
`DODO_PAYMENTS_BASE_URL`	`base_url`	自定义 API URL，用于替代 `environment` URL。
`DODO_PAYMENTS_LOG`	无对应参数	日志级别。请参阅日志记录。

如果已设置 DODO_PAYMENTS_BASE_URL，同时又传入 environment，构造函数会引发“Ambiguous URL”错误。在这种情况下，如需使用 environment，请传入 base_url=None。要验证 webhook，请将原始请求正文和 headers 传递给 client.webhooks.unwrap(payload, headers=headers)。它会使用 webhook key 检查签名，并返回解析后的事件。client.webhooks.unsafe_unwrap(payload) 会在不验证的情况下解析正文，因此只能用于测试。请参阅 Webhooks。

超时

请求默认在 1 分钟后超时，连接超时为 5 秒。请以秒为单位传入 timeout，或传入 httpx.Timeout，分别设置读取、写入和连接限制：

import httpx
from dodopayments import DodoPayments

# Configure default for all requests (default is 1 minute)
client = DodoPayments(
    timeout=20.0,  # 20 seconds
)

# More granular control
client = DodoPayments(
    timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)

# Override per-request
client.with_options(timeout=5.0).checkout_sessions.create(
    product_cart=[
        {
            "product_id": "pdt_123",
            "quantity": 1,
        }
    ],
)

请求超时时，SDK 会引发 APITimeoutError。超时的请求会自动重试，因此调用失败前可能需要经过比 timeout 更长的时间。

重试

在客户端上设置 max_retries，或使用 with_options() 对单个请求进行设置：

from dodopayments import DodoPayments

# Configure default for all requests (default is 2)
client = DodoPayments(
    max_retries=0,  # disable retries
)

# Override per-request
client.with_options(max_retries=5).checkout_sessions.create(
    product_cart=[
        {
            "product_id": "pdt_123",
            "quantity": 1,
        }
    ],
)

SDK 会重试连接错误，以及状态码为 408、409、429 或 500 及以上的响应。默认重试两次，并使用指数退避。当请求仍然失败时，SDK 会引发 dodopayments.APIError 的子类：

状态码	异常
400	`BadRequestError`
401	`AuthenticationError`
403	`PermissionDeniedError`
404	`NotFoundError`
409	`ConflictError`
422	`UnprocessableEntityError`
429	`RateLimitError`
500 及以上	`InternalServerError`
无响应	`APIConnectionError`

状态异常继承自 dodopayments.APIStatusError，该类包含 status_code 和 response 属性。APITimeoutError 是 APIConnectionError 的子类。

常见操作

本节示例使用快速开始中的 client。

创建结账会话

创建结账会话，然后将客户重定向到返回的 checkout_url：

session = client.checkout_sessions.create(
    product_cart=[
        {
            "product_id": "pdt_123",
            "quantity": 1
        }
    ],
    return_url="https://yourdomain.com/return"
)

print(f"Checkout URL: {session.checkout_url}")

每个 checkout_url 只能使用一次，并会在 24 小时后过期。有关每个会话选项，请参阅结账会话。

管理客户

使用电子邮件地址和姓名创建客户，然后通过 ID 获取该客户：

# Create a customer
customer = client.customers.create(
    email="customer@example.com",
    name="John Doe",
    metadata={
        "user_id": "12345"
    }
)

# Retrieve customer
customer = client.customers.retrieve("cus_123")
print(f"Customer: {customer.name} ({customer.email})")

处理订阅

创建订阅、收取按需订阅的费用，并读取订阅的用量历史记录。

POST /subscriptions（SDK 的 subscriptions.create 方法）已弃用。它仍可用于现有集成，但新集成应通过结账会话创建订阅。

# Create a subscription
subscription = client.subscriptions.create(
    billing={
        "country": "US",
        "city": "San Francisco",
        "state": "CA",
        "street": "1 Market St",
        "zipcode": "94105",
    },
    customer={"customer_id": "cus_123"},  # or {"email": "...", "name": "..."} for a new customer
    product_id="pdt_456",
    quantity=1,
)

# Charge an on-demand subscription
# product_price is in the lowest currency denomination (e.g., 2500 = $25.00 USD)
charge_response = client.subscriptions.charge(
    subscription_id=subscription.subscription_id,
    product_price=2500,
)

# Retrieve usage history (for metered subscriptions)
usage_history = client.subscriptions.retrieve_usage_history(
    subscription_id=subscription.subscription_id,
    start_date="2024-01-01T00:00:00Z",
)

billing 只需要 country，即两个字母的 ISO 国家代码。customer 接受 {"customer_id": ...} 以关联现有客户，或接受 {"email": ..., "name": ...} 以创建客户。charge 用于按需订阅，product_price 使用货币的最小单位。retrieve_usage_history 返回分页列表，你可以按照分页中的示例进行迭代。

基于用量的计费

接收用量事件

为客户发送用量事件：

from datetime import datetime, timezone

response = client.usage_events.ingest(
    events=[
        {
            "event_id": "api_call_12345",
            "customer_id": "cus_abc123",
            "event_name": "api_request",
            "timestamp": datetime.now(timezone.utc),
            "metadata": {
                "endpoint": "/api/v1/users",
                "method": "GET",
                "tokens_used": "150"
            }
        }
    ]
)

event_id 是幂等键，因此请为每个事件提供唯一值。如果同一个 event_id 在一次请求中出现两次，整个请求都会被拒绝。如果某个 event_id 已被接收，新事件会被忽略。单个请求最多接受 1,000 个事件。timestamp 默认为当前时间；如果时间早于当前时间超过 1 小时，或晚于当前时间超过 5 分钟，则会被拒绝。

列出和获取事件

通过 event_id 获取单个事件，或按客户和事件名称筛选事件列表：

# Get a specific event
event = client.usage_events.retrieve("api_call_12345")

# List events with filtering
events = client.usage_events.list(
    customer_id="cus_abc123",
    event_name="api_request",
    page_size=20
)

for event in events.items:
    print(f"Event: {event.event_id} at {event.timestamp}")

usage_events.list 还接受 meter_id、start 和 end 筛选条件。

分页

自动分页

列表方法返回一个迭代器，循环时会获取下一页：

from dodopayments import DodoPayments

client = DodoPayments()
all_payments = []

# Automatically fetches more pages as needed
for payment in client.payments.list():
    all_payments.append(payment)
print(all_payments)

异步分页

使用异步客户端时，请使用 async for 进行循环：

import asyncio
from dodopayments import AsyncDodoPayments

client = AsyncDodoPayments()

async def main() -> None:
    all_payments = []
    # Iterate through items across all pages
    async for payment in client.payments.list():
        all_payments.append(payment)
    print(all_payments)

asyncio.run(main())

手动分页

如需一次处理一页，请读取 items，并调用 has_next_page() 和 get_next_page()。next_page_info() 会返回下一次请求所需的参数：

# Access items from current page
first_page = client.payments.list()
for payment in first_page.items:
    print(payment.brand_id)

# Check for more pages
if first_page.has_next_page():
    next_page = first_page.get_next_page()
    print(f"Fetched {len(next_page.items)} more items")

HTTP 客户端配置

要添加 proxy、自定义 transport 或其他 httpx 设置，请传入自定义的 http_client。DefaultHttpxClient 会保留 SDK 默认的连接限制、超时和重定向设置：

import httpx
from dodopayments import DodoPayments, DefaultHttpxClient

client = DodoPayments(
    base_url="http://my.test.server.example.com:8083",
    http_client=DefaultHttpxClient(
        proxy="http://my.test.proxy.example.com",
        transport=httpx.HTTPTransport(local_address="0.0.0.0"),
    ),
)

要为单个请求使用不同的 HTTP client，请调用 client.with_options(http_client=...)。

使用 AIOHTTP 的异步客户端

默认情况下，异步客户端使用 httpx 发送请求。要获得更好的并发性能，请安装 aiohttp extra，并将 DefaultAioHttpClient() 作为 http_client 传入：

import os
import asyncio
from dodopayments import DefaultAioHttpClient
from dodopayments import AsyncDodoPayments

async def main() -> None:
    async with AsyncDodoPayments(
        bearer_token=os.environ.get("DODO_PAYMENTS_API_KEY"),
        http_client=DefaultAioHttpClient(),
    ) as client:
        checkout_session_response = await client.checkout_sessions.create(
            product_cart=[
                {
                    "product_id": "pdt_123",
                    "quantity": 1,
                }
            ],
        )
        print(checkout_session_response.session_id)

asyncio.run(main())

日志记录

SDK 使用标准库的 logging 模块记录日志。要启用日志记录，请将 DODO_PAYMENTS_LOG 设置为 info：

export DODO_PAYMENTS_LOG=info

如需更多详细信息，请将其设置为 debug：

export DODO_PAYMENTS_LOG=debug

框架集成

这些示例会从 web endpoint 创建结账会话，并返回其 URL。

FastAPI

此 endpoint 使用异步客户端：

from fastapi import FastAPI, HTTPException
from dodopayments import AsyncDodoPayments
from pydantic import BaseModel
import os

app = FastAPI()
dodo = AsyncDodoPayments(bearer_token=os.getenv("DODO_PAYMENTS_API_KEY"))

class CheckoutRequest(BaseModel):
    product_id: str
    quantity: int

@app.post("/create-checkout")
async def create_checkout(request: CheckoutRequest):
    try:
        session = await dodo.checkout_sessions.create(
            product_cart=[{
                "product_id": request.product_id,
                "quantity": request.quantity
            }],
            return_url="https://yourdomain.com/return"
        )
        return {"checkout_url": session.checkout_url}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

Django

此 view 使用同步客户端：

from django.http import JsonResponse
from django.views.decorators.http import require_POST
from django.views.decorators.csrf import csrf_exempt
from dodopayments import DodoPayments
import os
import json

client = DodoPayments(bearer_token=os.getenv("DODO_PAYMENTS_API_KEY"))

@csrf_exempt
@require_POST
def create_checkout(request):
    try:
        data = json.loads(request.body)
        session = client.checkout_sessions.create(
            product_cart=[{
                "product_id": data.get("product_id"),
                "quantity": data.get("quantity", 1)
            }],
            return_url="https://yourdomain.com/return"
        )
        return JsonResponse({
            "status": "success",
            "checkout_url": session.checkout_url,
            "session_id": session.session_id
        })
    except Exception as e:
        return JsonResponse({
            "status": "error",
            "message": str(e)
        }, status=400)

资源

GitHub Repository

源代码、版本发布信息和完整的方法列表。

API Reference

每个 endpoint、参数和响应。

Discord Community

提出问题并与其他开发者交流。

Report Issues

报告 bug 或请求新功能。

支持

如需 Python SDK 方面的帮助：

Discord：加入 community server 获取实时帮助。
Email：联系 support@dodopayments.com。
GitHub：在 repository 中提交 issue。

贡献

如需贡献代码，请阅读贡献指南。

​安装

​快速开始

​同步客户端

​异步客户端

​核心功能

Pythonic Interface

Async/Await

Type Hints

Auto-Pagination

​配置

​环境变量

​超时

​重试

​常见操作

​创建结账会话

​管理客户

​处理订阅

​基于用量的计费

​接收用量事件

​列出和获取事件

​分页

​自动分页

​异步分页

​手动分页

​HTTP 客户端配置

​使用 AIOHTTP 的异步客户端

​日志记录

​框架集成

​FastAPI

​Django

​资源

GitHub Repository

API Reference

Discord Community

Report Issues

​支持

​贡献

安装

快速开始

同步客户端

异步客户端

核心功能

配置

环境变量

超时

重试

常见操作

创建结账会话

管理客户

处理订阅

基于用量的计费

接收用量事件

列出和获取事件

分页

自动分页

异步分页

手动分页

HTTP 客户端配置

使用 AIOHTTP 的异步客户端

日志记录

框架集成

FastAPI

Django

资源

支持

贡献