安装
使用 pip 安装 SDK:复制
pip install dodopayments
复制
pip install dodopayments[aiohttp]
SDK 需要 Python 3.7 或更高版本。我们建议使用最新的稳定版本的 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": "product_id",
"quantity": 1
}
],
)
print(checkout_session_response.session_id)
异步客户端
复制
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": "product_id",
"quantity": 1,
}
],
)
print(checkout_session_response.session_id)
asyncio.run(main())
请始终通过环境变量安全存储您的 API 密钥。切勿将其提交到版本控制中。
核心功能
Pythonic 接口
干净、符合 PEP 8 指南和 Python 约定的 Python 代码
Async/Await
完全支持使用 asyncio 的异步操作和可选的 aiohttp 集成
类型提示
完整的类型提示,以便更好的 IDE 支持和 mypy 类型检查
自动分页
对列表响应的自动分页,支持简单迭代
配置
环境变量
使用环境变量进行配置:.env
复制
DODO_PAYMENTS_API_KEY=your_api_key_here
超时
全局或每个请求配置请求超时:复制
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": "product_id",
"quantity": 0,
}
],
)
重试
配置自动重试行为:复制
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": "product_id",
"quantity": 0,
}
],
)
常见操作
创建结账会话
生成结账会话:复制
session = client.checkout_sessions.create(
product_cart=[
{
"product_id": "prod_123",
"quantity": 1
}
],
return_url="https://yourdomain.com/return"
)
print(f"Checkout URL: {session.url}")
管理客户
创建和检索客户信息:复制
# Create a customer
customer = client.customers.create(
email="[email protected]",
name="John Doe",
metadata={
"user_id": "12345"
}
)
# Retrieve customer
customer = client.customers.retrieve("cus_123")
print(f"Customer: {customer.name} ({customer.email})")
处理订阅
创建和管理定期订阅:复制
# Create a subscription
subscription = client.subscriptions.create(
customer_id="cus_123",
product_id="prod_456",
price_id="price_789"
)
# Retrieve usage history
usage_history = client.subscriptions.retrieve_usage_history(
subscription.id,
start_date="2024-01-01T00:00:00Z"
)
基于使用的计费
记录使用事件
跟踪基于使用的计费的自定义事件:复制
response = client.usage_events.ingest(
events=[
{
"event_id": "api_call_12345",
"customer_id": "cus_abc123",
"event_name": "api_request",
"timestamp": "2024-01-15T10:30:00Z",
"metadata": {
"endpoint": "/api/v1/users",
"method": "GET",
"tokens_used": "150"
}
}
]
)
列出和检索事件
复制
# 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",
limit=20
)
for event in events.data:
print(f"Event: {event.event_id} at {event.timestamp}")
分页
自动分页
自动迭代所有项目:复制
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)
异步分页
复制
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())
手动分页
以获得对分页的更多控制:复制
# 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 客户端配置
自定义底层httpx 客户端:
复制
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"),
),
)
使用 aiohttp 的异步
使用 aiohttp 提升异步性能:复制
import asyncio
from dodopayments import DefaultAioHttpClient
from dodopayments import AsyncDodoPayments
async def main() -> None:
async with AsyncDodoPayments(
bearer_token="My Bearer Token",
http_client=DefaultAioHttpClient(),
) as client:
checkout_session_response = await client.checkout_sessions.create(
product_cart=[
{
"product_id": "product_id",
"quantity": 0,
}
],
)
print(checkout_session_response.session_id)
asyncio.run(main())
日志记录
通过设置环境变量启用日志记录:复制
export DODO_PAYMENTS_LOG=info
复制
export DODO_PAYMENTS_LOG=debug
框架集成
FastAPI
复制
from fastapi import FastAPI, HTTPException
from dodopayments import AsyncDodoPayments
from pydantic import BaseModel
import os
app = FastAPI()
dodo = AsyncDodoPayments(bearer_token=os.getenv("DODO_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.url}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
Django
复制
from django.http import JsonResponse
from dodopayments import DodoPayments
import os
dodo = DodoPayments(bearer_token=os.getenv("DODO_API_KEY"))
def create_payment(request):
try:
payment = dodo.payments.create(
amount=5000,
currency="USD",
customer_id=request.user.customer_id
)
return JsonResponse({
"status": "success",
"payment_id": payment.id
})
except Exception as e:
return JsonResponse({
"status": "error",
"message": str(e)
}, status=400)
资源
支持
需要 Python SDK 的帮助吗?- Discord:加入我们的 社区服务器 获取实时支持
- 电子邮件:通过 [email protected] 联系我们
- GitHub:在 仓库 上打开一个问题