Bỏ qua để đến nội dung

SDK Python

~2 phút đọc

Gói dongtien gồm hai lớp. dongtien_api là client có kiểu, sinh bằng openapi-python-client từ đặc tả OpenAPI công khai: mỗi thao tác là một hàm sync hoặc asyncio. dongtien là lớp viết tay bọc bên ngoài, thêm xác thực, phiên bản, idempotency, thử lại, phân trang và lỗi có cấu trúc. Cần Python 3.11 trở lên.

Terminal window
pip install dongtien
import os
from dongtien import ApiError, DongTien
from dongtien_api.api.giao_dich import list_transactions
with DongTien(api_key=os.environ["DT_API_KEY"]) as dt: # dt_test_… hoặc dt_live_…
# Mọi phần tử của mọi trang; SDK tự xử lý cursor.
for tx in dt.paginate(list_transactions.sync_detailed, status="confirmed", limit=50):
print(tx.id, tx.amount)
try:
list_transactions.sync(client=dt.client, limit=500)
except ApiError as error:
print(error.code, error.request_id, error.errors)

Hàm của từng thao tác nằm trong module theo nhóm của đặc tả, ví dụ dongtien_api.api.giao_dich.list_transactions hay dongtien_api.api.tai_khoan_ngan_hang.list_bank_accounts. Tên nhóm là tên tiếng Việt không dấu của thẻ trong OpenAPI; bạn truyền client=dt.client vào hàm.

AsyncDongTien dùng các hàm asyncio và asyncio_detailed của client sinh, và async for với paginate.

import asyncio
import os
from dongtien import AsyncDongTien
from dongtien_api.api.giao_dich import list_transactions
async def main() -> None:
async with AsyncDongTien(api_key=os.environ["DT_API_KEY"]) as dt:
async for tx in dt.paginate(list_transactions.asyncio_detailed, limit=50):
print(tx.id)
asyncio.run(main())
Tuỳ chọnÝ nghĩa
api_keyBắt buộc. Khoá dt_live_… hoặc dt_test_…; chỉ gửi trong header Authorization.
base_urlMặc định https://api.dongtien.pro.
versionNgày phiên bản gửi trong DongTien-Version. Bỏ trống thì dùng phiên bản ghim ở khoá.
timeoutThời gian chờ mỗi yêu cầu, tính bằng giây; mặc định 30.
max_retriesSố lần thử lại sau lần đầu; mặc định 3.
retryChính sách thử lại chi tiết.
transportThay httpx transport, ví dụ trong kiểm thử.
headersHeader thêm vào mọi yêu cầu.

Với route khai báo idempotent, SDK tự gắn Idempotency-Key (UUIDv7) một lần cho mỗi lời gọi và giữ nguyên qua mọi lần thử lại. SDK thử lại tối đa 3 lần: 429 theo Retry-After, 5xx và lỗi mạng với yêu cầu đọc hoặc có khoá idempotency. Xem trang SDK và Idempotency.

ApiError có status, code, title, detail, type, request_id, errors (danh sách pointer, detail, code), retryable và retry_after. DongTienConnectionError báo yêu cầu không có phản hồi nào sau khi hết số lần thử lại. Cả hai cùng kế thừa DongTienError. Ý nghĩa từng trường ở trang Lỗi.

Gói dongtien xuất hàm verify_webhook (hoặc webhooks.verify) và ngoại lệ WebhookVerificationError để xác minh chữ ký Standard Webhooks:

import os
from dongtien import webhooks, WebhookVerificationError
def webhook_handler(raw_body: bytes, headers: dict[str, str]):
secret = os.environ["DTP_WEBHOOK_SECRET"] # whsec_...
try:
event = webhooks.verify(raw_body, headers, secret, tolerance=300)
print("Sự kiện hợp lệ:", event["type"])
return {"ok": True}, 200
except WebhookVerificationError as exc:
print("Chữ ký không hợp lệ:", exc)
return {"error": "invalid signature"}, 400