initial

.gitignore

@@ -0,0 +1,4 @@
+__pycache__/
+.idea/
+venv/
+.env

README.md

@@ -0,0 +1,88 @@
+# aiomoney — простая асинхронная библиотека для работы с API ЮMoney
+
+### Авторизация приложения
+
+1. Зарегистрируйте новое приложение YooMoney по ссылке https://yoomoney.ru/myservices/new
+   (без указания чекбокса OAuth2!).
+2. Получите и скопируйте `client_id` после создания приложения
+3. Создайте запрос на получение api-токена.
+   [О правах приложения](https://yoomoney.ru/docs/wallet/using-api/authorization/protocol-rights)
+
+   ```python
+   import asyncio
+   from os import environ
+   from aiomoney import authorize_app
+
+
+   async def main():
+       await authorize_app(
+           client_id=environ.get("CLIENT_ID"),
+           redirect_uri=environ.get("REDIRECT_URI"),
+           app_permissions=[
+               "account-info",
+               "operation-history",
+               "operation-details",
+               "incoming-transfers",
+               "payment-p2p",
+               "payment-shop",
+           ]
+       )
+
+
+   if __name__ == "__main__":
+       asyncio.run(main())
+   ```
+
+4. Во время перенаправления по `redirect_uri` в адресной строке появится параметр `code=`.
+   Скопируйте значение и вставьте его в консоль
+5. Если авторизация прошла успешно, в консоли отобразится Ваш api-token.
+   Сохраните его в переменную окружения (рекомендация)
+
+### Получение основной информации об аккаунте
+
+```python
+import asyncio
+from aiomoney.types import AccountInfo, Operation, OperationDetails
+from aiomoney.wallet import YooMoneyWallet
+
+
+async def main():
+    wallet = YooMoneyWallet(access_token="ACCESS_TOKEN")
+
+    account_info: AccountInfo = await wallet.account_info
+    operation_history: list[Operation] = await wallet.get_operation_history()
+    operation_details: OperationDetails = await wallet.get_operation_details(operation_id="999")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+### Создание платёжной формы и проверка оплаты
+
+```python
+import asyncio
+from aiomoney.wallet import YooMoneyWallet, PaymentSource
+
+
+async def main():
+    wallet = YooMoneyWallet(access_token="ACCESS_TOKEN")
+
+    payment_form = await wallet.create_payment_form(
+        amount_rub=990,
+        unique_label="myproject_second_unicorn",
+        payment_source=PaymentSource.YOOMONEY_WALLET,
+        success_redirect_url="https://t.me/fofmow (nonono =/)"
+    )
+    # проверка платежа по label
+    payment_is_completed: bool = await wallet.check_payment_on_successful(payment_form.payment_label)
+
+    print(f"Ссылка на оплату:\n{payment_form.link_for_customer}\n\n"
+          f"Форма оплачена: {'Да' if payment_is_completed else 'Нет'}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+
+```
+

aiomoney/__init__.py

@@ -0,0 +1,2 @@
+from .authorization import authorize_app
+from .wallet import YooMoneyWallet

aiomoney/authorization.py

@@ -0,0 +1,32 @@
+from aiomoney.request import send_request
+
+AUTH_APP_URL = "https://yoomoney.ru/oauth/authorize?client_id={client_id}&response_type=code" \
+               "&redirect_uri={redirect_uri}&scope={permissions}"
+
+GET_TOKEN_URL = "https://yoomoney.ru/oauth/token?code={code}&client_id={client_id}&" \
+                "grant_type=authorization_code&redirect_uri={redirect_uri}"
+
+
+async def authorize_app(client_id, redirect_uri, app_permissions: list[str, ...]):
+    formatted_auth_app_url = AUTH_APP_URL.format(
+        client_id=client_id,
+        redirect_uri=redirect_uri,
+        permissions="%20".join(app_permissions)
+    )
+    response = await send_request(formatted_auth_app_url, response_without_data=True)
+
+    print(f"Перейдите по URL и подтвердите доступ для приложения\n{response.url}")
+    code = input("Введите код в консоль >  ").strip()
+
+    get_token_url = GET_TOKEN_URL.format(
+        code=code,
+        client_id=client_id,
+        redirect_uri=redirect_uri
+    )
+    _, data = await send_request(get_token_url)
+
+    access_token = data.get("access_token")
+    if not access_token:
+        return print(f"Не удалось получить токен. {data.get('error', '')}")
+
+    return print(f"Ваш токен — {access_token}. Сохраните его в безопасном месте!")

aiomoney/exceptions.py

@@ -0,0 +1,6 @@
+class UnresolvedRequestMethod(Exception):
+    ...
+
+
+class BadResponse(Exception):
+    ...

aiomoney/request.py

@@ -0,0 +1,47 @@
+from aiohttp import ClientSession, ClientResponse
+from aiohttp.client_exceptions import ContentTypeError
+
+from aiomoney.exceptions import UnresolvedRequestMethod, BadResponse
+
+ALLOWED_METHODS = ("post", "get")
+
+
+async def send_request(url: str,
+                       method: str = "post",
+                       response_without_data: bool = False,
+                       **kwargs) -> (ClientResponse, dict | None):
+    headers = {
+        "Content-Type": "application/x-www-form-urlencoded"
+    }
+    if add_headers := kwargs.pop("headers", {}):
+        headers |= add_headers
+
+    method = method.lower().strip()
+    await check_method(method)
+
+    async with ClientSession() as session:
+        async with getattr(session, method)(url, headers=headers, **kwargs) as response:
+            await post_handle_response(response)
+
+            if response_without_data:
+                return response
+
+            return response, await response.json()
+
+
+async def check_method(method: str):
+    if method not in ALLOWED_METHODS:
+        raise UnresolvedRequestMethod
+
+
+async def post_handle_response(response: ClientResponse):
+    try:
+        response_data = await response.json()
+        if isinstance(response_data, dict) and response_data.get("error"):
+            raise BadResponse(f"error — {response_data.get('error')}, response is {response}")
+
+    except ContentTypeError:
+        ...
+
+    if response.status >= 400:
+        raise BadResponse(response)

aiomoney/types/__init__.py

@@ -0,0 +1,4 @@
+from .account_info import AccountInfo
+from .operation_details import OperationDetails
+from .operation_history import Operation
+from .payment import PaymentForm, PaymentSource

aiomoney/types/account_info.py

@@ -0,0 +1,29 @@
+from pydantic import BaseModel, Field
+
+
+class BalanceDetails(BaseModel):
+    total: int
+    available: int
+    deposition_pending: int | None
+    blocked: int | None
+    debt: int | None
+    hold: int | None
+
+
+class LinkedCard(BaseModel):
+    pan_fragment: str
+    card_type: str = Field(None, alias="type")
+
+
+class AccountInfo(BaseModel):
+    """
+    Получение информации о состоянии счета пользователя
+    https://yoomoney.ru/docs/wallet/user-account/account-info
+    """
+    account: str  # номер счета
+    balance: int  # баланс счета
+    currency: str  # код валюты счета
+    account_status: str
+    account_type: str
+    balance_details: BalanceDetails | None
+    cards_linked: list[LinkedCard, ...] | None

aiomoney/types/operation_details.py

@@ -0,0 +1,29 @@
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class OperationDetails(BaseModel):
+    """
+    Детальная информация об операции из истории
+    https://yoomoney.ru/docs/wallet/user-account/operation-details
+    """
+    error: str | None
+    operation_id: str
+    status: str
+    pattern_id: str | None
+    direction: Literal["in"] | Literal["out"]
+    amount: int
+    amount_due: int | None
+    fee: int | None
+    operation_datetime: datetime = Field(alias="datetime")
+    title: str
+    sender: int | None
+    recipient: str | None
+    recipient_type: str | None
+    message: str | None
+    comment: str | None
+    label: str | None
+    details: str | None
+    operation_type: str = Field(alias="type")

aiomoney/types/operation_history.py

@@ -0,0 +1,20 @@
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class Operation(BaseModel):
+    """
+    Описание платежной операции
+    https://yoomoney.ru/docs/wallet/user-account/operation-history#response-operation
+    """
+    operation_id: str
+    status: str
+    execution_datetime: datetime = Field(alias="datetime")
+    title: str
+    pattern_id: str | None
+    direction: Literal["in"] | Literal["out"]
+    amount: int
+    label: str | None
+    operation_type: str = Field(alias="type")

aiomoney/types/payment.py

@@ -0,0 +1,13 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class PaymentSource:
+    BANK_CARD = "AC"
+    YOOMONEY_WALLET = "PC"
+
+
+@dataclass(frozen=True, slots=True)
+class PaymentForm:
+    link_for_customer: str
+    payment_label: str

aiomoney/wallet.py

@@ -0,0 +1,72 @@
+from aiomoney.request import send_request
+from aiomoney.types import AccountInfo, OperationDetails, Operation, PaymentSource, PaymentForm
+
+
+class YooMoneyWallet:
+    def __init__(self, access_token: str):
+        self.host = "https://yoomoney.ru"
+        self.__headers = dict(Authorization=f"Bearer {access_token}")
+
+    @property
+    async def account_info(self) -> AccountInfo:
+        url = self.host + "/api/account-info"
+        response, data = await send_request(
+            url, headers=self.__headers
+        )
+        return AccountInfo.parse_obj(data)
+
+    async def get_operation_details(self, operation_id: str) -> OperationDetails:
+        url = self.host + "/api/operation-details"
+        response, data = await send_request(
+            url, headers=self.__headers, data={"operation_id": operation_id}
+        )
+        return OperationDetails.parse_obj(data)
+
+    async def get_operation_history(self, label: str | None = None) -> list[Operation, ...]:
+        """
+        Получение последних 30 операций. На 10.03.2023 API yoomoney напросто игнорирует указанные
+        в документации параметры https://yoomoney.ru/docs/payment-buttons/using-api/forms?lang=ru#parameters
+        """
+        history_url = self.host + "/api/operation-history"
+        response, data = await send_request(
+            history_url, headers=self.__headers
+        )
+        if operations := data.get("operations"):
+            parsed = [Operation.parse_obj(operation) for operation in operations]
+            if label:
+                parsed = [operation for operation in parsed if operation.label == label]
+            return parsed
+
+    async def create_payment_form(self,
+                                  amount_rub: int,
+                                  unique_label: str,
+                                  success_redirect_url: str | None = None,
+                                  payment_source: PaymentSource = PaymentSource.BANK_CARD
+                                  ) -> PaymentForm:
+        account_info = await self.account_info
+        quickpay_url = "https://yoomoney.ru/quickpay/confirm.xml?"
+        params = {
+            "receiver": account_info.account,
+            "quickpay-form": "button",
+            "paymentType": payment_source,
+            "sum": amount_rub,
+            "successURL": success_redirect_url,
+            "label": unique_label
+        }
+        params = {k: v for k,v in params.items() if v}
+        response = await send_request(quickpay_url, response_without_data=True, params=params)
+
+        return PaymentForm(
+            link_for_customer=response.url,
+            payment_label=unique_label
+        )
+
+    async def check_payment_on_successful(self, label: str) -> bool:
+        need_operations = await self.get_operation_history(label=label)
+        return bool(need_operations) and need_operations.pop().status == "success"
+
+    async def revoke_token(self) -> None:
+        url = self.host + "/api/revoke"
+        response = await send_request(url=url, response_without_data=True, headers=self.__headers)
+        print(f"Запрос на отзыв токена завершен с кодом {response.status} "
+              f"https://yoomoney.ru/docs/wallet/using-api/authorization/revoke-access-token#response")

examples/authorization.py

@@ -0,0 +1,23 @@
+import asyncio
+from os import environ
+
+from aiomoney import authorize_app
+
+
+async def main():
+    await authorize_app(
+        client_id=environ.get("CLIENT_ID"),
+        redirect_uri=environ.get("REDIRECT_URI"),
+        app_permissions=[
+            "account-info",
+            "operation-history",
+            "operation-details",
+            "incoming-transfers",
+            "payment-p2p",
+            "payment-shop",
+        ]
+    )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())