Python SDK

В репозитории продукта есть серверная библиотека для Python. Она не опубликована, поэтому добавить её в зависимости проекта нельзя. Эта страница показывает её настоящий публичный API — чтобы вы понимали, что вас ждёт, — и называет рабочий путь для бэкенда сегодня.

Версия в репозитории — 0.1.0, она общая для всех трёх серверных библиотек (Python, Go, Node.js). Клиент синхронный, требует Python 3.9 и выше, внешних зависимостей нет — только стандартная библиотека. Пакет размечен типами.

Имя дистрибутива и имя импорта различаются: дистрибутив называется actionpulse-server, а импортируется как actionpulse.

Чем SDK отличается от прямого HTTP

Прямой HTTP — это один запрос и полный контроль. Библиотека добавляет к нему то, что иначе пришлось бы написать самому:

Что делает библиотека	Что это значит на практике
Проверяет контракт до отправки	неверное имя события поднимает ValidationError, а не приходит в rejected
Держит ограниченную очередь в памяти	всплеск событий не превращается в лавину запросов
Сама режет батчи	соблюдает 500 элементов и 256 КБ на запрос без вашего участия
Повторяет только то, что стоит повторять	ошибка транспорта, 429 и 5xx, с учётом Retry-After
Пересчитывает индексы отказов	rejected[i].i указывает на позицию в вашем исходном вызове, а не в очередном HTTP-запросе
Санирует исключения	текст не содержит ключ, тело запроса, идентификаторы и свойства

Чего библиотека не делает: она не создаёт фоновых потоков и ничего не отправляет сама. Пока вы не вызвали flush() или close(), события лежат в памяти процесса.

Клиент

import os
from actionpulse import ActionPulse

with ActionPulse(
    endpoint="https://YOUR_ACTIONPULSE_HOST",
    write_key=os.environ["ACTIONPULSE_SERVER_KEY"],
    app_version="billing-2026.07.28",
    batch_size=100,
) as pp:
    pp.track(
        {
            "event": "order_paid",
            "user_id": "YOUR_USER_ID",
            "schema_version": 1,
            "props": {"order_id": "A-1024", "amount_minor": 99000},
        }
    )
    result = pp.flush()
    for item in result.rejected:
        print("event rejected", item.i, item.code)

Все аргументы конструктора — только именованные: позиционный вызов не работает. Конструктор проверяет аргументы и не делает сетевых запросов: неверный адрес или пустой ключ поднимает ValidationError сразу, а не при первой отправке.

Блок with вызывает close() на выходе — это самый надёжный способ не потерять очередь.

Аргументы конструктора

Аргумент	По умолчанию	Что важно знать
endpoint	—	обязательно, абсолютный http/https адрес приёма
write_key	—	обязательно, серверный ключ pp_sk_…, без переводов строки
timeout	5.0	таймаут одной попытки, в секундах
queue_capacity	1000	предел очереди в памяти
batch_size	100	элементов в одном запросе, от 1 до 500
max_retries	3	целое от 0 до 10; 0 означает «без повторов»
base_backoff	0.1	начальная задержка, в секундах
max_backoff	2.0	потолок задержки в секундах, не меньше base_backoff
app_version	None	до 128 печатных ASCII-байт, уходит в заголовок диагностики
release	None	то же, для идентификатора релиза
processor	None	последний фильтр перед очередью
logger	None	вызывается на каждой повторной попытке
sleeper	time.sleep	своя реализация ожидания для тестов

Аргументы source и schema_version объявлены устаревшими. source игнорируется: источник события выводится из типа ключа, а не из того, что указал вызывающий. schema_version в конструкторе работает как значение по умолчанию для одноимённого поля события.

Обратите внимание на единицы: в Python задержки и таймаут — в секундах (0.1, 2.0, 5.0), в Node.js те же значения задаются в миллисекундах.

processor получает событие как неизменяемое отображение. Верните новое отображение или None, чтобы событие выбросить. event_id и time восстанавливаются после вызова, поэтому обработчик не может нарушить идемпотентность. Любое исключение внутри превращается в ValidationError. Обработчик должен быть безопасен для вызова из нескольких потоков.

Методы

Метод	Подпись	Поведение
queue_size	int (свойство)	длина очереди
track	(Mapping) -> str	ставит одно событие в очередь, возвращает его event_id
batch	(Iterable[Mapping]) -> List[str]	проверяет все события и ставит их все или ни одного
group	(Mapping) -> str	ставит в очередь обычное событие group_identify
identify	(Mapping) -> Mapping	отправляет сразу, без очереди и без повторов
alias	(Mapping) -> Mapping	отправляет сразу; поле traits игнорируется
flush	() -> TrackResponse	отправляет накопленное
close	() -> TrackResponse	отправляет накопленное и закрывает клиент

batch принимает именно последовательность событий. Если передать в него одно событие-отображение, будет ValidationError: это защита от опечатки, которая иначе разошлась бы по ключам словаря.

identify и alias возвращают разобранное тело ответа. У этих операций нет поля идемпотентности, поэтому они никогда не повторяются автоматически — решение о повторе принимаете вы.

Клиент потокобезопасен: постановка в очередь и отправка защищены отдельными блокировками, поэтому один экземпляр можно держать на процесс и вызывать из любого потока.

Результат отправки

TrackResponse — неизменяемый датакласс с полями accepted, n и rejected; rejected — кортеж элементов Rejected(i, code).

result = pp.flush()
if result.rejected:
    for item in result.rejected:
        print(f"rejected {item.i}: {item.code}")

Событие удаляется из очереди после успешного ответа, даже если оно отклонено по элементу: повтор такого события не поможет, исправлять надо разметку.

Исключения

Класс	Когда	Что делать
ValidationError	локальная проверка контракта	исправить вызов
QueueFullError	очередь заполнена	увеличить queue_capacity или вызывать flush чаще
TransportError	сеть или таймаут, повторы исчерпаны	повторить позже
APIError	приём ответил не 202	смотреть status и code

Все четыре наследуют ActionPulseError:

from actionpulse import APIError, ActionPulseError

try:
    pp.flush()
except APIError as error:
    if error.status == 401:
        print("actionpulse: credential rejected")
except ActionPulseError:
    print("actionpulse: delivery failed")

APIError содержит только operation, status и канонический code из тела ответа. Тела ответа, ключа и свойств события в тексте исключения нет намеренно: такое исключение безопасно записать в журнал.

logger получает неизменяемое отображение с ключами operation, attempt и status — и ничего больше. По умолчанию журналирование выключено.

Завершение процесса

Автоматической отправки при выходе нет. Незакрытый клиент теряет всё, что осталось в очереди.

Если клиент живёт дольше одного блока with — например, лежит в модуле приложения, — закрывайте его явно на остановке:

import atexit

pp = ActionPulse(
    endpoint="https://YOUR_ACTIONPULSE_HOST",
    write_key=os.environ["ACTIONPULSE_SERVER_KEY"],
)

def shutdown() -> None:
    try:
        pp.close()
    except Exception:
        pass

atexit.register(shutdown)

Важные детали:

• close() сначала отправляет очередь, и только потом перестаёт принимать вызовы.
• Если отправка не удалась, клиент остаётся открытым: можно починить связь и повторить close().
• Повторный close() на уже закрытом клиенте возвращает пустой результат без исключения; вызов во время закрытия поднимает ошибку.
• В веб-приложениях за менеджером процессов (Gunicorn, uWSGI) регистрируйте закрытие на событии остановки воркера, а не только в atexit.

Рабочий путь сегодня

Пока библиотека не опубликована, отправляйте события прямым HTTP: это тот же контракт, те же лимиты и те же коды отказов. Готовый пример на Python есть в отправке событий с бэкенда — там же перечислены поля события, коды отказов и правила повторов.

Если вы захотите позже перейти на библиотеку, полезно заранее сделать две вещи: складывать event_id вместе с бизнес-фактом и держать отправку за одной функцией в своём коде. Тогда замена транспорта не затронет вызовы.

Дальше: лимиты и квоты и доставка, повторы и дубли.