Das Python SDK bietet eine pythonische Schnittstelle zur Dodo Payments API und stellt sowohl synchrone als auch asynchrone Clients mit Typdefinitionen für Anfragen und Antworten bereit. Es unterstützt Python 3.9+ und umfasst umfassende Testabdeckung.
Installation
Installieren Sie das SDK mit pip:
Für verbesserte asynchrone Leistung mit aiohttp:
pip install dodopayments[aiohttp]
Das SDK erfordert Python 3.9 oder höher. Wir empfehlen die Verwendung der aktuellsten stabilen Python-Version für das beste Erlebnis und Sicherheitsupdates.
Schnellstart
Synchroner Client
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)
Asynchroner Client
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())
Speichere deine API-Schlüssel stets sicher über Umgebungsvariablen. Committe sie niemals in ein Versionskontrollsystem.
Kernfunktionen
Pythonic Interface Sauberer, idiomatischer Python-Code, der den PEP-8-Richtlinien und Python-Konventionen folgt
Async/Await Volle Unterstützung für asynchrone Operationen mit asyncio und optionaler aiohttp-Integration
Type Hints Umfassende Type-Hints für bessere IDE-Unterstützung und Typüberprüfung mit mypy
Auto-Pagination Automatische Paginierung für Listenantworten mit einfacher Iteration
Konfiguration
Umgebungsvariablen
Konfigurieren Sie mit Umgebungsvariablen:
DODO_PAYMENTS_API_KEY = your_api_key_here
Zeitüberschreitungen
Konfigurieren Sie die Zeitüberschreitungen für Anfragen global oder pro Anfrage:
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" : 1 ,
}
],
)
Wiederholungen
Konfigurieren Sie das automatische Wiederholungsverhalten:
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" : 1 ,
}
],
)
Häufige Operationen
Erstellen einer Checkout-Sitzung
Generieren Sie eine Checkout-Sitzung:
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 } " )
Kunden verwalten
Erstellen und Abrufen von Kundeninformationen:
# 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 } )" )
Abonnements verwalten
Erstellen und Verwalten von wiederkehrenden Abonnements:
# 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"
)
Nutzungsbasierte Abrechnung
Nutzungsereignisse erfassen
Verfolgen Sie benutzerdefinierte Ereignisse für die nutzungsbasierte Abrechnung:
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"
}
}
]
)
Ereignisse auflisten und abrufen
# 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 } " )
Paginierung
Automatische Paginierung
Durchlaufen Sie alle Elemente automatisch:
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)
Asynchrone Paginierung
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())
Manuelle Paginierung
Für mehr Kontrolle über die Paginierung:
# 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-Client-Konfiguration
Passe den zugrunde liegenden httpx-Client an:
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" ),
),
)
Async mit aiohttp
Verwenden Sie aiohttp für verbesserte asynchrone Leistung:
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" : 1 ,
}
],
)
print (checkout_session_response.session_id)
asyncio.run(main())
Protokollierung
Aktivieren Sie die Protokollierung, indem Sie die Umgebungsvariable festlegen:
export DODO_PAYMENTS_LOG = info
Oder für Protokollierung auf Debug-Ebene:
export DODO_PAYMENTS_LOG = debug
Framework-Integration
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 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.url,
"session_id" : session.session_id
})
except Exception as e:
return JsonResponse({
"status" : "error" ,
"message" : str (e)
}, status = 400 )
Ressourcen
Unterstützung
Brauchen Sie Hilfe mit dem Python SDK?
Mitwirken
Wir freuen uns über Beiträge! Überprüfen Sie die Mitwirkungsrichtlinien , um loszulegen.