Repository: leshchenko1979/fast_bitrix24
Pull request #87 (2020-11-25 11:27:07)

Асинхронный клиент BitrixAsync


Patch:
diff --git a/README.md b/README.md
index af3e3e4..2c85618 100644
--- a/README.md
+++ b/README.md
@@ -25,6 +25,9 @@ API wrapper для Питона для быстрого получения да
 - Позволяет задавать параметры запроса именно в таком виде, как они приведены в [документации к Bitrix24 REST API](https://dev.1c-bitrix.ru/rest_help/index.php). Параметры проверяются на корректность для облегчения отладки.
 - Выполнение запросов автоматически сопровождается прогресс-баром из пакета `tqdm`, иллюстрирующим не только количество обработанных элементов, но и прошедшее и оставшееся время выполнения запроса.
 
+### Синхронный и асинхронный клиенты
+- Наличие асинхронного клиента позволяет использовать библиотеку для написания веб-приложений (например, телеграм-ботов).
+
 ## Начало
 Установите модуль через `pip`:
 ```shell
@@ -118,7 +121,16 @@ results = b.call_batch ({
 })
 
 ```
-
+### Асинхронные вызовы
+Если требуется использование бибилиотеки в асинхронном коде, то вместо клиента `Bitrix()` создавайте клиент класса `BitrixAsync()`:
+```python
+from fast_bitrix24 import BitrixAsync
+b = BitrixAsync(webhook)
+```
+Все методы у него - синхронные аналоги методов из `Bitrix()`, описанных выше:
+```python
+leads = await b.get_all('crm.lead.list')
+```
 ## Как это работает
 1. Перед обращением к серверу во всех методах класса `Bitrix` происходит проверка корректности самых популярных параметров, передаваемых к серверу, и поднимаются исключения `TypeError` и `ValueError` при наличии ошибок.
 2. Cоздаются запросы на получение всех элементов из запрошенного списка.
diff --git a/fast_bitrix24/__init__.py b/fast_bitrix24/__init__.py
index 6228f98..95c695c 100644
--- a/fast_bitrix24/__init__.py
+++ b/fast_bitrix24/__init__.py
@@ -1,3 +1,3 @@
 '''Высокоуровневый API для доступа к Битрикс24'''
 
-from fast_bitrix24.bitrix import Bitrix, slow
+from fast_bitrix24.bitrix import Bitrix, BitrixAsync, slow
diff --git a/fast_bitrix24/bitrix.py b/fast_bitrix24/bitrix.py
index 7668bab..25c1861 100644
--- a/fast_bitrix24/bitrix.py
+++ b/fast_bitrix24/bitrix.py
@@ -26,7 +26,7 @@ class Bitrix:
         Создает объект класса Bitrix.
 
         '''
-        
+
         self.srh = ServerRequestHandler(webhook, verbose)
 
 
@@ -50,6 +50,7 @@ class Bitrix:
 
         return self.srh.run(GetAllUserRequest(self.srh, method, params).run())
 
+
     def get_by_ID(self, method: str, ID_list: Sequence, ID_field_name: str = 'ID',
         params: dict = None) -> list:
         '''
@@ -62,7 +63,7 @@ class Bitrix:
         Параметры:
         - method - метод REST API для запроса к серверу
         - ID_list - список ID
-        - ID_list_name - название поля, которе будет подаваться в запрос для 
+        - ID_list_name - название поля, которе будет подаваться в запрос для
             каждого элемента ID_list
         - params - параметры для передачи методу. Используется именно тот
             формат, который указан в документации к REST API Битрикс24
@@ -70,8 +71,8 @@ class Bitrix:
         Возвращает список кортежей вида:
 
             [
-                (ID, <результат запроса>), 
-                (ID, <результат запроса>), 
+                (ID, <результат запроса>),
+                (ID, <результат запроса>),
                 ...
             ]
 
@@ -83,13 +84,14 @@ class Bitrix:
 
         return self.srh.run(GetByIDUserRequest(self.srh, method, params, ID_list, ID_field_name).run())
 
+
     def call(self, method: str, items):
         '''
         Вызвать метод REST API по списку.
 
         Параметры:
         - method - метод REST API
-        - items - список параметров вызываемого метода 
+        - items - список параметров вызываемого метода
             либо dict с параметрами для единичного вызова
 
         Возвращает список ответов сервера для каждого из элементов items
@@ -102,7 +104,7 @@ class Bitrix:
             return self.srh.run(CallUserRequest(self.srh, method, [items]).run())[0]
         else:
             raise TypeError(f'call() accepts either a list of params dicts or a single params dict, but got a {type(items)} instead')
-    
+
 
     def call_batch(self, params: dict) -> dict:
         '''
@@ -114,4 +116,117 @@ class Bitrix:
         Возвращает ответы сервера.
         '''
 
-        return self.srh.run(BatchUserRequest(self.srh, params).run())
\ No newline at end of file
+        return self.srh.run(BatchUserRequest(self.srh, params).run())
+
+
+class BitrixAsync:
+    '''
+    Класс, повторяющий интерфейс класса Bitrix, но с асинхронными методами.
+
+    Параметры:
+    - webhook: str - URL вебхука, полученного от сервера Битрикс
+    - verbose: bool = True - показывать ли прогрессбар при выполнении запроса
+
+    Корутины:
+    - get_all(self, method: str, params: dict = None) -> list
+    - get_by_ID(self, method: str, ID_list: Sequence, ID_field_name: str = 'ID', params: dict = None) -> list
+    - call(self, method: str, item_list: Sequence) -> list
+    '''
+
+    def __init__(self, webhook: str, verbose: bool = True):
+        '''
+        Создает объект класса Bitrix.
+
+        '''
+
+        self.srh = ServerRequestHandler(webhook, verbose)
+
+
+    async def get_all(self, method: str, params: dict = None) -> list:
+        '''
+        Получить полный список сущностей по запросу method.
+
+        Под капотом использует параллельные запросы и автоматическое построение
+        батчей, чтобы ускорить получение данных. Также самостоятельно
+        обратывает постраничные ответы сервера, чтобы вернуть полный список.
+
+        Параметры:
+        - method - метод REST API для запроса к серверу
+        - params - параметры для передачи методу. Используется именно тот формат,
+                который указан в документации к REST API Битрикс24. get_all() не
+                поддерживает параметры 'start', 'limit' и 'order'.
+
+        Возвращает полный список сущностей, имеющихся на сервере,
+        согласно заданным методу и параметрам.
+        '''
+
+        return await self.srh.run_async(GetAllUserRequest(self.srh, method, params).run())
+
+
+    async def get_by_ID(self, method: str, ID_list: Sequence, ID_field_name: str = 'ID',
+        params: dict = None) -> list:
+        '''
+        Получить список сущностей по запросу method и списку ID.
+
+        Используется для случаев, когда нужны не все сущности,
+        имеющиеся в базе, а конкретный список поименованных ID.
+        Например, все контакты, привязанные к сделкам.
+
+        Параметры:
+        - method - метод REST API для запроса к серверу
+        - ID_list - список ID
+        - ID_list_name - название поля, которе будет подаваться в запрос для
+            каждого элемента ID_list
+        - params - параметры для передачи методу. Используется именно тот
+            формат, который указан в документации к REST API Битрикс24
+
+        Возвращает список кортежей вида:
+
+            [
+                (ID, <результат запроса>),
+                (ID, <результат запроса>),
+                ...
+            ]
+
+        Вторым элементом каждого кортежа будет результат выполнения запроса
+        относительно этого ID. Это может быть, например, список связанных
+        сущностей или пустой список, если не найдено ни одной привязанной
+        сущности.
+        '''
+
+        return await self.srh.run_async(GetByIDUserRequest(
+            self.srh, method, params, ID_list, ID_field_name).run())
+
+
+    async def call(self, method: str, items):
+        '''
+        Вызвать метод REST API по списку.
+
+        Параметры:
+        - method - метод REST API
+        - items - список параметров вызываемого метода
+            либо dict с параметрами для единичного вызова
+
+        Возвращает список ответов сервера для каждого из элементов items
+        либо просто результат для единичного вызова.
+        '''
+
+        if isinstance(items, Sequence):
+            return await self.srh.run_async(CallUserRequest(self.srh, method, items).run())
+        elif isinstance(items, dict):
+            return await self.srh.run_async(CallUserRequest(self.srh, method, [items]).run())[0]
+        else:
+            raise TypeError(f'call() accepts either a list of params dicts or a single params dict, but got a {type(items)} instead')
+
+
+    async def call_batch(self, params: dict) -> dict:
+        '''
+        Вызвать метод `batch`.
+
+        Параметры:
+        - params - список параметров вызываемого метода
+
+        Возвращает ответы сервера.
+        '''
+
+        return await self.srh.run_async(BatchUserRequest(self.srh, params).run())
\ No newline at end of file
diff --git a/fast_bitrix24/mult_request.py b/fast_bitrix24/mult_request.py
index 28b3513..e07f3ff 100644
--- a/fast_bitrix24/mult_request.py
+++ b/fast_bitrix24/mult_request.py
@@ -1,13 +1,16 @@
-import more_itertools
-from .utils import http_build_query
 import itertools
+from asyncio import ensure_future, as_completed
+
+import more_itertools
 
-from .srh import BITRIX_MAX_BATCH_SIZE
 from .server_response import ServerResponse
+from .srh import ServerRequestHandler, BITRIX_MAX_BATCH_SIZE
+from .utils import http_build_query
+
 
 class MultipleServerRequestHandler:
 
-    def __init__(self, srh, method, item_list, real_len=None, real_start=0):
+    def __init__(self, srh: ServerRequestHandler, method, item_list, real_len=None, real_start=0):
         self.srh = srh
         self.method = method
         self.item_list = item_list
@@ -15,7 +18,7 @@ class MultipleServerRequestHandler:
         self.real_start = real_start
         self.results = []
 
-                
+
     async def run(self):
         self.prepare_batches()
         self.prepare_tasks()
@@ -29,13 +32,13 @@ class MultipleServerRequestHandler:
         batches = [{
             'halt': 0,
             'cmd': {
-                self.batch_command_label(i, item): 
+                self.batch_command_label(i, item):
                 f'{self.method}?{http_build_query(item)}'
                 for i, item in enumerate(next_batch)
             }}
             for next_batch in more_itertools.chunked(self.item_list, batch_size)
         ]
-            
+
         self.method = 'batch'
         self.item_list = batches
 
@@ -45,14 +48,15 @@ class MultipleServerRequestHandler:
 
 
     def prepare_tasks(self):
-        for i in self.item_list:
-            self.srh.add_request_task(self.method, i)
+        self.tasks = []
+        for item in self.item_list:
+            self.tasks.append(ensure_future(self.srh.single_request(self.method, item)))
 
 
     async def get_results(self):
         self.pbar = self.srh.get_pbar(self.real_len, self.real_start)
 
-        for task in self.srh.get_server_serponses():
+        for task in as_completed(self.tasks):
             batch_response = await task
             unwrapped_result = ServerResponse(batch_response.result).result
             batch_results = self.extract_result_from_batch_response(unwrapped_result)
@@ -75,17 +79,17 @@ class MultipleServerRequestHandlerPreserveIDs(MultipleServerRequestHandler):
         super().__init__(srh, method, item_list)
         self.ID_field = ID_field
         self.original_item_list = item_list.copy()
-        
+
 
     async def run(self):
         await super().run()
         self.sort_results()
         return self.results
-                        
+
 
     def batch_command_label(self, i, item):
         return item[self.ID_field]
-    
+
 
     def extract_result_from_batch_response(self, unwrapped_result):
         result_list_of_tuples = unwrapped_result.items()
@@ -95,7 +99,7 @@ class MultipleServerRequestHandlerPreserveIDs(MultipleServerRequestHandler):
     def sort_results(self):
         # выделяем ID для облегчения дальнейшего поиска
         IDs_only = [str(i[self.ID_field]) for i in self.original_item_list]
-            
+
         # сортируем results на базе порядка ID в original_item_list
-        self.results.sort(key = lambda item: 
-            IDs_only.index(item[0]))
\ No newline at end of file
+        self.results.sort(key = lambda item:
+            IDs_only.index(item[0]))
diff --git a/fast_bitrix24/srh.py b/fast_bitrix24/srh.py
index 3e60a52..36751bf 100644
--- a/fast_bitrix24/srh.py
+++ b/fast_bitrix24/srh.py
@@ -1,7 +1,8 @@
 import asyncio
-import aiohttp
 import time
+from collections import deque
 
+import aiohttp
 from tqdm import tqdm
 
 from .server_response import ServerResponse
@@ -28,13 +29,13 @@ class ServerRequestHandler():
         self.webhook = self._standardize_webhook(webhook)
         self._verbose = verbose
 
-        self._stopped_time = None
-        self._stopped_value = None
         self.requests_per_second = BITRIX_RPS
-        self._pool_size = BITRIX_POOL_SIZE
+        self.pool_size = BITRIX_POOL_SIZE
 
+        self.active_requests = set()
         self.session = None
-        self.tasks = []
+
+        self.rr = deque() # rr - requests register - список отправленных запросов к серверу
 
 
     def _standardize_webhook(self, webhook):
@@ -54,145 +55,67 @@ class ServerRequestHandler():
 
 
     def run(self, coroutine):
-
-        async def async_wrapper(coroutine):
-            async with self:
-                result = await coroutine
-
-            if not _SLOW:
-                self.release_sem_task.cancel()
-
-            return result
-
         try:
             loop = asyncio.get_event_loop()
         except RuntimeError:
             asyncio.set_event_loop(asyncio.new_event_loop())
             loop = asyncio.get_event_loop()
 
-        result = loop.run_until_complete(async_wrapper(coroutine))
+        return loop.run_until_complete(self.run_async(coroutine))
 
-        return result
 
+    async def run_async(self, coroutine):
+        if not self.active_requests and (not self.session or self.session.closed):
+            self.session = aiohttp.ClientSession(raise_for_status=True)
 
-    def add_request_task(self, method, params):
-        self.tasks.append(asyncio.ensure_future(self._single_request(method, params)))
+        self.active_requests.add(coroutine)
 
+        result = await coroutine
 
-    def get_server_serponses(self):
-        global _SLOW
+        self.active_requests -= {coroutine}
 
-        tasks_to_process = len(self.tasks)
+        if not self.active_requests and self.session and not self.session.closed:
+            await self.session.close()
 
-        if not _SLOW:
-            self.release_sem_task = asyncio.ensure_future(self._release_sem())
-            self.tasks.append(self.release_sem_task)
+        return result
 
-        for task in asyncio.as_completed(self.tasks):
-            if tasks_to_process == 1:
-                self.tasks = []
-                yield task
-                break
-            else:
-                yield task
-                tasks_to_process -= 1
 
+    async def single_request(self, method, params=None):
+        await self._acquire()
+        async with self.session.post(url = self.webhook + method,
+                                     json = params) as response:
+            r = await response.json(encoding='utf-8')
+        return ServerResponse(r)
 
-    async def __aenter__(self):
-        global _SLOW
-        self._sem = asyncio.BoundedSemaphore(self._pool_size)
-        if _SLOW:
-            self._slow_lock = asyncio.Lock()
-        else:
-            if self._stopped_time:
-                '''
------v-----------------------------v---------------------
-     ^ - _stopped_time             ^ - current time
-     |-------- time_passed --------|
-     |- step -|- step -|- step |          - add_steps (whole steps to add)
-                               |- step -| - additional 1 step added
-                                   |-aw-| - additional waiting time
-                '''
-                time_passed = time.monotonic() - self._stopped_time
-
-                # сколько шагов должно было пройти
-                add_steps = time_passed / self.requests_per_second // 1
-
-                # сколько шагов могло пройти с учетом ограничений + еще один
-                real_add_steps = min(self._pool_size - self._stopped_value,
-                                    add_steps + 1)
-
-                # добавляем пропущенные шаги
-                self._sem._value += real_add_steps
-
-                # ждем время, излишне списанное при добавлении дополнительного шага
-                await asyncio.sleep((add_steps + 1) / self.requests_per_second - time_passed)
-
-                self._stopped_time = None
-                self._stopped_value = None
-
-        if not self.session or self.session.closed:
-            self.session = aiohttp.ClientSession(raise_for_status=True)
 
+    async def _acquire(self):
+        '''Ожидает, пока не станет безопасно делать запрос к серверу.'''
 
-    async def __aexit__(self, a1, a2, a3):
-        self._stopped_time = time.monotonic()
+        global _SLOW, _SLOW_RPS
+
+        # если пул заполнен, ждать
 
         if _SLOW:
-            # в slow-режиме обнуляем пул запросов, чтобы после выхода
-            # не выдать на сервер пачку запросов и не словить отказ
-            self._stopped_value = 0
+            await asyncio.sleep(1 / _SLOW_RPS)
         else:
-            self._stopped_value = self._sem._value
-
-        if self.session and not self.session.closed:
-            await self.session.close()
-
-
-    async def _release_sem(self):
-        '''
-        Корутина-метод, которая увеличивает счетчик доступных в пуле запросов.
+            if len(self.rr) >= self.pool_size:
+                time_from_last_request = time.monotonic() - self.rr[0]
+                time_to_wait = 1 / self.requests_per_second - time_from_last_request
+                if time_to_wait > 0:
+                    await asyncio.sleep(time_to_wait)
 
-        Должна запускаться единожды в параллели со всеми другими задачами, кроме случаев
-        выполнения в slow-режиме, когда она запускаться на должна.
-        '''
+        # зарегистрировать запрос
 
-        while True:
-            if self._sem._value < self._sem._bound_value:
-                self._sem.release()
-            await asyncio.sleep(1 / self.requests_per_second)
+        cur_time = time.monotonic()
+        self.rr.appendleft(cur_time)
 
+        # подчистить пул
 
-    async def _acquire(self):
-        '''
-        Вызов `await acquire()` должен предшествовать любому обращению
-        к серверу Битрикс. Он возвращает `True`, когда к серверу
-        можно осуществить запрос.
-
-        Использование:
-        ```
-        await self.aquire()
-        # теперь можно делать запросы
-        ...
-        ```
-        '''
-        global _SLOW, _SLOW_RPS
-        if _SLOW:
-            # ждать, пока отработают другие запросы, запущенные параллельно,
-            async with self._slow_lock:
-            # потом ждать основное время "остывания"
-                await asyncio.sleep(1 / _SLOW_RPS)
-            return True
-        else:
-            return await self._sem.acquire()
-
+        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()
 
-    async def _single_request(self, method, params=None):
-        await self._acquire()
-        async with self.session.post(url = self.webhook + method,
-                                     json = params) as response:
-            r = await response.json(encoding='utf-8')
-        return ServerResponse(r)
+        return
 
 
     def get_pbar(self, real_len, real_start):
@@ -211,15 +134,10 @@ class ServerRequestHandler():
             return MutePBar()
 
 
-##########################################
-#
-#   slow() context manager
-#
-##########################################
-
 _SLOW = False
 _SLOW_RPS = 0
 
+
 class slow:
     def __init__(self, requests_per_second = 0.5):
         global _SLOW_RPS
diff --git a/fast_bitrix24/user_request.py b/fast_bitrix24/user_request.py
index e41d03f..c3e7296 100644
--- a/fast_bitrix24/user_request.py
+++ b/fast_bitrix24/user_request.py
@@ -6,22 +6,23 @@ from collections.abc import Sequence
 from .mult_request import (MultipleServerRequestHandler,
                            MultipleServerRequestHandlerPreserveIDs)
 from .server_response import ServerResponse
+from .srh import ServerRequestHandler
 
 BITRIX_PAGE_SIZE = 50
 
 class UserRequestAbstract():
 
-    def __init__(self, srh, method: str, params: dict):
+    def __init__(self, srh: ServerRequestHandler, method: str, params: dict):
         self.srh = srh
         self.method = self.standardized_method(method)
         self.params = self.standardized_params(params) if params else None
         self.check_special_limitations()
-        
+
 
     def standardized_method(self, method):
         if not method:
             raise TypeError('Method cannot be empty')
-            
+
         if not isinstance(method, str):
             raise TypeError('Method should be a str')
 
@@ -29,10 +30,10 @@ class UserRequestAbstract():
 
         if method.lower().strip() == 'batch':
             raise ValueError("Method cannot be 'batch'. Use call_batch() instead.")
-        
+
         return method
-    
-    
+
+
     def standardized_params(self, p):
         if not isinstance(p, dict):
             raise TypeError('Params agrument should be a dict')
@@ -80,8 +81,8 @@ class UserRequestAbstract():
 
     def check_special_limitations(self):
         raise NotImplementedError
-    
-    
+
+
 class GetAllUserRequest(UserRequestAbstract):
     def check_special_limitations(self):
         if self.params and not set(self.params.keys()).isdisjoint(
@@ -89,7 +90,7 @@ class GetAllUserRequest(UserRequestAbstract):
         ):
             raise ValueError("get_all() doesn't support parameters 'start', 'limit' or 'order'")
 
-    
+
     async def run(self):
         self.add_order_parameter()
 
@@ -98,27 +99,26 @@ class GetAllUserRequest(UserRequestAbstract):
         if self.first_response.more_results_expected():
             await self.make_remaining_requests()
             self.dedup_results()
-                
+
         return self.results
 
 
     def add_order_parameter(self):
         # необходимо установить порядок сортировки, иначе сортировка будет рандомная
         # и сущности будут повторяться на разных страницах
-        
+
         order_clause = {'order': {'ID': 'ASC'}}
-        
+
         if self.params:
             if 'order' not in self.params:
                 self.params.update(order_clause)
         else:
             self.params = order_clause
 
-    
+
     async def make_first_request(self):
-        self.srh.add_request_task(self.method, self.params)
-        self.first_response = await next(self.srh.get_server_serponses())
-        self.results, self.total = self.first_response.result, self.first_response.total 
+        self.first_response = await self.srh.single_request(self.method, self.params)
+        self.results, self.total = self.first_response.result, self.first_response.total
 
 
     async def make_remaining_requests(self):
@@ -130,10 +130,10 @@ class GetAllUserRequest(UserRequestAbstract):
                 self.srh,
                 method = self.method,
                 item_list = item_list,
-                real_len = self.total, 
+                real_len = self.total,
                 real_start = len(self.results)
             ).run()
-        
+
         self.results.extend(remaining_results)
 
 
@@ -157,8 +157,8 @@ class GetByIDUserRequest(UserRequestAbstract):
         self.ID_list = ID_list
         self.ID_field_name = ID_field_name.upper().strip()
         super().__init__(srh, method, params)
-        
-        
+
+
     def check_special_limitations(self):
         if self.params and 'id' in self.params.keys():
             raise ValueError("get_by_ID() doesn't support parameter 'ID' within the 'params' argument")
@@ -170,34 +170,34 @@ class GetByIDUserRequest(UserRequestAbstract):
     async def run(self):
         if self.list_empty():
             return []
-        
+
         self.prepare_item_list()
-        
+
         results = await MultipleServerRequestHandlerPreserveIDs(
             self.srh,
             self.method,
             self.item_list,
             ID_field=self.ID_field_name
         ).run()
-        
+
         return results
 
-        
+
     def list_empty(self):
         return len(self.ID_list) == 0
-    
-    
+
+
     def prepare_item_list(self):
         if self.params:
             self.item_list = [
-                ChainMap({self.ID_field_name: ID}, self.params) 
+                ChainMap({self.ID_field_name: ID}, self.params)
                 for ID in self.ID_list
             ]
         else:
             self.item_list = [
-                {self.ID_field_name: ID} 
+                {self.ID_field_name: ID}
                 for ID in self.ID_list
-            ] 
+            ]
 
 
 class CallUserRequest(GetByIDUserRequest):
@@ -205,14 +205,14 @@ class CallUserRequest(GetByIDUserRequest):
         self.item_list = [self.standardized_params(item) for item in item_list]
         super().__init__(srh, method, None, None, '__order')
 
-        
+
     def check_special_limitations(self):
         if not isinstance(self.item_list, Sequence):
             raise TypeError("call(): 'item_list' should be a sequence")
 
     async def run(self):
         results = await super().run()
-        
+
         # убираем поле с порядковым номером из результатов
         return [item[1] for item in results]
 
@@ -220,11 +220,11 @@ class CallUserRequest(GetByIDUserRequest):
     def list_empty(self):
         return len(self.item_list) == 0
 
-    
+
     def prepare_item_list(self):
         # добавим порядковый номер
         self.item_list = [
-            ChainMap(item, {self.ID_field_name: 'order' + str(i)}) 
+            ChainMap(item, {self.ID_field_name: 'order' + str(i)})
             for i, item in enumerate(self.item_list)
         ]
 
@@ -242,15 +242,14 @@ class BatchUserRequest(UserRequestAbstract):
     def check_special_limitations(self):
         if not self.params:
             raise ValueError("Params for a batch call can't be empty")
-        
+
         if {'halt', 'cmd'} != self.params.keys():
             raise ValueError("Params for a batch call should contain only 'halt' and 'cmd' clauses at the highest level")
 
         if not isinstance(self.params['cmd'], dict):
             raise ValueError("'cmd' clause should contain a dict")
-    
+
 
     async def run(self):
-        self.srh.add_request_task(self.method, self.params)
-        response = await next(self.srh.get_server_serponses())
+        response = await self.srh.single_request(self.method, self.params)
         return ServerResponse(response.result).result
diff --git a/setup.py b/setup.py
index 64fbb1d..25eccbe 100644
--- a/setup.py
+++ b/setup.py
@@ -8,7 +8,7 @@ with open('requirements.txt', 'r', encoding='utf-8') as fh:
 
 setuptools.setup(
     name="fast_bitrix24",
-    version="0.4.7",
+    version="0.5.0",
     author="Alexey Leshchenko",
     author_email="leshchenko@gmail.com",
     description="API wrapper для быстрого получения данных от Битрикс24 через REST API. Параллельные запросы к серверу, упаковка запросов в батчи, контроль скорости запросов.",


Test patch:
diff --git a/.github/workflows/full-test.yml b/.github/workflows/full-test.yml
index 62eb612..db3bfcf 100644
--- a/.github/workflows/full-test.yml
+++ b/.github/workflows/full-test.yml
@@ -28,7 +28,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip setuptools wheel
-        pip install flake8 pytest
+        pip install flake8 pytest pytest-asyncio
         pip install -r requirements.txt
     - name: Lint with flake8
       run: |
diff --git a/.github/workflows/small-test.yml b/.github/workflows/small-test.yml
index 8a891d6..cfe5a47 100644
--- a/.github/workflows/small-test.yml
+++ b/.github/workflows/small-test.yml
@@ -24,7 +24,7 @@ jobs:
         python-version: 3.9
     - name: Install dependencies
       run: |
-        python -m pip install --upgrade pip setuptools wheel pytest
+        python -m pip install --upgrade pip setuptools wheel pytest pytest-asyncio
         pip install -r requirements.txt
     - name: Test with pytest
       env:
diff --git a/fast_bitrix24/test_.py b/fast_bitrix24/test_.py
index cbfb79f..38dfd49 100644
--- a/fast_bitrix24/test_.py
+++ b/fast_bitrix24/test_.py
@@ -1,9 +1,14 @@
-from .utils import http_build_query
+import asyncio
 import os
+from asyncio import gather
+from time import monotonic
 
 import pytest
 
-from fast_bitrix24 import Bitrix, slow
+from fast_bitrix24 import Bitrix, BitrixAsync, slow
+
+from .srh import ServerRequestHandler
+from .utils import http_build_query
 
 
 @pytest.fixture(scope='session')
@@ -15,6 +20,15 @@ def get_test():
         raise RuntimeError('Environment variable FAST_BITRIX24_TEST_WEBHOOK should be set to the webhook of your test Bitrix24 account.')
 
 
+@pytest.fixture(scope='session')
+def get_test_async():
+    test_webhook = os.getenv('FAST_BITRIX24_TEST_WEBHOOK')
+    if test_webhook:
+        return BitrixAsync(test_webhook)
+    else:
+        raise RuntimeError('Environment variable FAST_BITRIX24_TEST_WEBHOOK should be set to the webhook of your test Bitrix24 account.')
+
+
 @pytest.fixture(scope='session')
 def create_100_leads(get_test) -> Bitrix:
     b = get_test
@@ -41,6 +55,33 @@ def create_100_leads(get_test) -> Bitrix:
     b.get_by_ID('crm.lead.delete', lead_nos)
 
 
+@pytest.fixture(scope='function')
+@pytest.mark.asyncio
+async def create_100_leads_async(get_test_async) -> BitrixAsync:
+    b = get_test_async
+
+    # Подчистить тестовый аккаунт от лишних сущностей,
+    # созданных при неудачных тестах, чтобы не было блокировки
+    # аккаунта при создании более 1000 сущностей.
+    # Скорее всего, вызовет проблемы в параллельно
+    # запущенных тестах.
+    total_leads = len(await b.get_all('crm.lead.list'))
+    if total_leads > 500:
+        leads = await b.get_all('crm.lead.list', {'select': ['ID']})
+        await b.get_by_ID('crm.lead.delete', [l['ID'] for l in leads])
+
+    with slow(1.2):
+        lead_nos = await b.call('crm.lead.add', [{
+            'fields': {
+                'NAME': f'Customer #{n}',
+            }
+        } for n in range(100)])
+
+    yield b
+
+    await b.get_by_ID('crm.lead.delete', lead_nos)
+
+
 class TestBasic:
 
     def test_simple_add_lead(self, get_test):
@@ -236,3 +277,89 @@ class TestHttpBuildQuery:
 
         test = http_build_query(d)
         assert test == 'FILTER[%21STATUS_ID]=CLOSED&'
+
+
+class TestAsync:
+
+    @pytest.mark.asyncio
+    async def test_simple_async_calls(self, get_test_async):
+
+        b = get_test_async
+        await b.get_all('crm.lead.list')
+
+
+    @pytest.mark.asyncio
+    async def test_simple_async_calls(self, create_100_leads_async):
+
+        b = create_100_leads_async
+        await gather(b.get_all('crm.lead.list'), b.get_all('crm.lead.list'))
+
+
+class TestAcquire:
+
+    @pytest.mark.asyncio
+    async def test_acquire_sequential(self):
+
+        async def assert_time_acquire(pool_size, requests_per_second, acquire_amount, time_expected):
+            srh = ServerRequestHandler('http://www.bitrix24.ru/path', False)
+
+            srh.pool_size = pool_size
+            srh.requests_per_second = requests_per_second
+
+            t1 = monotonic()
+
+            for _ in range(acquire_amount):
+                await srh._acquire()
+
+            t2 = monotonic()
+
+            assert time_expected - 0.1 < t2 - t1 < time_expected + 0.1
+
+        await assert_time_acquire(1, 1, 1, 0)
+        await assert_time_acquire(10, 1, 10, 0)
+        await assert_time_acquire(1, 10, 2, 0.1)
+        await assert_time_acquire(50, 10, 60, 1)
+
+
+    @pytest.mark.asyncio
+    async def test_acquire_intermittent(self):
+
+        srh = ServerRequestHandler('http://www.bitrix24.ru/path', False)
+
+        srh.pool_size = 10
+        srh.requests_per_second = 10
+
+        async def assert_time_expected(acquire_times, time_expected):
+            t1 = monotonic()
+
+            for _ in range(acquire_times):
+                await srh._acquire()
+
+            t2 = monotonic()
+
+            assert time_expected - 0.1 < t2 - t1 < time_expected + 0.1
+
+        await assert_time_expected(10, 0)
+        await asyncio.sleep(0.3)
+        await assert_time_expected(10, 0.7)
+
+
+    @pytest.mark.asyncio
+    async def test_acquire_sequential_slow(self):
+
+        async def assert_time_acquire_slow(pool_size, requests_per_second, acquire_amount, time_expected):
+            srh = ServerRequestHandler('http://www.bitrix24.ru/path', False)
+
+            srh.pool_size = pool_size
+
+            t1 = monotonic()
+
+            with slow(requests_per_second):
+                for _ in range(acquire_amount):
+                    await srh._acquire()
+
+            t2 = monotonic()
+
+            assert time_expected - 0.1 < t2 - t1 < time_expected + 0.1
+
+        await assert_time_acquire_slow(1, 10, 5, 0.5)