aiowata 0.1.0

Async Python SDK for the WATA payment system

aiowata

Async Python SDK for the WATA payment system.

Built on top of aiohttp, provides full coverage of the WATA H2H Payment API: payment links, transactions, refunds, and webhook verification.

Features

• Async/await — built on aiohttp, uses a single reusable session per client
• Payment links — create, retrieve, and search one-time or multi-use payment links
• Transactions — retrieve and search payment transactions
• Refunds — full or partial refund for paid transactions
• Webhook verification — RSA-SHA512 signature validation
• Sandbox support — switch between production and sandbox with a single flag
• Typed — full type annotations, py.typed marker for PEP 561

Installation

pip install aiowata

Quick Start

import asyncio
from aiowata import WataClient, Currency


async def main():
    async with WataClient(token="your-access-token") as client:
        link = await client.create_link(
            amount=1500.00,
            currency=Currency.RUB,
            order_id="order-001",
            description="Premium subscription",
        )
        print(f"Payment URL: {link.url}")


asyncio.run(main())

Usage

Payment Links

from aiowata import WataClient, Currency, LinkType, LinkStatus

async with WataClient(token="...") as client:
    # One-time link (default)
    link = await client.create_link(
        amount=1000.00,
        currency=Currency.RUB,
        order_id="order-123",
    )

    # Multi-use link with custom expiry
    link = await client.create_link(
        amount=500.00,
        currency=Currency.RUB,
        type=LinkType.MANY_TIME,
        expiration_date_time="2025-12-31T23:59:59Z",
    )

    # Retrieve link by ID
    link = await client.get_link("3fa85f64-5717-4562-b3fc-2c963f66afa6")

    # Search links
    result = await client.search_links(
        currencies=[Currency.RUB],
        statuses=[LinkStatus.OPENED],
        amount_from=100,
        max_result_count=50,
    )
    for item in result.items:
        print(item.id, item.amount, item.status)
    print(f"Total: {result.total_count}")

Transactions

from aiowata import WataClient, TransactionStatus

async with WataClient(token="...") as client:
    # Get transaction by ID
    tx = await client.get_transaction("3a16a4f0-27b0-09d1-16da-ba8d5c63eae3")
    print(tx.status, tx.amount)

    # Search paid transactions
    result = await client.search_transactions(
        statuses=[TransactionStatus.PAID],
        amount_from=500,
        max_result_count=100,
    )
    for tx in result.items:
        print(tx.id, tx.amount, tx.currency)

Refunds

async with WataClient(token="...") as client:
    refund = await client.create_refund(
        original_transaction_id="3a16a4f0-27b0-09d1-16da-ba8d5c63eae3",
        amount=500.00,
    )
    print(refund.transaction_id, refund.transaction_status)

The final refund status (Paid / Declined) is delivered asynchronously via webhook.

Webhook Verification

from aiowata import WataClient, verify_webhook_signature, parse_webhook

# Fetch the public key once and cache it
async with WataClient(token="...") as client:
    public_key = await client.get_public_key()


# Inside your webhook handler (e.g. aiohttp / FastAPI / Django)
async def handle_webhook(request):
    body: bytes = await request.read()
    signature: str = request.headers["X-Signature"]

    if not verify_webhook_signature(body, signature, public_key):
        return web.Response(status=400)

    payload = parse_webhook(body)
    print(payload.transaction_status, payload.amount, payload.order_id)

    # IMPORTANT: return 200 so WATA does not retry
    return web.Response(status=200)

Sandbox

Pass sandbox=True to use the WATA test environment:

client = WataClient(token="sandbox-token", sandbox=True)

Test cards (sandbox only):

Type	Number	Result
MIR 3DS	2200 0000 0000 0004	Success
MIR no 3DS	2200 0000 2222 2222	Success
VISA 3DS	4242 4242 4242 4242	Success
VISA no 3DS	4000 0000 0000 3055	Success
VISA 3DS	4012 8888 8888 1881	Declined
VISA no 3DS	4111 1111 1111 1111	Declined

Error Handling

All API errors inherit from WataError:

from aiowata import (
    WataError,
    WataValidationError,
    WataAuthError,
    WataRateLimitError,
)

try:
    link = await client.create_link(amount=-1, currency="RUB")
except WataValidationError as e:
    print(f"Validation: {e}")
    for err in e.validation_errors:
        print(f"  {err.members}: {err.message}")
except WataAuthError:
    print("Invalid or expired token")
except WataRateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except WataError as e:
    print(f"API error: {e}")

Exception hierarchy:

WataError
└── WataAPIError
    ├── WataValidationError   (400)
    ├── WataAuthError         (401)
    ├── WataForbiddenError    (403)
    ├── WataRateLimitError    (429)
    └── WataServerError       (5xx)

Custom Session

You can pass your own aiohttp.ClientSession for connection pooling or proxy configuration:

import aiohttp

async with aiohttp.ClientSession() as session:
    client = WataClient(token="...", session=session)
    link = await client.create_link(amount=100, currency="RUB")
    # session lifecycle is managed by you

License

MIT