aioredis стал частью redis-py

README.md

@@ -7,9 +7,9 @@
 
 
 
-## Асинхронная библиотека для работы с SberPay QR/Плати QR.
+## Асинхронная и синхронная библиотека для работы с SberPay QR/Плати QR.
 
-Асинхронная библиотека для работы с SberPay QR/Плати QR.
+Асинхронная и синхронная библиотека для работы с SberPay QR/Плати QR.
 
 Позволяет создавать динамический QR и проверять статус платежа.
 
@@ -58,6 +58,45 @@ if __name__ == '__main__':
     asyncio.run(creation_qr())
 ```
 
+## Пример (sync)
+
+```python
+import os
+from SberQR import SberQR
+
+member_id = '00000105'  # выдается через почту [email protected]
+tid = '24601234'  # ID  терминала/Точки. Получить в ЛК Сбрербанк бизнес на странице Информация о точке
+id_qr = '1000301234'  # Номер наклейки с QR-кодом. Получить в ЛК Сбрербанк бизнес Информация о точке/список оборудования
+client_id = '6e7254e2-6de8-4074-b458-b7238689772b'  # получить на api.developer.sber.ru
+client_secret = '3a0ea8cb-886c-4efa-ac45-e3d36aaba335'  # получить на api.developer.sber.ru
+
+#
+crt_from_pkcs12 = f'{os.getcwd()}/cert.crt'  # Для асинхронной версии требуется распаковать сертификат
+key_from_pkcs12 = f'{os.getcwd()}/private.key'  # Для асинхронной версии требуется распаковать приватный ключ
+pkcs12_password = 'SomeSecret'  # Пароль от файла сертификат. Получается на api.developer.sber.ru
+russian_crt = f'{os.getcwd()}/Cert_CA.pem'  # Сертификат мин.цифры для установления SSL соединения
+# Если требуется передайте аргумент redis=
+# redis = aioredis.from_url("redis://localhost", decode_responses=True)
+# redis = "redis://localhost"
+# Redis используется только для временного хранения токена
+sber_qr = SberQR(member_id=member_id, id_qr=tid, tid=tid,
+                      client_id=client_id, client_secret=client_secret,
+                      crt_file_path=crt_from_pkcs12, key_file_path=key_from_pkcs12,
+                      pkcs12_password=pkcs12_password,
+                      russian_crt=russian_crt)
+positions = [{"position_name": 'Товар ра 10 рублей',
+              "position_count": 1,
+              "position_sum": 1000,
+              "position_description": 'Какой-то товар за 10 рублей'}
+             ]
+def creation_qr():
+    data = sber_qr.creation(description=f'Оплата заказа 3', order_sum=1000, order_number="3", positions=positions)
+    print(data)
+
+if __name__ == '__main__':
+    creation_qr()
+```
+
 Для работы потребуется получить от банка следующие параметры
 
 ```python

SberQR/AsyncSberQR.py

@@ -8,10 +8,10 @@
 from typing import Optional, Type, Union, List, Dict
 
 import aiohttp
-import aioredis
+from redis import asyncio as aioredis
 import certifi
 import ujson as json
-from aioredis.client import Redis
+from redis.asyncio.client import Redis
 
 from .api import make_request, Methods
 from .payload import generate_payload
@@ -126,8 +126,9 @@ async def token(self, scope: Scope):
                        'rquid': ''.join(choices(hexdigits, k=32))}
             data = {'grant_type': 'client_credentials', 'scope': scope.value}
             token_data = await self.request(Methods.oauth, headers, data)
-            await self._redis.set(f'{self._client_id}token_{scope.value}', token_data['access_token'],
-                                  int(token_data['expires_in']) - 10)
+            if self._redis:
+                await self._redis.set(f'{self._client_id}token_{scope.value}', token_data['access_token'],
+                                      int(token_data['expires_in']) - 10)
             return token_data['access_token']
 
     async def creation(self, description: str, order_sum: int, order_number: str, positions: Union[List, Dict]):

SberQR/SberQr.py

@@ -0,0 +1,197 @@
+import asyncio
+import base64
+import ssl
+from datetime import datetime
+from logging import getLogger
+from random import choices
+from string import hexdigits
+from typing import Optional, Type, Union, List, Dict
+
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.ssl_ import create_urllib3_context
+
+import certifi
+import ujson as json
+from redis.client import Redis
+
+from .api_sync import make_request, Methods
+from .payload import generate_payload
+from .scope import Scope
+from .types import RegistryType, CancelType
+
+logger = getLogger(__name__)
+
+
+class SberQR:
+
+    def __init__(self, member_id: str, id_qr: str, tid: str,
+                 client_id: str, client_secret: str,
+                 crt_file_path: str, key_file_path: str,
+                 pkcs12_password: str,
+                 russian_crt: str,
+                 redis: Union[str, Redis] = None,
+                 loop: Optional[Union[asyncio.BaseEventLoop, asyncio.AbstractEventLoop]] = None,
+                 timeout: Optional[Union[int, float, requests.Timeout]] = None):
+        """
+
+        :param member_id:
+        :param id_qr:
+        :param tid:
+        :param client_id:
+        :param client_secret: l
+        i]
+        :param crt_file_path:
+        :param key_file_path:
+        :param pkcs12_password:
+        :param redis:
+        :param loop:
+        :param timeout:
+        """
+
+        self._main_loop = loop
+
+        self._member_id = member_id
+        self._sbp_member_id = "100000000111"
+        self._id_qr = id_qr
+        self._tid = tid
+        self._client_id = client_id
+        self._client_secret = client_secret
+
+        self._currency = "643"
+
+        class SSLAdapter(HTTPAdapter):
+            def init_poolmanager(self, *args, **kwargs):
+                context = create_urllib3_context()
+                context.load_cert_chain(certfile=crt_file_path, keyfile=key_file_path, password=pkcs12_password)
+                context.load_verify_locations(cafile=russian_crt)
+                kwargs['ssl_context'] = context
+                return super().init_poolmanager(*args, **kwargs)
+
+        self._session: Optional[requests.Session] = None
+        self._https_class = SSLAdapter()
+        if isinstance(redis, Redis):
+            self._redis = redis
+        else:
+            self._redis = Redis.from_url(redis, decode_responses=True) if redis is not None else None
+
+        self.timeout = timeout
+
+    def get_new_session(self) -> requests.Session:
+        session = requests.Session()
+        session.mount("https://", self._https_class)
+        return session
+
+    @property
+    def loop(self) -> Optional[asyncio.AbstractEventLoop]:
+        return self._main_loop
+
+    def get_session(self) -> Optional[requests.Session]:
+        if self._session is None:
+            self._session = self.get_new_session()
+
+        return self._session
+
+    def close(self):
+        """
+        Close all client sessions
+        """
+        if self._session:
+            self._session.close()
+
+    def request(self, method, headers, data):
+        headers = {**headers, **{'Accept': 'application/json', 'x-ibm-client-id': self._client_id}}
+        return make_request(self.get_session(), method, headers, data)
+
+    def get_token_from_redis(self, scope):
+        """
+        Возвращает токен, если он не истек
+        :param scope Область токена
+        :return: token string
+        """
+        return self._redis.get(f'{self._client_id}token_{scope.value}')
+
+    def token(self, scope: Scope):
+        redis_token = self.get_token_from_redis(scope) if self._redis is not None else None
+        if redis_token is not None:
+            return redis_token
+        else:
+            auth = base64.b64encode(f'{self._client_id}:{self._client_secret}'.encode('utf-8')).decode('utf-8')
+            headers = {'Authorization': f'Basic {auth}',
+                       'Content-Type': 'application/x-www-form-urlencoded',
+                       'rquid': ''.join(choices(hexdigits, k=32))}
+            data = {'grant_type': 'client_credentials', 'scope': scope.value}
+            token_data = self.request(Methods.oauth, headers, data)
+            if self._redis:
+                self._redis.set(f'{self._client_id}token_{scope.value}', token_data['access_token'],
+                                      int(token_data['expires_in']) - 10)
+            return token_data['access_token']
+
+    def creation(self, description: str, order_sum: int, order_number: str, positions: Union[List, Dict]):
+        """
+        Создание заказа
+        """
+        dt = f'{datetime.utcnow().isoformat(timespec="seconds")}Z'
+        rq_uid = ''.join(choices(hexdigits, k=32))
+        headers = {'Authorization': f'Bearer {self.token(Scope.create)}', 'RqUID': rq_uid}
+
+        rq_tm, order_create_date = dt, dt
+        member_id, id_qr, currency = self._member_id, self._id_qr, self._currency
+
+        sbp_member_id = self._sbp_member_id if self._tid == self._id_qr else None
+
+        if isinstance(positions, dict):
+            order_params_type = [positions]
+        else:
+            order_params_type = positions
+        del positions
+        payload = generate_payload(exclude=['dt', 'headers'], **locals())
+        return self.request(Methods.creation, headers, payload)
+
+    def status(self, order_id: str, partner_order_number: str):
+        rq_uid = ''.join(choices(hexdigits, k=32))
+        headers = {'Authorization': f'Bearer {self.token(Scope.status)}', 'RqUID': rq_uid}
+        tid = self._tid
+        rq_tm = f'{datetime.utcnow().isoformat(timespec="seconds")}Z'
+        payload = generate_payload(exclude=['headers'], **locals())
+        return self.request(Methods.status, headers, payload)
+
+    def revoke(self, order_id: str):
+        rq_uid = ''.join(choices(hexdigits, k=32))
+        headers = {'Authorization': f'Bearer {self.token(Scope.revoke)}', 'RqUID': rq_uid}
+
+        rq_tm = f'{datetime.utcnow().isoformat(timespec="seconds")}Z'
+        payload = generate_payload(exclude=['headers'], **locals())
+        return self.request(Methods.revocation, headers, payload)
+
+    def cancel(
+            self, order_id: str, operation_id: str, cancel_operation_sum: int, auth_code: str,
+            operation_type: CancelType = CancelType.REVERSE, sbp_payer_id: str = None
+    ):
+        """
+        Отмена/возврат
+        """
+        rq_uid = ''.join(choices(hexdigits, k=32))
+        headers = {'Authorization': f'Bearer {self.token(Scope.cancel)}', 'RqUID': rq_uid}
+
+        rq_tm = f'{datetime.utcnow().isoformat(timespec="seconds")}Z'
+        id_qr, tid, operation_currency = self._id_qr, self._tid, self._currency
+        operation_type = operation_type.value
+        payload = generate_payload(exclude=['headers'], **locals())
+        return self.request(Methods.cancel, headers, payload)
+
+    def registry(self, start_period: datetime, end_period: datetime,
+                       registry_type: RegistryType = RegistryType.REGISTRY):
+        """
+        Запрос реестра операций
+        """
+        rq_uid = ''.join(choices(hexdigits, k=32))
+        headers = {'Authorization': f'Bearer {self.token(Scope.registry)}', 'RqUID': rq_uid}
+        payload = {"rqUid": rq_uid,
+                   "rqTm": f'{datetime.utcnow().isoformat(timespec="seconds")}Z',
+                   "idQR": self._id_qr,
+                   "startPeriod": f'{start_period.isoformat(timespec="seconds")}Z',
+                   "endPeriod": f'{end_period.isoformat(timespec="seconds")}Z',
+                   "registryType": registry_type.value}
+
+        return self.request(Methods.registry, headers, payload)

SberQR/__init__.py

@@ -5,6 +5,7 @@
                        'Python 3.7+'.format('.'.join(map(str, sys.version_info[:3]))))
 
 from .AsyncSberQR import AsyncSberQR
+from .SberQr import SberQR
 from .api import make_request, Methods
 from .exceptions import (NetworkError, SberQrAPIError)
 

SberQR/api_sync.py

@@ -0,0 +1,20 @@
+from .api import check_result, Methods
+
+
+def make_request(session, method, headers, data, **kwargs):
+    url = f'https://mc.api.sberbank.ru/prod/{method}'
+
+    if method != Methods.oauth:
+        with session.post(url, json=data, headers=headers) as response:
+            try:
+                body = response.json()
+            except Exception:
+                body = response.text
+            return check_result(method, response.headers.get('Content-Type'), response.status_code, body)
+    else:
+        with session.post(url, data=data, headers=headers) as response:
+            try:
+                body = response.json()
+            except Exception:
+                body = response.text
+            return check_result(method, response.headers.get('Content-Type'), response.status_code, body)

requirements.txt

@@ -2,6 +2,7 @@ qrcode[pil]==7.4.2
 aiohttp~=3.8.4
 ujson>=5.8.0
 setuptools>=65.3.0
-aioredis>=2.0.1
+redis>=4.2.0rc1
 SberQR==1.0.3
-certifi==2023.7.22
+certifi==2023.7.22
+requests