API wrapper для быстрого получения данных от Битрикс24 через REST API. Параллельные запросы к серверу, упаковка запросов в батчи, контроль скорости запросов, есть синхронный и асинхронный клиенты.
API wrapper для Питона для быстрого получения данных от Битрикс24 через REST API.
tqdm, иллюстрирующим не только количество обработанных элементов, но и прошедшее и оставшееся время выполнения запроса.Установите модуль через pip:
pip install fast-bitrix24
Далее в python:
from fast_bitrix24 import Bitrix
# замените на ваш вебхук для доступа к Bitrix24
webhook = "https://your_domain.bitrix24.ru/rest/1/your_code/"
bx = Bitrix(webhook)
Методы полученного объекта bx в дальнейшем используются для взаимодействия с сервером Битрикс24.
get_all()Чтобы получить полностью список сущностей, используйте метод get_all():
# список лидов
leads = bx.get_all('crm.lead.list')
Метод get_all() возвращает список, где каждый элемент списка является словарем, описывающим одну сущность из запрошенного списка.
Вы также можете использовать параметр params, чтобы кастомизировать запрос:
# список сделок в работе, включая пользовательские поля
deals = bx.get_all(
'crm.deal.list',
params={
'select': ['*', 'UF_*'],
'filter': {'CLOSED': 'N'}
})
get_by_ID()Если у вас есть список ID сущностей, то вы можете получить их свойства при помощи метода get_by_ID()
и использовании методов вида *.get:
'''
получим список всех контактов, привязанных к сделкам, в виде
{
ID_сделки_1: [контакт_1, контакт_2, ...],
ID_сделки_2: [контакт_1, контакт_2, ...],
...
}
'''
contacts = bx.get_by_ID(
'crm.deal.contact.items.get',
[d['ID'] for d in deals])
Метод get_by_ID() возвращает словарь с элементами вида ID: result, где result - ответ сервера относительно этого ID.
call()Чтобы создавать, изменять или удалять список сущностей, используйте метод call():
# вставим в начало названия всех сделок их ID
tasks = [
{
'ID': d['ID'],
'fields': {
'TITLE': f'{d["ID"]} - {d["TITLE"]}'
}
}
for d in deals
]
bx.call('crm.deal.update', tasks)
Метод call() возвращает список ответов сервера по каждому элементу переданного списка.
Вызов call с парамтером raw=True отправляет на сервер переданные ему параметры в оригинальном, необработанном виде (пропуская упаковку в батчи), и возвращает ответ сервера без какой-либо обработки.
Подобный вызов можно использовать в отладочных целях, но кроме того, придется его использовать для отправки запросов, которые:
None (None применяется для стирания значения полей, а упаковка в батчи мешает передавать None),# стереть DESCRIPTION в лиде 123
params = {"ID": 123, "fields": {"DESCRIPTION": None}}
bx.call('crm.lead.update', params, raw=True)
# добавить комментарий к задаче
bx.call(
'task.commentitem.add',
[123, {"POST_MESSAGE": "Комментарий к задаче"}],
raw=True
)
call_batch()Если вы хотите вызвать пакетный метод, используйте call_batch():
results = bx.call_batch ({
'halt': 0,
'cmd': {
'deals': 'crm.deal.list', # берем список сделок
# и берем список дел по первой из них
'activities': 'crm.activity.list?filter[ENTITY_TYPE]=3&filter[ENTITY_ID]=$result[deals][0][ID]'
}
})
Если требуется использование бибилиотеки в асинхронном коде (или в ноутбуках), то вместо клиента Bitrix() создавайте клиент класса BitrixAsync():
from fast_bitrix24 import BitrixAsync
bx = BitrixAsync(webhook)
Все методы у него - синхронные аналоги методов из Bitrix(), описанных выше:
leads = await bx.get_all('crm.lead.list')
Если требуется авторизация через OAuth, то при инициализации клиента Bitrix() необходимо передать в параметре webhook ссылку на эндпойнт приложения, а в параметре token_func - ссылку на асинхронную функцию, которая будет возвращать токен авторизации:
from fast_bitrix24 import Bitrix
import requests
async def get_new_token() -> str:
oauth_url = 'https://oauth.bitrix.info/oauth/token/'
params={
'grant_type': 'refresh_token',
'client_id': ...,
'client_secret': ...,
'refresh_token': ...
}
result = requests.get(oauth_url, timeout=10, params=params)
return result.json()["access_token"]
bx = Bitrix(webhook=ENDPOINT_ADDRESS, token_func=get_new_token)
...
token_func будет вызываться каждый раз, когда необходимо получить впервые или обновить токен авторизации.
Bitrix происходит проверка корректности самых популярных параметров, передаваемых к серверу, и поднимаются исключения TypeError и ValueError при наличии ошибок.result) собираются в единый плоский список и возвращаются пользователю.
aiohttp.ClientError, если сервер Битрикс вернул HTTP-ошибку, и RuntimeError, если код ответа был 200, но ошибка сдержалась в теле ответа сервера.get_all()) - порядок элементов в списке результатов совпадает с порядком соответствующих запросов в списке запросов.В случае с методом get_all() пункт 2 выше выглядит немного сложнее:
get_all() делает первый запрос к серверу Битрикс24 с указанным методом и параметрами.total - общее количество элементов, найденных по запросу.total // 50 - 1 запросов), необходимых для получения всех запрошенных элементов.В связи с тем, что выполнение get_all() по длинным спискам может занимать долгое время, в течение которого пользователи могут добавлять новые элементы в список, может возникнуть ситуация, когда общее полученное количество элементов может не соответствовать изначальному значению total. В таких случаях будет выдано стандартное питоновское предупреждение (warning).
fast_bitrix24 регулирует скорость запросовБиблиотека соблюдает официальные ограничения Битрикс24 по скорости запросов (см. ниже "Официальная политика Битрикс24 по скорости запросов"). Одновременно, она начинает снижать скорость запросов, если сервер начинает возвращать ошибки (autothrottling). Подобный подход позволяет на порядки увеличить скорость получения данных (см. тесты скорости).
Одновременно работает два ограничения:
Оба эти ограничения соблюдаются библиотекой.
Посмотрите в справочник по API. Если не нашли ответа, свяжитесь с автором.
Обращаемся к официальной документации по REST API
Создание Входящего вебхука и следуем инструкции:Создать входящий вебхук можно из раздела Разработчикам (Приложения > Разработчикам, вкладка "Готовые сценарии" > Другое > Входящий вебхук).Генератором запросов является вспомогательным и может быть использован для подбора необходимых методов и посмотреть какой результат они будут возвращать. На настройку самого вебхука ни как не влияет.Включите логирование запросов и ответов сервера.
import logging
logging.getLogger('fast_bitrix24').addHandler(logging.StreamHandler())
Оберните обращения к Битриксу в slow:
with bx.slow():
results = bx.call('crm.lead.add', tasks)
См. подробнее о slow.
При инициализации клиента поэкспериментируйте со значениями параметров, регулирующих нагрузку на сервер:
bx = Bitrix(
webhook,
request_pool_size = 20, # по умолчанию - 50
requests_per_second = 1.0, # по умолчанию - 2.0
batch_size = 20, # по умолчанию - 50
operating_time_limit = 100, # по умолчанию - 480
)
call() только один раз, а не по списку.Передавайте параметры запроса методу call(), он может делать как запросы по списку, так и единичный запрос:
method = 'crm.lead.add'
params = {'fields': {'TITLE': 'Чпок'}}
bx.call(method, params)
Результатом будет ответ сервера по этому одному элементу.
Однако, если такие вызовы делаются несколько раз, то более эффективно формировать из них список и вызывать call() единожды по всему списку.
get_all()?Пока что никак.
Все обращения к серверу происходят асинхронно и список результатов отсортирован в том порядке, в котором сервер возвращал ответы. Если вам требуется сортировка, то вам нужно делать ее самостоятельно, например:
deals = bx.get_all('crm.deal.list')
deals.sort(key = lambda d: int(d['ID']))
get_all() для получения всех полей всех элементов списка, но это происходит слишком долго. Как ускорить этот процесс?Сейчас, кода Битрикс ограничивает скорость запросов к серверу, ключевым методом ускорения остается сокращение количества информации, которую вы загружаете с сервера, и сохранение скорости скачивания в пределах, установленных Битриксом. Это может быть достигнуто за счет кэширования и сокращения количества скачиваемых полей.
Нарушение политики Битриска по скорости запросов влечет за собой штрафы, поэтому мы рекомендуем соблюдать её (параметр respect_velocity_policy=True).
Если вы получаете SSLCertVerificationError / CERTIFICATE_VERIFY_FAILED, попробуйте отключить верификацию сертификата SSL:
bx = BitrixAsync(webhook, ssl=False)
Ваша cреда выполнения самостоятельно управляет примитивами asyncio (см. больше).
Используйте асинхронный клиент. То есть, вместо кода:
from fast_bitrix24 import Bitrix
bx = Bitrix(webhook)
leads = bx.get_all('crm.lead.list')
используйте код:
from fast_bitrix24 import BitrixAsync
bx = BitrixAsync(webhook)
leads = await bx.get_all('crm.lead.list')
В конструкторе указывайте параметры request_pool_size=250 и requests_per_second=5:
from fast_bitrix24 import Bitrix
bx = Bitrix(webhook, request_pool_size=250, requests_per_second=5)
FAQs
API wrapper для быстрого получения данных от Битрикс24 через REST API. Параллельные запросы к серверу, упаковка запросов в батчи, контроль скорости запросов, есть синхронный и асинхронный клиенты.
We found that fast-bitrix24 demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
PyPI confirms no security flaws were exploited in the Ultralytics supply chain attack and highlights improvements for safer package publishing.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.