> ## Documentation Index
> Fetch the complete documentation index at: https://docs.dodopayments.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Python

> 使用 Dodo Payments Python SDK，通过同步和异步客户端，从 Python 3.9+ 创建结账会话、客户、订阅和用量事件。

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

## 安装

使用 pip 安装 SDK：

```bash theme={null}
pip install dodopayments
```

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

```bash theme={null}
pip install "dodopayments[aiohttp]"
```

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

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

## 快速开始

### 同步客户端

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

```python theme={null}
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` 具有相同的方法。请等待每次调用完成：

```python theme={null}
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())
```

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

## 核心功能

<CardGroup cols={2}>
  <Card title="Pythonic Interface" icon="code">
    参数使用关键字参数，嵌套对象使用 `TypedDict` 类型，响应使用 Pydantic 模型。
  </Card>

  <Card title="Async/Await" icon="bolt">
    用于 asyncio 的 `AsyncDodoPayments`，并将 aiohttp 作为可选 HTTP 后端。
  </Card>

  <Card title="Type Hints" icon="check">
    每个方法都提供类型提示，支持编辑器自动补全以及使用 mypy 进行类型检查。
  </Card>

  <Card title="Auto-Pagination" icon="arrows-rotate">
    列表方法返回迭代器，循环时会获取下一页。
  </Card>
</CardGroup>

## 配置

### 环境变量

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

```bash .env theme={null}
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` | 无对应参数 | 日志级别。请参阅[日志记录](#logging)。 |

如果已设置 `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](/developer-resources/webhooks)。

### 超时

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

```python theme={null}
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()` 对单个请求进行设置：

```python theme={null}
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` 的子类。

## 常见操作

本节示例使用[快速开始](#quick-start)中的 `client`。

### 创建结账会话

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

```python theme={null}
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 小时后过期。有关每个会话选项，请参阅[结账会话](/developer-resources/checkout-session)。

### 管理客户

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

```python theme={null}
# 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})")
```

### 处理订阅

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

<Warning>
  `POST /subscriptions`（SDK 的 `subscriptions.create` 方法）已**弃用**。它仍可用于现有集成，但新集成应通过[结账会话](/developer-resources/checkout-session)创建订阅。
</Warning>

```python expandable theme={null}
# 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",
)
```

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

## 基于用量的计费

### 接收用量事件

为客户发送用量事件：

```python theme={null}
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"
            }
        }
    ]
)
```

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

### 列出和获取事件

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

```python theme={null}
# 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` 筛选条件。

## 分页

### 自动分页

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

```python theme={null}
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` 进行循环：

```python theme={null}
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()` 会返回下一次请求所需的参数：

```python theme={null}
# 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 默认的连接限制、超时和重定向设置：

```python theme={null}
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` 传入：

```python theme={null}
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`：

```bash theme={null}
export DODO_PAYMENTS_LOG=info
```

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

```bash theme={null}
export DODO_PAYMENTS_LOG=debug
```

## 框架集成

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

### FastAPI

此 endpoint 使用异步客户端：

```python theme={null}
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 使用同步客户端：

```python theme={null}
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)
```

## 资源

<CardGroup cols={2}>
  <Card title="GitHub Repository" icon="github" href="https://github.com/dodopayments/dodopayments-python">
    源代码、版本发布信息和完整的方法列表。
  </Card>

  <Card title="API Reference" icon="book" href="/api-reference/introduction">
    每个 endpoint、参数和响应。
  </Card>

  <Card title="Discord Community" icon="discord" href="https://discord.gg/bYqAp4ayYh">
    提出问题并与其他开发者交流。
  </Card>

  <Card title="Report Issues" icon="bug" href="https://github.com/dodopayments/dodopayments-python/issues">
    报告 bug 或请求新功能。
  </Card>
</CardGroup>

## 支持

如需 Python SDK 方面的帮助：

* **Discord**：加入 [community server](https://discord.gg/bYqAp4ayYh) 获取实时帮助。
* **Email**：联系 [support@dodopayments.com](mailto:support@dodopayments.com)。
* **GitHub**：在 [repository](https://github.com/dodopayments/dodopayments-python) 中提交 issue。

## 贡献

如需贡献代码，请阅读[贡献指南](https://github.com/dodopayments/dodopayments-python/blob/main/CONTRIBUTING.md)。


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.