SDK Python
Dùng gói dongtien để gọi API Dòng Tiền Pro bằng Python 3.11 trở lên: DongTien và AsyncDongTien, paginate, idempotency, thử lại và xử lý ApiError. dung goi dongtien de goi api dong tien pro bang python 3.11 tro len: dongtien va asyncdongtien, paginate, idempotency, thu lai va xu ly apierror.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.
pip install dongtienGọi API
Phần tiêu đề “Gọi API”import osfrom dongtien import ApiError, DongTienfrom 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.
Bất đồng bộ
Phần tiêu đề “Bất đồng bộ”AsyncDongTien dùng các hàm asyncio và asyncio_detailed của client sinh, và async for với paginate.
import asyncioimport osfrom dongtien import AsyncDongTienfrom 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
Phần tiêu đề “Tuỳ chọn”| Tuỳ chọn | Ý nghĩa |
|---|---|
api_key | Bắt buộc. Khoá dt_live_… hoặc dt_test_…; chỉ gửi trong header Authorization. |
base_url | Mặc định https://api.dongtien.pro. |
version | Ngày phiên bản gửi trong DongTien-Version. Bỏ trống thì dùng phiên bản ghim ở khoá. |
timeout | Thời gian chờ mỗi yêu cầu, tính bằng giây; mặc định 30. |
max_retries | Số lần thử lại sau lần đầu; mặc định 3. |
retry | Chính sách thử lại chi tiết. |
transport | Thay httpx transport, ví dụ trong kiểm thử. |
headers | Header thêm vào mọi yêu cầu. |
Idempotency và thử lại
Phần tiêu đề “Idempotency và thử lại”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.
Xác minh Webhook
Phần tiêu đề “Xác minh Webhook”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 osfrom 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