Repository: leshchenko1979/fast_bitrix24
Pull request #141 (2021-01-23 15:11:25)

Исправить `limit_request_velocity` и добавить параметр `respect_velocity_policy`


Patch:
diff --git a/.gitignore b/.gitignore
index d54dfc4..cf4af0b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -130,4 +130,4 @@ dmypy.json
 .pypirc
 tests_private/**
 .vscode/settings.json
-speed_tests/karpov.py
+speed_tests/webhook.py
diff --git a/README.md b/README.md
index f44c3cb..676b81a 100644
--- a/README.md
+++ b/README.md
@@ -141,7 +141,7 @@ leads = await b.get_all('crm.lead.list')
 1. Перед обращением к серверу во всех методах класса `Bitrix` происходит проверка корректности самых популярных параметров, передаваемых к серверу, и поднимаются исключения `TypeError` и `ValueError` при наличии ошибок.
 2. Cоздаются запросы на получение всех элементов из запрошенного списка.
 3. Созданные запросы упаковываются в батчи по 50 запросов в каждом.
-4. Полученные батчи параллельно отправляются на сервер с соблюдением установленных скоростных ограничений (см. ниже "Как Битрикс24 ограничивает скорость заросов").
+4. Полученные батчи параллельно отправляются на сервер с регулировкой скорости запросов (см. ниже "Как fast_bitrix24 регулирует скорость запросов").
 5. Ответы (содержимое поля `result`) собираются в единый плоский список и возвращаются пользователю.
     - Поднимаются исключения класса `aiohttp.ClientError`, если сервер Битрикс вернул HTTP-ошибку, и `RuntimeError`, если код ответа был `200`, но ошибка сдержалась в теле ответа сервера.
     - Происходит сортировка ответов (кроме метода `get_all()`) - порядок элементов в списке результатов совпадает с порядком соответствующих запросов в списке запросов.
@@ -153,23 +153,27 @@ leads = await b.get_all('crm.lead.list')
 
 В связи с тем, что выполнение `get_all()` по длинным спискам может занимать долгое время, в течение которого пользователи могут добавлять новые элементы в список, может возникнуть ситуация, когда общее полученное количество элементов может не соответствовать изначальному значению `total`. В таких случаях будет выдано стандартное питоновское предупреждение (`warning`).
 
-### Как Битрикс24 ограничивает скорость запросов
+### Как `fast_bitrix24` регулирует скорость запросов
+По умолчанию `fast_bitrix24` игнорирует официальные ограничения Битрикс24 по скорости запросов (см. ниже "Официальная политика Битрикс24 по скорости запросов") и вместо этого начинает снижать скорость запросов, если сервер начинает возвращать ошибки (autothrottling). Подобный подход позволяет на порядки увеличить скорость получения данных (см. [тесты скорости](https://github.com/leshchenko1979/fast_bitrix24/blob/master/speed_tests/strategies.ipynb)).
+
+Чтобы соблюдать официальные правила, при создании экземпляра класса `Bitrix` укажите параметр `respect_velocity_policy=True`.
+### Официальная политика Битрикс24 по скорости запросов
 1. Существует пул из 50 запросов, которые можно направить без ожидания.
 2. Пул пополняется со скоростью 2 запроса в секунду.
-3. При исчерпании пула и несоблюдении режима ожидания сервер выдаёт ответ `429 Too Many Requests`.
+3. При исчерпании пула и несоблюдении режима ожидания сервер возвращает ошибку.
 
 ## Подробный справочник по классу `Bitrix`
 Объект класса `Bitrix` создаётся, чтобы через него выполнять все запросы к серверу Битрикс24.
 
 Внутри объекта ведётся учёт скорости отправки запросов к серверу, поэтому важно, чтобы все запросы приложения в отношении одного аккаунта с одного IP-адреса отправлялись из одного экземпляра `Bitrix`.
 
-### Метод ` __init__(self, webhook: str, verbose: bool = True):`
+### Метод ` __init__(self, webhook: str, verbose: bool = True, respect_velocity_policy: bool = False):`
 Создаёт экземпляр объекта `Bitrix`.
 
 #### Параметры
 * `webhook: str` - URL вебхука, полученного от сервера Битрикс.
-
 * `verbose: bool = True` - показывать прогрессбар при выполнении запроса.
+* `respect_velocity_policy: bool = False` - соблюдать политику Битрикса о скорости запросов.
 
 ### Метод `get_all(self, method: str, params: dict = None) -> list | dict`
 Получить полный список сущностей по запросу `method`.
@@ -226,8 +230,6 @@ contacts = b.get_by_ID(
 
 Ключом каждого элемента возвращаемого словаря будет ID из списка `ID_list`. Значением будет результат выполнения запроса относительно этого ID. Это может быть, например, список связанных сущностей или пустой список, если не найдено ни одной привязанной сущности.
 
-`get_by_ID()` гарантированно возвращает словарь такой же длины, как и поданный на вход `ID_list`.
-
 ### Метод `list_and_get(self, method_branch: str, ID_field_name='ID') -> dict`
 Скачать список всех ID при помощи метода `method_branch + '.list'`,
 а затем значения всех полей всех элементов при помощи метода `method_branch + '.get'`. `method_branch` - группа методов, в которой есть подметоды `*.list` и `*.get`, например `crm.lead` или `tasks.task`.
diff --git a/fast_bitrix24/bitrix.py b/fast_bitrix24/bitrix.py
index 9996719..111895b 100644
--- a/fast_bitrix24/bitrix.py
+++ b/fast_bitrix24/bitrix.py
@@ -12,7 +12,8 @@ from .user_request import (BatchUserRequest, CallUserRequest,
 
 class BitrixAbstract(object):
 
-    def __init__(self, webhook: str, verbose: bool = True):
+    def __init__(self, webhook: str, verbose: bool = True,
+                 respect_velocity_policy: bool = False):
         '''
         Создает объект класса Bitrix.
 
@@ -20,11 +21,14 @@ class BitrixAbstract(object):
         - `webhook: str` - URL вебхука, полученного от сервера Битрикс
         - `verbose: bool = True` - показывать ли прогрессбар при выполнении
         запроса
+        - `respect_velocity_policy: bool = False` - соблюдать ли политику
+        Битрикса о скорости запросов
         '''
 
-        self.srh = ServerRequestHandler(webhook)
+        self.srh = ServerRequestHandler(webhook, respect_velocity_policy)
         self.verbose = verbose
 
+
 class Bitrix(BitrixAbstract):
     '''Клиент для запросов к серверу Битрикс24.'''
 
diff --git a/fast_bitrix24/srh.py b/fast_bitrix24/srh.py
index b07a7e0..043e638 100644
--- a/fast_bitrix24/srh.py
+++ b/fast_bitrix24/srh.py
@@ -43,8 +43,9 @@ class ServerRequestHandler():
     последовательных запросов к серверу.
     '''
 
-    def __init__(self, webhook):
+    def __init__(self, webhook, respect_velocity_policy):
         self.webhook = self.standardize_webhook(webhook)
+        self.respect_velocity_policy = respect_velocity_policy
 
         self.requests_per_second = BITRIX_RPS
         self.pool_size = BITRIX_POOL_SIZE
@@ -173,9 +174,12 @@ class ServerRequestHandler():
 
         await self.autothrottle()
 
-        async with self.limit_concurrent_requests(), \
-                self.limit_request_velocity():
-            yield
+        async with self.limit_concurrent_requests():
+            if self.respect_velocity_policy:
+                async with self.limit_request_velocity():
+                    yield
+            else:
+                yield
 
     async def autothrottle(self):
         '''Если было несколько неудач, делаем таймаут и уменьшаем скорость
@@ -218,16 +222,18 @@ class ServerRequestHandler():
         '''Ограничивает скорость запросов к серверу.'''
 
         # если пул заполнен, ждать
-        if len(self.rr) >= self.pool_size:
+        while 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 sleep(time_to_wait)
+            else:
+                break
 
         # зарегистрировать запрос в очереди
-        cur_time = time.monotonic()
-        self.rr.appendleft(cur_time)
+        start_time = time.monotonic()
+        self.rr.appendleft(start_time)
 
         # отдать управление
         try:
@@ -235,6 +241,6 @@ class ServerRequestHandler():
 
         # подчистить пул
         finally:
-            trim_time = cur_time - self.pool_size / self.requests_per_second
-            while self.rr and self.rr[len(self.rr) - 1] < trim_time:
+            trim_time = start_time - self.pool_size / self.requests_per_second
+            while self.rr[-1] < trim_time:
                 self.rr.pop()


Test patch:
diff --git a/speed_tests/strategies.ipynb b/speed_tests/strategies.ipynb
index e69de29..b0e1c18 100644
--- a/speed_tests/strategies.ipynb
+++ b/speed_tests/strategies.ipynb
@@ -0,0 +1,113 @@
+{
+ "metadata": {
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.8.6-final"
+  },
+  "orig_nbformat": 2,
+  "kernelspec": {
+   "name": "python3",
+   "display_name": "Python 3.8.6 64-bit",
+   "metadata": {
+    "interpreter": {
+     "hash": "1baa965d5efe3ac65b79dfc60c0d706280b1da80fedb7760faf2759126c4f253"
+    }
+   }
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2,
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": 5,
+   "metadata": {},
+   "outputs": [
+    {
+     "output_type": "stream",
+     "name": "stderr",
+     "text": [
+      "100%|██████████| 41537/41537 [00:24<00:00, 1721.99it/s]\n"
+     ]
+    },
+    {
+     "output_type": "error",
+     "ename": "AssertionError",
+     "evalue": "",
+     "traceback": [
+      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
+      "\u001b[1;31mAssertionError\u001b[0m                            Traceback (most recent call last)",
+      "\u001b[1;32m<ipython-input-5-570fdb38f5be>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m\u001b[0m\n\u001b[0;32m      8\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      9\u001b[0m \u001b[0mresult_list_and_get\u001b[0m \u001b[1;33m=\u001b[0m \u001b[1;32mawait\u001b[0m \u001b[0mbitrix\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mlist_and_get\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'crm.lead'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 10\u001b[1;33m \u001b[1;32massert\u001b[0m \u001b[0mall\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mlen\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mfields\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;33m==\u001b[0m \u001b[0mlen\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mx\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;32mfor\u001b[0m \u001b[0m_\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mx\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mresult_list_and_get\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mitems\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     11\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     12\u001b[0m \u001b[0mresult_get_all\u001b[0m \u001b[1;33m=\u001b[0m \u001b[1;32mawait\u001b[0m \u001b[0mbitrix\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mget_all\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'crm.lead.list'\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m{\u001b[0m\u001b[1;34m'select'\u001b[0m\u001b[1;33m:\u001b[0m \u001b[0mfields\u001b[0m\u001b[1;33m}\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
+      "\u001b[1;31mAssertionError\u001b[0m: "
+     ]
+    }
+   ],
+   "source": [
+    "from karpov import webhook\n",
+    "from fast_bitrix24 import BitrixAsync\n",
+    "from contextlib import asynccontextmanager\n",
+    "\n",
+    "bitrix = BitrixAsync(webhook)\n",
+    "\n",
+    "fields = (await bitrix.get_all('crm.lead.fields')).keys()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "result_list_and_get = await bitrix.list_and_get('crm.lead')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 11,
+   "metadata": {},
+   "outputs": [
+    {
+     "output_type": "execute_result",
+     "data": {
+      "text/plain": [
+       "True"
+      ]
+     },
+     "metadata": {},
+     "execution_count": 11
+    }
+   ],
+   "source": [
+    "lens = [len(x) >= 85 for x in result_list_and_get.values()]\n",
+    "all(lens)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "result_get_all = await bitrix.get_all('crm.lead.list', {'select': fields})"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "lens = [len(x) >= 85 for x in result_get_all]\n",
+    "all(lens)"
+   ]
+  }
+ ]
+}
\ No newline at end of file
diff --git a/tests/test_async.py b/tests/test_async.py
index 39e2cc6..c9cd705 100644
--- a/tests/test_async.py
+++ b/tests/test_async.py
@@ -1,10 +1,13 @@
-import asyncio
-from asyncio import gather
+from asyncio import create_task, ensure_future, gather, sleep, wait
+from collections import namedtuple
+from contextlib import asynccontextmanager
 from time import monotonic
 
 import pytest
-
 from fast_bitrix24 import BitrixAsync
+from fast_bitrix24.srh import (BITRIX_POOL_SIZE, BITRIX_RPS,
+                               ServerRequestHandler)
+
 from .fixtures import (create_100_leads, create_100_leads_async, create_a_lead,
                        get_test, get_test_async)
 
@@ -17,7 +20,7 @@ class TestAsync:
         b: BitrixAsync = create_100_leads_async
 
         leads = await b.get_all('crm.lead.list')
-        await b.get_by_ID('crm.lead.list', [lead['ID'] for lead in leads])
+        await b.get_by_ID('crm.lead.get', [lead['ID'] for lead in leads])
         await b.list_and_get('crm.lead')
         await b.call('crm.lead.get', {'ID': leads[0]['ID']})
         await b.call_batch({
@@ -41,7 +44,8 @@ class TestAsync:
 
 
 def get_custom_bitrix(pool_size, requests_per_second):
-    bitrix = BitrixAsync('http://www.bitrix24.ru/path')
+    bitrix = BitrixAsync('http://www.bitrix24.ru/path',
+                         respect_velocity_policy=True)
 
     bitrix.srh.pool_size = pool_size
     bitrix.srh.requests_per_second = requests_per_second
@@ -68,15 +72,191 @@ class TestAcquire:
 
         await assert_time_acquire(get_custom_bitrix(1, 1), 1, 0)
         await assert_time_acquire(get_custom_bitrix(10, 1), 10, 0)
-        await assert_time_acquire(get_custom_bitrix(1, 10), 3, 0.2)
+        await assert_time_acquire(get_custom_bitrix(1, 5), 5, 0.8)
         await assert_time_acquire(get_custom_bitrix(50, 10), 60, 1)
 
-
     @pytest.mark.asyncio
     async def test_acquire_intermittent(self):
 
         bitrix = get_custom_bitrix(10, 10)
 
         await assert_time_acquire(bitrix, 10, 0)
-        await asyncio.sleep(0.3)
+        await sleep(0.3)
         await assert_time_acquire(bitrix, 10, 0.7)
+
+    @pytest.mark.asyncio
+    async def test_acquire_speed(self):
+
+        i = 100
+        bitrix = get_custom_bitrix(50, 2)
+
+        start = monotonic()
+
+        while i > 50:
+            async with bitrix.srh.acquire():
+                i -= 1
+
+        elapsed = monotonic() - start
+        assert elapsed < 1
+
+        await sleep(10)
+
+        while i:
+            async with bitrix.srh.acquire():
+                i -= 1
+
+        elapsed = monotonic() - start
+        assert 10 + 15 < elapsed < 10 + 15 + 1
+
+
+class MockResponse(object):
+    def __init__(self, stored_json=None):
+        self.stored_json = stored_json
+
+    async def json(self, **args):
+        return self.stored_json
+
+
+class MockSession(object):
+    def __init__(self, post_callback):
+        self.post_callback = post_callback
+        self.pool = BITRIX_POOL_SIZE
+        self.rps = BITRIX_RPS
+        self.num_requests = 0
+
+    @asynccontextmanager
+    async def post(self, url, json):
+        self.pool -= 1
+        self.num_requests += 1
+
+        if self.pool < 0:
+            raise RuntimeError(
+                f'Pool exhausted after {self.num_requests} requests')
+
+        yield self.post_callback(self, url, json)
+
+
+class MockSRH(ServerRequestHandler):
+    def __init__(self, post_callback):
+        super().__init__('http://www.google.com/',
+                         respect_velocity_policy=True)
+        self.post_callback = post_callback
+
+    @asynccontextmanager
+    async def handle_sessions(self):
+        self.session = MockSession(self.post_callback)
+        yield
+
+    async def restore_pool(self):
+        while True:
+            if self.session.pool < BITRIX_POOL_SIZE:
+                self.session.pool += 1
+            await sleep(1 / self.session.rps)
+
+
+class TestMocks:
+    @pytest.mark.asyncio
+    async def test_mock(self):
+
+        bitrix = BitrixAsync('http://www.google.com/')
+        bitrix.srh = MockSRH(lambda *args: MockResponse({'result': ('OK',)}))
+
+        assert await bitrix.get_all('abc') == ('OK',)
+
+    @pytest.mark.asyncio
+    async def test_mock_get_all(self):
+
+        record_ID = iter(range(1_000_000))
+
+        def post_callback(self: ServerRequestHandler, url: str, json: dict):
+
+            if 'batch' not in url:
+                page = [{'ID': next(record_ID)} for _ in range(50)]
+                response = {
+                    'result': page,
+                    'total': 5000
+                }
+
+            else:
+                cmds = {command: [{'ID': next(record_ID)} for _ in range(50)]
+                        for command in json['cmd']}
+                response = {'result': {'result': cmds,
+                                       'total': 5000}}
+
+            return MockResponse(response)
+
+        bitrix = BitrixAsync('http://www.google.com/')
+        bitrix.srh = MockSRH(post_callback)
+
+        result = await bitrix.get_all('abc')
+        assert bitrix.srh.session.num_requests == 3
+        assert len(result) == 5000
+
+    @pytest.mark.asyncio
+    async def test_get_by_ID(self):
+
+        ParsedCommand = namedtuple('ParsedCommand', ['metod', 'params'])
+
+        def post_callback(self: ServerRequestHandler, url: str, json: dict):
+
+            def parse_command(value: str):
+                split = value.split('?')
+                method, param_str = split[0], split[1]
+                pairs = param_str.split('&')
+                split = (pair.split('=') for pair in pairs if pair)
+                params = {key: value for key, value in split}
+                return ParsedCommand(method, params)
+
+            commands = {key: parse_command(value)
+                        for key, value in json['cmd'].items()}
+
+            items = {label: {'ID': parsed.params['ID']}
+                     for label, parsed in commands.items()}
+
+            response = {'result': {'result': items}}
+
+            return MockResponse(response)
+
+        bitrix = BitrixAsync('http://www.google.com/')
+        bitrix.srh = MockSRH(post_callback)
+
+        SIZE = 4000
+
+        bitrix_task = create_task(bitrix.get_by_ID('abc', list(range(SIZE))))
+        restore_pool_task = create_task(bitrix.srh.restore_pool())
+
+        timeout = (SIZE / 50 - BITRIX_POOL_SIZE) / BITRIX_RPS + 1
+
+        await wait({bitrix_task, restore_pool_task}, timeout=timeout)
+
+        result = bitrix_task.result()
+        restore_pool_task.cancel()
+
+        assert len(result) == SIZE
+
+    @pytest.mark.asyncio
+    async def test_limit_request_velocity(self):
+
+        async def mock_request(srh: ServerRequestHandler):
+            async with srh.limit_request_velocity():
+                print(len(srh.rr), min(srh.rr), max(srh.rr),
+                      max(srh.rr) - min(srh.rr))
+
+        srh = MockSRH(None)
+        tasks = set()
+
+        SIZE = 70
+
+        for _ in range(SIZE):
+            tasks |= {ensure_future(mock_request(srh))}
+
+        timeout = (SIZE - BITRIX_POOL_SIZE) / BITRIX_RPS + 1
+
+        start = monotonic()
+
+        await wait(tasks, timeout=timeout)
+
+        elapsed = monotonic() - start
+        print(elapsed)
+
+        assert timeout - 1 <= elapsed < timeout