rapid_bitrix_connector

Python interface library engineered for accelerated interaction with the Bitrix24 platform via its external web services interface.

• Core Capabilities
• Getting Started
• Usage Illustrations
• Operational Mechanics
• Optimization Tips
• Author Contact
• Comprehensive API Reference

Core Capabilities

Superior Data Exchange Throughput

• Data throughput rates can reach thousands of records per second when dealing with extensive datasets.
• Automatic request aggregation into operational batches significantly reduces the total count of necessary server interactions, boosting speed.
• Batches are dispatched to the server concurrently, not sequentially.
• Sophisticated pagination handling strategies yield massive acceleration in data extraction (refer to performance benchmarks).

Server Resilience Assurance

• Adherence to all established Bitrix service velocity constraints (see Bitrix24 Official Request Rate Policy).
• Dynamic throttling mechanism: If server returns errors, transmission speed is automatically attenuated.
• For intricate operations that trigger server retries, velocity can be manually dialed down using a single configuration line.

Developer Experience Enhancements

• High-level abstraction methods minimize boilerplate code; most operations require only a single line. Parallelism, batching, and other overhead are managed internally.
• Allows specification of request arguments precisely as documented in the Bitrix24 REST API specification. Input validation is performed to aid debugging.
• API calls are automatically augmented with a progress bar utilizing the tqdm library, visualizing processed item count alongside elapsed and remaining execution time.
• OAuth Authentication Support: When interacting with applications, token expiry is monitored, and the token is automatically refreshed via a user-supplied callback function.

Synchronous and Asynchronous Clients

• The availability of an asynchronous client implementation makes the library suitable for event-driven architectures (e.g., real-time messaging bots).

Adoption Footprint

• Currently employed by Yandex

Getting Started

Install the package via the Python package installer: shell pip install rapid-bitrix-connector

Then, within your Python script:

python from rapid_bitrix_connector import Bitrix

Substitute with your actual domain webhook endpoint

webhook_url = "https://your_domain.bitrix24.com/rest/1/your_secret_key/" bx_client = Bitrix(webhook_url)

Subsequent interactions with the Bitrix24 server are facilitated through methods invoked on the bx_client object.

Usage Illustrations

fetch_all_records()

To retrieve the complete set of entities matching a list query, utilize the fetch_all_records() function:

python

Retrieve all leads

all_leads = bx_client.fetch_all_records('crm.lead.list')

fetch_all_records() outputs a standard Python list, where each element is a dictionary representing a single retrieved entity.

You can customize the extraction by supplying the optional params argument:

python

Fetch open deals, including all custom user fields

deals_in_progress = bx_client.fetch_all_records( 'crm.deal.list', params={ 'select': ['', 'UF_'], 'filter': {'CLOSED': 'N'} })

retrieve_by_identifier_set()

If you possess a collection of entity identifiers, fetch their attributes using retrieve_by_identifier_set() by invoking relevant *.get style API calls:

python ''' Group contacts associated with each deal into a dictionary structure: { Deal_ID_1: [contact_entity_A, contact_entity_B, ...], Deal_ID_2: [contact_entity_C, ...], ... } '''

associated_contacts = bx_client.retrieve_by_identifier_set( 'crm.deal.contact.items.get', [d['ID'] for d in deals_in_progress])

retrieve_by_identifier_set() returns a mapping structure: Identifier: ServerResponsePayload, where the payload pertains to that specific identifier query.

execute_command()

To perform bulk creation, modification, or deletion operations, use the execute_command() method:

python

Prepend the deal ID to the title of every deal

tasks_for_update = [ { 'ID': d['ID'], 'fields': { 'TITLE': f'{d["ID"]} - {d["TITLE"]}' } } for d in deals_in_progress ]

update_responses = bx_client.execute_command('crm.deal.update', tasks_for_update)

execute_command() yields a list containing the server's response for each item processed from the input collection.

execute_command(bypass_optimization=True)

Invoking execute_command with the flag bypass_optimization=True forwards the provided arguments directly to the server endpoints (skipping batch serialization logic), returning the raw server response without internal processing.

This mode is useful for debugging, but also mandatory for requests that: - Contain None values in parameters (as serialization logic might inadvertently strip None, which is sometimes used to clear field values). - Target deprecated Bitrix methods expecting a list as input (reference issue #157).

python

Clear the DESCRIPTION field for lead ID 123

update_payload = {"ID": 123, "fields": {"DESCRIPTION": None}} bx_client.execute_command('crm.lead.update', update_payload, bypass_optimization=True)

Post a comment to a specific task

bx_client.execute_command( 'task.commentitem.add', [123, {"POST_MESSAGE": "New comment content"}], bypass_optimization=True )

invoke_batch_operations()

For invoking methods that support the native Bitrix batch call format, utilize invoke_batch_operations():

python multi_results = bx_client.invoke_batch_operations ({ 'halt': 0, 'cmd': { 'deals_snapshot': 'crm.deal.list', # Fetch the deal list # Fetch activities linked to the first deal fetched above 'linked_activities': 'crm.activity.list?filter[ENTITY_TYPE]=3&filter[ENTITY_ID]=$result[deals_snapshot][0][ID]' } })

Asynchronous Invocation

If the application environment requires asynchronous I/O (e.g., within an asyncio framework or a Jupyter Notebook), instantiate the BitrixAsync client class instead of the standard Bitrix(): python from rapid_bitrix_connector import BitrixAsync bx_async_client = BitrixAsync(webhook_url)

All methods on this object mirror their synchronous counterparts but require awaiting: python retrieved_leads = await bx_async_client.fetch_all_records('crm.lead.list')

OAuth Authorization

For operations requiring OAuth security, supply the application endpoint URL to the webhook parameter during Bitrix() instantiation. Furthermore, provide an asynchronous function reference to the token_refresher parameter, which the library will invoke to obtain or refresh the access token: python from rapid_bitrix_connector import Bitrix import aiohttp

async def refresh_access_token() -> str: async with aiohttp.ClientSession() as session: oauth_endpoint = 'https://oauth.bitrix.info/oauth/token/' request_payload={ 'grant_type': 'refresh_token', 'client_id': 'YOUR_CLIENT_ID', 'client_secret': 'YOUR_CLIENT_SECRET', 'refresh_token': 'YOUR_REFRESH_TOKEN' } async with session.get(oauth_endpoint, params=request_payload, timeout=10) as response: token_data = await response.json() return token_data["access_token"]

Initialize client, providing the application endpoint and the token refresh handler

bx_oauth_client = Bitrix(webhook=APP_ENDPOINT_ADDRESS, token_refresher=refresh_access_token) ...

The token_refresher callback is executed whenever the initial token is needed or when the existing authorization token requires renewal.

Operational Mechanics

1. Pre-flight Validation: Before any server communication, all primary arguments passed to methods within the Bitrix class undergo rigorous validation. Exceptions such as TypeError and ValueError are raised immediately upon detecting input anomalies.
2. Request Generation: Instructions are formulated to fetch all intended items from the requested list endpoint.
3. Batch Aggregation: Generated requests are grouped into discrete batches, with a maximum capacity of 50 requests per batch.
4. Concurrent Dispatch: These batches are then transmitted to the server in parallel streams, with request velocity actively managed (details below).
5. Result Aggregation: The successful results (content of the result field) are flattened into a single, unified list and returned to the user.
   • Exceptions of type aiohttp.ClientError are surfaced if the Bitrix server returns an HTTP failure status. A RuntimeError is raised if the HTTP status is 200 but an operational error is embedded within the response body.
   • Results are reordered (except for fetch_all_records()) to ensure the sequence of returned elements matches the sequence of their originating requests.

For the fetch_all_records() routine, Step 2 is handled differently: - fetch_all_records() initiates a singular query to Bitrix24 using the specified method and parameters. - The server responds with the initial page set (up to 50 items) and a total count—the absolute number of entities matching the query criteria. - Based on this total, subsequent page-fetching requests (approximately total // 50 - 1 requests) are programmatically constructed to retrieve the remaining entities.

Since processing lengthy lists via fetch_all_records() can span considerable time—during which new items might be added to the source list—the final count may diverge from the initial total. In such cases, a standard Python warning notification is issued.

How rapid_bitrix_connector Manages Server Load

The library strictly respects Bitrix24's official rate constraints (see section below). Crucially, it also initiates speed reduction if the server indicates stress by returning errors (autothrottling). This adaptive approach can significantly boost data acquisition speed (refer to velocity testing suite).

Official Bitrix24 Request Rate Policy

Two primary constraints are actively enforced: 1. The Leaky Bucket methodology: https://dev.1c-bitrix.ru/learning/course/index.php?COURSE_ID=93&LESSON_ID=7885 2. The Sliding Window mechanism applied per API method: https://helpdesk.bitrix24.ru/open/15959788

Both policies are faithfully observed by this library.

Optimization Tips

Q: Does your package handle [Specific Task X]?

Consult the API Reference Guide. If the answer remains elusive, contact the originator.

Q: How do I configure the Webhook?

Refer to the official REST API documentation 1. Navigate to the 'Incoming Webhook' setup section and follow instructions:
Set up an incoming webhook via Developer Tools (Applications > Developers, 'Ready Scenarios' tab > Other > Incoming Webhook). 2. Assign a descriptive name to the webhook. 3. The generated URL string containing the token is your required webhook endpoint. 4. Ensure correct access permissions are configured (omitting permissions results in an authorization failure).

• The 'Request Generator' block is an auxiliary utility for method discovery and inspecting expected server responses. It does not affect webhook setup.
• Additional foundational information can be found in the help article

Q: How should I structure an API call to achieve [Specific Action Y]?

1. Search within the official REST API reference.
2. If the documentation provides no resolution, pose your question to the Bitrix24 Developer Community REST API group.
3. Inquire in the Bitrix24 Developers Telegram channel.
4. Ask in the rapid_bitrix_connector user Telegram group.
5. Post on Russian StackOverflow.

Q: How can I inspect the payload being sent and the response received?

Enable request and response logging: python import logging

logging.getLogger('rapid_bitrix_connector').addHandler(logging.StreamHandler())

Q: I encounter server errors under heavy load. What is the remedy?

Option 1

Wrap your Bitrix interactions within the throttle_traffic context manager:

python with bx_client.throttle_traffic(max_concurrent_calls=5): # Reduced concurrency results = bx_client.execute_command('crm.lead.add', tasks)

See details regarding the throttle_traffic manager.

Option 2

During client initialization, adjust the parameters governing server load:

python bx_client = Bitrix( webhook_url, connection_pool_size = 20, # Default: 50 server_rps_limit = 1.0, # Default: 2.0 requests per second batch_payload_limit = 20, # Default: 50 items per batch operation_timeout_seconds = 100, # Default: 480 )

Q: I only need to execute execute_command() once, not for a list of items.

The execute_command() interface is versatile enough to handle both iterative list processing and single-shot command execution. Simply pass the singular request parameters:

python endpoint = 'crm.lead.add' single_params = {'fields': {'TITLE': 'One-Off Lead'}} # Single parameter set bx_client.execute_command(endpoint, single_params)

The return value will be the server's response corresponding to that single API invocation.

However, if these single calls are frequent, aggregating them into a list and calling execute_command() once on the entire collection remains the superior performance choice.

Q: How are results ordered when using fetch_all_records()?

Currently, the results are not intrinsically ordered.

All server communications occur asynchronously, meaning the order of the resulting list reflects the sequence in which the server completed processing each request. If chronological or ID-based ordering is mandatory, it must be implemented by the caller, for example:

python deals_data = bx_client.fetch_all_records('crm.deal.list') deals_data.sort(key = lambda d: int(d['ID']))

Q: Fetching all fields for all list items via fetch_all_records() is excessively slow. How can I expedite this?

The primary mechanism for acceleration, while respecting server constraints, involves minimizing the data volume transferred and ensuring the download rate stays within Bitrix's imposed limits. This can be achieved through optimized data caching strategies and reducing the scope of fields requested.

Violating the Bitrix request policy leads to throttling penalties; thus, adherence is strongly recommended (controlled by the respect_velocity_policy=True setting).

Q: I am receiving an SSL certificate verification error. What action should I take?

If you encounter an SSLCertVerificationError or CERTIFICATE_VERIFY_FAILED, you can bypass the SSL certificate validation process: python bx_async_client = BitrixAsync(webhook_url, ssl_verify=False)

I am executing this library within notebooks or the Spyder IDE and encountering runtime errors.

Your execution environment often manages the asyncio event loop internally (see related discussion).

Switch to utilizing the asynchronous client implementation. Instead of:

python from rapid_bitrix_connector import Bitrix bx = Bitrix(webhook_url) leads = bx.fetch_all_records('crm.lead.list')

Employ the asynchronous client pattern:

python from rapid_bitrix_connector import BitrixAsync bx = BitrixAsync(webhook_url) leads = await bx.fetch_all_records('crm.lead.list')

I am utilizing an Enterprise Bitrix instance. How can I configure higher request throughput?

Specify elevated values for connection_pool_size and server_rps_limit in the constructor: python from rapid_bitrix_connector import Bitrix bx = Bitrix(webhook_url, connection_pool_size=250, server_rps_limit=5.0)

Author Contact

• Telegram channel for discussions: https://t.me/rapid_bitrix_connector
• To report a bug or suggest a feature: https://github.com/new_user/rapid_bitrix_connector/issues/new