Python SDK 提供了一个 Pythonic 接口来访问 Dodo Payments API,提供同步和异步客户端,并为请求和响应提供类型定义。它支持 Python 3.9+ 并包含全面的测试覆盖。
使用 pip 安装 SDK:
为了增强与 aiohttp 的异步性能:
pip install dodopayments[aiohttp]
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" : "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 类型检查
环境变量
使用环境变量进行配置:
DODO_PAYMENTS_API_KEY = your_api_key_here
全局或每个请求配置请求超时:
TODO: 代码占位符_9abf2b90689fdbeba7066523f30a85e5
配置自动重试行为:
TODO: 代码占位符_756b4fbd1bc978e1469b7e14e5b3224d
常见操作
创建结账会话
生成结账会话:
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 = "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 } )" )
处理订阅
创建和管理定期订阅:
# 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 提升异步性能:
TODO: 代码占位符_cadb6979db1a61bc98dafee77d3e1c4c
日志记录
通过设置环境变量启用日志记录:
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
TODO: 代码占位符_bda064b9709a59a0808f70b518b6ba48
需要 Python SDK 的帮助吗?
我们欢迎贡献!查看 贡献指南 开始。