Repository: leshchenko1979/fast_bitrix24 (python)

Issue:
Делать повторные запросы при получении ошибок внутри ServerRequestHandler.single_request()
Ошибки, при которых делать повторы:
- aiohttp.client_exceptions.ClientPayloadError
- aiohttp.client_exceptions.ServerDisconnectedError

На третий раз можно поднимать полученную ошибку наверх.

Pull request description:
Делать повторные запросы при получении ошибок внутри ServerRequestHandler.single_request()
Fixes #112

Patch:
diff --git a/fast_bitrix24/srh.py b/fast_bitrix24/srh.py
index 8246bd9..0927a34 100644
--- a/fast_bitrix24/srh.py
+++ b/fast_bitrix24/srh.py
@@ -4,10 +4,13 @@ from collections import deque
 from contextlib import asynccontextmanager
 
 import aiohttp
+from aiohttp.client_exceptions import (ClientPayloadError, ClientResponseError,
+                                       ServerDisconnectedError)
 from tqdm import tqdm
 
 from .server_response import ServerResponse
-from .utils import _url_valid
+from .utils import _url_valid, retry
+
 
 BITRIX_POOL_SIZE = 50
 BITRIX_RPS = 2.0
@@ -15,6 +18,10 @@ BITRIX_MAX_BATCH_SIZE = 50
 BITRIX_MAX_CONCURRENT_REQUESTS = 20
 
 
+class ServerError(Exception):
+    pass
+
+
 class ServerRequestHandler():
     '''
     Используется для контроля скорости доступа к серверам Битрикс.
@@ -91,18 +98,24 @@ class ServerRequestHandler():
         if not self.active_runs and self.session and not self.session.closed:
             await self.session.close()
 
+    @retry(exceptions=[ClientPayloadError, ServerDisconnectedError,
+                       ServerError])
     async def single_request(self, method, params=None):
         '''Делает единичный запрос к серверу, ожидая при необходимости.'''
 
-        await self._acquire()
-
-        async with self.concurrent_requests_sem, self.session.post(
-                url=self.webhook + method, json=params) as response:
-            r = await response.json(encoding='utf-8')
-
-        return ServerResponse(r)
+        try:
+            async with self.concurrent_requests_sem, self.acquire(), \
+                    self.session.post(url=self.webhook + method, json=params) \
+                    as response:
+                return ServerResponse(await response.json(encoding='utf-8'))
+        except ClientResponseError as error:
+            if error.status // 100 == 5:  # ошибки вида 5XX
+                raise ServerError('The server returned an error') from error
+            else:
+                raise
 
-    async def _acquire(self):
+    @asynccontextmanager
+    async def acquire(self):
         '''Ожидает, пока не станет безопасно делать запрос к серверу.'''
 
         # если пул заполнен, ждать
@@ -114,18 +127,17 @@ class ServerRequestHandler():
                 await asyncio.sleep(time_to_wait)
 
         # зарегистрировать запрос в очереди
-
         cur_time = time.monotonic()
         self.rr.appendleft(cur_time)
 
-        # подчистить пул
+        # отдать управление
+        yield
 
+        # подчистить пул
         trim_time = cur_time - self.pool_size / self.requests_per_second
         while self.rr and self.rr[len(self.rr) - 1] < trim_time:
             self.rr.pop()
 
-        return
-
     def get_pbar(self, real_len, real_start):
         '''Возвращает прогресс бар `tqdm()` или пустышку,
         если `self.verbose is False`.'''
diff --git a/fast_bitrix24/utils.py b/fast_bitrix24/utils.py
index b482f4d..d8f9e08 100644
--- a/fast_bitrix24/utils.py
+++ b/fast_bitrix24/utils.py
@@ -1,11 +1,8 @@
-##########################################
-#
-#   internal functions
-#
-##########################################
-
+from functools import wraps
+from time import sleep
 from urllib.parse import quote, urlparse
 
+
 def _url_valid(url):
     try:
         result = urlparse(url)
@@ -27,7 +24,8 @@ def http_build_query(params, convention="%s"):
 
         elif type(params[key]) is list:
 
-            new_params = {str(i): element for i, element in enumerate(params[key])}
+            new_params = {str(i): element for i, element
+                          in enumerate(params[key])}
 
             output += http_build_query(
                 new_params, convention % key + "[%s]")
@@ -38,4 +36,36 @@ def http_build_query(params, convention="%s"):
             key = quote(key)
             output = output + convention % key + "=" + val + "&"
 
-    return output
\ No newline at end of file
+    return output
+
+
+def retry(exceptions, total_tries=3, initial_wait=0.5, backoff_factor=2):
+    """
+    Calling the decorated function applying an exponential backoff.
+    Args:
+        exceptions: Exeption(s) that trigger a retry, can be a tuple
+        total_tries: Total tries
+        initial_wait: Time to first retry
+        backoff_factor: Backoff multiplier (e.g. value of 2 will double
+            the delay each retry).
+    """
+    def retry_decorator(f):
+
+        @wraps(f)
+        def func_with_retries(*args, **kwargs):
+
+            _tries, _delay = total_tries + 1, initial_wait
+
+            while _tries > 1:
+                try:
+                    return f(*args, **kwargs)
+                except tuple(exceptions):
+                    _tries -= 1
+                    if _tries == 1:
+                        raise
+                    sleep(_delay)
+                    _delay *= backoff_factor
+
+        return func_with_retries
+
+    return retry_decorator


Test patch:
diff --git a/fast_bitrix24/tests/test_async.py b/fast_bitrix24/tests/test_async.py
index 4d348f5..bf39070 100644
--- a/fast_bitrix24/tests/test_async.py
+++ b/fast_bitrix24/tests/test_async.py
@@ -52,7 +52,8 @@ async def assert_time_acquire(bitrix, acquire_amount, time_expected):
     t1 = monotonic()
 
     for _ in range(acquire_amount):
-        await bitrix.srh._acquire()
+        async with bitrix.srh.acquire():
+            pass
 
     t2 = monotonic()