Python - Dodo Payments Documentation

Python SDK を使用すると、Python アプリケーションから型付きで Dodo Payments REST API にアクセスできます。同期クライアント DodoPayments と非同期クライアント AsyncDodoPayments があり、どちらも httpx 上に構築されています。ネストされたリクエストパラメータは型付き辞書で、レスポンスは Pydantic モデルです。

インストール

SDK を pip でインストールします。

pip install dodopayments

非同期クライアントの HTTP バックエンドとして aiohttp を使用するには、aiohttp extra をインストールします。

pip install "dodopayments[aiohttp]"

client.webhooks.unwrap() で webhook の署名を検証するには、webhooks extra もインストールします。pip install "dodopayments[webhooks]"。

SDK には Python 3.9 以降が必要です。セキュリティ更新を受け取るため、最新の安定版 Python を使用してください。

クイックスタート

同期クライアント

クライアントを作成し、checkout session を作成します。

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 と同じメソッドがあります。各呼び出しを await してください。

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 key は環境変数または secrets manager に保存してください。バージョン管理にコミットしないでください。

主な機能

Pythonic Interface

パラメータにはキーワード引数、ネストされたオブジェクトには TypedDict 型、レスポンスには Pydantic モデルを使用します。

Async/Await

asyncio 用の AsyncDodoPayments。aiohttp をオプションの HTTP バックエンドとして使用できます。

Type Hints

すべてのメソッドに型ヒントがあり、エディターの自動補完と mypy による型チェックに対応します。

Auto-Pagination

List メソッドは、ループ中に次のページを取得する iterator を返します。

設定

環境変数

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 の検証に使用する webhook signing secret。
`DODO_PAYMENTS_BASE_URL`	`base_url`	`environment` URL を置き換えるカスタム API URL。
`DODO_PAYMENTS_LOG`	対応する引数なし	ログレベル。Logging を参照してください。

DODO_PAYMENTS_BASE_URL が設定されていて、environment も渡すと、コンストラクターは “Ambiguous URL” エラーを発生させます。その場合に environment を使用するには、base_url=None を渡してください。 webhook を検証するには、未加工のリクエスト本文とヘッダーを 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 以上のレスポンスを再試行します。デフォルトでは指数バックオフを使用して 2 回再試行します。それでもリクエストが失敗すると、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 Session の作成

checkout session を作成し、返された 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 時間後に期限切れになります。すべての session オプションについては、Checkout Sessions を参照してください。

顧客の管理

メールアドレスと名前を指定して顧客を作成し、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})")

サブスクリプションの処理

サブスクリプションを作成し、on-demand subscription に課金し、サブスクリプションの使用量履歴を読み取ります。

POST /subscriptions（SDK の subscriptions.create メソッド）は deprecated です。既存のインテグレーションでは引き続き機能しますが、新しいインテグレーションでは Checkout Session を通じてサブスクリプションを作成してください。

# 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 に必要なのは、2 文字の ISO 国コードである country だけです。customer では、既存の顧客を関連付ける {"customer_id": ...}、または顧客を作成する {"email": ..., "name": ...} を指定できます。charge は on-demand subscriptions 用で、product_price は通貨の最小単位で指定します。retrieve_usage_history はページ分割されたリストを返し、Pagination に示すように反復処理できます。

使用量ベースの課金

使用量イベントの取り込み

顧客の使用量イベントを送信します。

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 は idempotency key であるため、各イベントに一意の値を指定してください。同じ event_id が 1 つのリクエスト内に 2 回現れると、リクエスト全体が拒否されます。event_id がすでに取り込まれている場合、新しいイベントは無視されます。1 つのリクエストで最大 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 のフィルターも受け付けます。

ページネーション

自動ページネーション

List メソッドは、ループ中に次のページを取得する iterator を返します。

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())

手動ページネーション

一度に 1 ページずつ処理するには、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 クライアント設定

プロキシ、カスタムトランスポート、その他の 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"),
    ),
)

1 つのリクエストで別の HTTP クライアントを使用するには、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 エンドポイントから checkout session を作成し、その URL を返します。

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_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

すべてのエンドポイント、パラメータ、レスポンス。

Discord Community

質問したり、他の開発者と交流したりできます。

Report Issues

バグを報告したり、機能をリクエストしたりできます。

サポート

Python SDK についてサポートが必要な場合：

Discord: コミュニティサーバーに参加すると、リアルタイムでサポートを受けられます。
メール: support@dodopayments.com にお問い合わせください。
GitHub: リポジトリで issue を作成してください。

コントリビューション

コントリビューションするには、コントリビューションガイドラインをお読みください。

​インストール

​クイックスタート

​同期クライアント

​非同期クライアント

​主な機能

Pythonic Interface

Async/Await

Type Hints

Auto-Pagination

​設定

​環境変数

​タイムアウト

​再試行

​一般的な操作

​Checkout Session の作成

​顧客の管理

​サブスクリプションの処理

​使用量ベースの課金

​使用量イベントの取り込み

​イベントの一覧表示と取得

​ページネーション

​自動ページネーション

​非同期ページネーション

​手動ページネーション

​HTTP クライアント設定

​AIOHTTP を使用した非同期処理

​ロギング

​フレームワークとの統合

​FastAPI

​Django

​リソース

GitHub Repository

API Reference

Discord Community

Report Issues

​サポート

​コントリビューション

インストール

クイックスタート

同期クライアント

非同期クライアント

主な機能

設定

環境変数

タイムアウト

再試行

一般的な操作

Checkout Session の作成

顧客の管理

サブスクリプションの処理

使用量ベースの課金

使用量イベントの取り込み

イベントの一覧表示と取得

ページネーション

自動ページネーション

非同期ページネーション

手動ページネーション

HTTP クライアント設定

AIOHTTP を使用した非同期処理

ロギング

フレームワークとの統合

FastAPI

Django

リソース

サポート

コントリビューション