Добавить в корзинуПозвонить
Найти в Дзене
Павлин Шарит

Как должна быть устроена качественная Python библиотека

Как должна быть устроена качественная Python библиотека Недавно изучал исходники fast_bitrix24 для работы с api bitrix и был приятно удивлен — это пример того, как нужно писать серьезные библиотеки Сначала привлекла практическая польза — библиотека решает проблему работы с Битрикс24 API: лимиты, батчинг, обработка ошибок. Но когда заглянул в код, нашел несколько любопытных моментов Валидация на всех уровнях: @staticmethod @icontract.ensure( lambda result: result != "batch", "Method cannot be 'batch'. Use call_batch() instead.", ) def standardized_method(method: str): return method.lower().strip() Библиотека использует icontract для валидации условий. Ошибки ловятся на этапе вызова, а не во время выполнения Элегантное решение синхронности: class Bitrix(BitrixAsync): """Клиент для неасинхронных запросов к серверу Битрикс24. Имплементируется путем обертки всех методов родителя в неасинхронные методы. """ def sync_decorator(coroutine): ft.wraps(coroutine) def sync_wrapper(*args, **k

Как должна быть устроена качественная Python библиотека

Недавно изучал исходники fast_bitrix24 для работы с api bitrix и был приятно удивлен — это пример того, как нужно писать серьезные библиотеки

Сначала привлекла практическая польза — библиотека решает проблему работы с Битрикс24 API: лимиты, батчинг, обработка ошибок. Но когда заглянул в код, нашел несколько любопытных моментов

Валидация на всех уровнях:

@staticmethod

@icontract.ensure(

lambda result: result != "batch",

"Method cannot be 'batch'. Use call_batch() instead.",

)

def standardized_method(method: str):

return method.lower().strip()

Библиотека использует icontract для валидации условий. Ошибки ловятся на этапе вызова, а не во время выполнения

Элегантное решение синхронности:

class Bitrix(BitrixAsync):

"""Клиент для неасинхронных запросов к серверу Битрикс24.

Имплементируется путем обертки всех методов родителя в неасинхронные методы.

"""

def sync_decorator(coroutine):

ft.wraps(coroutine)

def sync_wrapper(*args, **kwargs):

try:

loop = asyncio.get_event_loop()

except RuntimeError:

return asyncio.run(coroutine(*args, **kwargs))

if loop.is_running():

return loop.create_task(coroutine(*args, **kwargs))

return loop.run_until_complete(coroutine(*args, **kwargs))

return sync_wrapper

for method in dir(BitrixAsync):

if not method.startswith("__") and method != "slow":

locals()[method] = sync_decorator(getattr(BitrixAsync, method))

Один класс для async, второй наследуется и добавляет sync методы. Работает и в Jupyter, и в обычном коде, и в асинхронных приложениях

Продвинутые алгоритмы throttling:

- LeakyBucket для общего лимита запросов

- SlidingWindow для лимитов по времени выполнения методов

- Автоматическое снижение скорости при ошибках сервера

Обработка ошибок и retry:

while True:

try:

result = await self.request_attempt(method, params)

self.success()

return result

except TokenRejectedError:

await self.ensure_new_token()

except RETRIED_ERRORS as err:

self.failure(err)

Автоматический retry с экспоненциальным backoff, обновление токенов OAuth, обработка разных типов ошибок

Батчинг и параллелизм:

Запросы автоматически группируются в батчи, отправляются параллельно с учетом лимитов сервера, результаты собираются в правильном порядке

В итоге получается библиотека, которая:

- Просто используется (один импорт, понятный API)

- Надежно работает (валидация, обработка ошибок, тесты)

- Эффективно выполняется (параллелизм, батчинг, throttling)

Респект автору Антону Лещенко — показал как нужно делать качественные библиотеки для продакшена