TelegramWalletPay

1
import asyncio
2
import ssl
3
import warnings
4
from contextlib import asynccontextmanager
5
from decimal import Decimal
6
from http import HTTPStatus
7
from typing import (
8
    Any,
9
    AsyncIterator,
10
    Dict,
11
    Literal,
12
    Mapping,
13
    Optional,
14
    Type,
15
    TypeVar,
16
    Union,
17
)
18

19
import certifi
20
from aiohttp import ClientResponse, ClientSession, TCPConnector
21
from pydantic import BaseModel
22

23
from telegram_wallet_pay.enums import Currency
24
from telegram_wallet_pay.errors import (
25
    InvalidAPIKeyError,
26
    InvalidRequestError,
27
    NotFountError,
28
    RequestLimitReachedError,
29
    TelegramWalletPayError,
30
    UnexpectedError,
31
)
32
from telegram_wallet_pay.schemas import (
33
    CreateOrderRequest,
34
    CreateOrderResponse,
35
    GetOrderPreviewResponse,
36
    GetOrderReconciliationListResponse,
37
    MoneyAmount,
38
    OrderAmountResponse,
39
)
40

41
T = TypeVar("T", bound=BaseModel)
42

43
AUTH_HEADER = "Wpay-Store-Api-Key"
44
DEFAULT_API_HOST = "https://pay.wallet.tg"
45

46
EXCEPTIONS_MAPPING: Dict[Union[HTTPStatus, int], Type[TelegramWalletPayError]] = {
47
    HTTPStatus.BAD_REQUEST: InvalidRequestError,
48
    HTTPStatus.UNAUTHORIZED: InvalidAPIKeyError,
49
    HTTPStatus.NOT_FOUND: NotFountError,
50
    HTTPStatus.TOO_MANY_REQUESTS: RequestLimitReachedError,
51
    HTTPStatus.INTERNAL_SERVER_ERROR: UnexpectedError,
52
}
53

54

55
class TelegramWalletPay:
56
    """Telegram Wallet API client."""
57

58
    def __init__(self, token: str, api_host: str = DEFAULT_API_HOST) -> None:
59
        if not token or not isinstance(token, str):
60
            msg = f"String token should be provided. You passed: {token}"
61
            raise RuntimeError(msg)
62

63
        self._base_url = api_host
64
        self._session: Optional[ClientSession] = None
65
        self._headers = {AUTH_HEADER: token}
66

67
    async def create_order(  # noqa: PLR0913
68
        self,
69
        *,
70
        amount: Union[str, Decimal, float],
71
        currency_code: Literal[
72
            Currency.TON,
73
            Currency.NOT,
74
            Currency.BTC,
75
            Currency.USDT,
76
            Currency.EUR,
77
            Currency.USD,
78
            Currency.RUB,
79
        ],
80
        description: str,
81
        external_id: str,
82
        timeout_seconds: int,
83
        customer_telegram_user_id: int,
84
        auto_conversion_currency: Optional[
85
            Literal[
86
                Currency.TON,
87
                Currency.NOT,
88
                Currency.BTC,
89
                Currency.USDT,
90
            ]
91
        ] = None,
92
        return_url: Optional[str] = None,
93
        fail_return_url: Optional[str] = None,
94
        custom_data: Optional[str] = None,
95
    ) -> CreateOrderResponse:
96
        """Create an order.
97

98
        Docs:
99
        https://docs.wallet.tg/pay/#tag/Order/operation/create
100
        """
101
        create_order_request = CreateOrderRequest(
102
            amount=MoneyAmount(
103
                amount=str(amount),
104
                currency_code=currency_code,
105
            ),
106
            auto_conversion_currency=auto_conversion_currency,
107
            description=description,
108
            return_url=return_url,
109
            fail_return_url=fail_return_url,
110
            custom_data=custom_data,
111
            external_id=external_id,
112
            timeout_seconds=timeout_seconds,
113
            customer_telegram_user_id=customer_telegram_user_id,
114
        )
115

116
        async with self._make_request(
117
            method="POST",
118
            url="/wpay/store-api/v1/order",
119
            json=create_order_request.model_dump(by_alias=True),
120
        ) as response:
121
            return await self._prepare_result(response, CreateOrderResponse)
122

123
    async def get_preview(self, order_id: str) -> GetOrderPreviewResponse:
124
        """Retrieve the order information.
125

126
        Deprecated! Use method `.get_order_preview()` instead.
127
        """
128
        warnings.warn(
129
            "Method `.get_preview()` is deprecated and will be removed in v1.0.0\n"
130
            "Use method `.get_order_preview()` instead.",
131
            category=DeprecationWarning,
132
            stacklevel=2,
133
        )
134
        return await self.get_order_preview(order_id)
135

136
    async def get_order_preview(self, order_id: str) -> GetOrderPreviewResponse:
137
        """Retrieve the order information.
138

139
        Docs:
140
        https://docs.wallet.tg/pay/#tag/Order/operation/getPreview
141
        """
142
        async with self._make_request(
143
            method="GET",
144
            url="/wpay/store-api/v1/order/preview",
145
            params={"id": order_id},
146
        ) as response:
147
            return await self._prepare_result(response, GetOrderPreviewResponse)
148

149
    async def get_order_list(
150
        self,
151
        *,
152
        offset: int,
153
        count: int,
154
    ) -> GetOrderReconciliationListResponse:
155
        """Get list of store orders.
156

157
        Items sorted by creation time in ascending order.
158

159
        Docs:
160
        https://docs.wallet.tg/pay/#tag/Order-Reconciliation/operation/getOrderList
161
        """
162
        query_params: Dict[str, Any] = {
163
            "offset": offset,
164
            "count": count,
165
        }
166

167
        async with self._make_request(
168
            method="GET",
169
            url="/wpay/store-api/v1/reconciliation/order-list",
170
            params=query_params,
171
        ) as response:
172
            return await self._prepare_result(
173
                response,
174
                GetOrderReconciliationListResponse,
175
            )
176

177
    async def get_order_amount(self) -> OrderAmountResponse:
178
        """Get total count of all created orders in the Store.
179

180
        Including all - paid and unpaid.
181

182
        Docs:
183
        https://docs.wallet.tg/pay/#tag/Order-Reconciliation/operation/getOrderAmount
184
        """
185
        async with self._make_request(
186
            method="GET",
187
            url="/wpay/store-api/v1/reconciliation/order-amount",
188
        ) as response:
189
            return await self._prepare_result(response, OrderAmountResponse)
190

191
    async def close(self) -> None:
192
        """Graceful session close."""
193
        if not self._session:
194
            return
195

196
        await self._session.close()
197

198
        # Wait 250 ms for the underlying SSL connections to close
199
        # https://docs.aiohttp.org/en/stable/client_advanced.html#graceful-shutdown
200
        await asyncio.sleep(0.25)
201

202
    async def _get_session(self) -> ClientSession:
203
        """Get aiohttp session with cache."""
204
        if self._session is None or self._session.closed:
205
            ssl_context = ssl.create_default_context(cafile=certifi.where())
206
            connector = TCPConnector(ssl=ssl_context)
207
            self._session = ClientSession(
208
                base_url=self._base_url,
209
                connector=connector,
210
                headers=self._headers,
211
            )
212

213
        return self._session
214

215
    @asynccontextmanager
216
    async def _make_request(
217
        self,
218
        method: str,
219
        url: str,
220
        params: Optional[Mapping[str, str]] = None,
221
        json: Optional[Mapping[str, str]] = None,
222
    ) -> AsyncIterator[ClientResponse]:
223
        """Make request with cached session."""
224
        session = await self._get_session()
225
        async with session.request(
226
            method=method,
227
            url=url,
228
            params=params,
229
            json=json,
230
        ) as response:
231
            yield response
232

233
    @staticmethod
234
    async def _prepare_result(response: ClientResponse, schema: Type[T]) -> T:
235
        """Prepare response result or raise an exception."""
236
        status = response.status
237
        body = await response.text()
238

239
        if status == HTTPStatus.OK:
240
            return schema.model_validate_json(body)
241

242
        exc_type = EXCEPTIONS_MAPPING.get(status, TelegramWalletPayError)
243
        raise exc_type(body)
244