Runnev
EN Войти
Документация / SDK

Python SDK

Пакет runnev — официальный Python-SDK. Только стандартная библиотека, без зависимостей во время выполнения, Python 3.9+. Публикация синхронна; подписка — это генератор, который можно итерировать или запускать в потоке.

Установка

bash
pip install runnev   # 1.5.3

Настройка

python
import os
from runnev import Runnev

runnev = Runnev(
    api_key=os.environ["RUNNEV_API_KEY"],   # обязательно
    base_url="https://runnev.dev/v1",        # по умолчанию
    timeout=30.0,                            # секунды на запрос
    max_retries=4,                           # повторы при 429 / 5xx
)

Публикация

python
stream = runnev.create_stream(name="orders-eu", retention_seconds=86400)

# явный номер — идемпотентно, безопасно повторять
res = runnev.publish(stream.id, 41823, [
    {"type": "order.paid", "id": "o_5521", "amount": 1999},
])
print(res.cursor, res.duplicate)   # 41823 False

# серверный номер
runnev.publish_auto(stream.id, [{"type": "tick"}])

# сырые байты для потока в режиме raw
runnev.publish_raw(stream.id, 9001, protobuf_bytes)

Подписка

subscribe отдаёт объекты Batch и переподключается автоматически, возобновляя с последнего номера.

python
for batch in runnev.subscribe(stream.id, cursor=0):
    print(batch.seq, batch.events)

Чтобы запустить её в фоне и корректно остановить из другого потока, используйте форму контекст-менеджера:

python
with runnev.subscribe(stream.id, cursor="head") as sub:
    for batch in sub:
        handle(batch)
        if should_stop():
            sub.close()   # быстро прерывает цикл

Возобновление между перезапусками

python
last = load_cursor()   # -1, если нет
for batch in runnev.subscribe(stream.id, cursor=last):
    if batch.seq <= last:      # дедуп «не менее одного раза»
        continue
    handle(batch)
    last = batch.seq
    save_cursor(last)

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

python
from runnev import RunnevError, RateLimitError, AuthenticationError

try:
    runnev.publish(stream_id, seq, events)
except RateLimitError as e:
    time.sleep(e.retry_after or 1)
except AuthenticationError:
    raise                       # не повторяемо
except RunnevError as e:
    print(e.code, e.status, e.request_id, e.doc_url)

Исключения: RunnevError (базовое), AuthenticationError, InvalidRequestError, RateLimitError, ApiError. Каждое отдаёт code, status, request_id и doc_url.

Поверхность API

МетодВозвращает
create_stream(name, retention_seconds=None, max_bytes=None, mode="json", id=None)Stream
list_streams(limit=20, cursor=None)StreamPage
get_stream(id)Stream
delete_stream(id)None
publish(id, seq, events)PublishResult
publish_raw(id, seq, data)PublishResult
publish_auto(id, events)PublishResult
get_cursor(id)Cursor
get_batch(id, seq)Batch
subscribe(id, cursor=None)Subscription (итерируемо по Batch)

Запускаемые примеры лежат в examples/: publish.py, subscribe.py и raw_msgpack.py.