Runnev
RU Sign in
Docs / SDKs

Python SDK

The runnev package is the official Python SDK. Standard library only, no runtime dependencies, Python 3.9+. Publishing is synchronous; subscribing is a generator you can iterate or run on a thread.

Install

bash
pip install runnev   # 1.5.3

Configure

python
import os
from runnev import Runnev

runnev = Runnev(
    api_key=os.environ["RUNNEV_API_KEY"],   # required
    base_url="https://runnev.dev/v1",        # default
    timeout=30.0,                            # per-request seconds
    max_retries=4,                           # retries on 429 / 5xx
)

Publish

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

# explicit sequence - idempotent, safe to retry
res = runnev.publish(stream.id, 41823, [
    {"type": "order.paid", "id": "o_5521", "amount": 1999},
])
print(res.cursor, res.duplicate)   # 41823 False

# server-assigned sequence
runnev.publish_auto(stream.id, [{"type": "tick"}])

# raw bytes for a raw-mode stream
runnev.publish_raw(stream.id, 9001, protobuf_bytes)

Subscribe

subscribe yields Batch objects and reconnects automatically, resuming from the last sequence.

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

To run it in the background and stop it cleanly from another thread, use the context manager form:

python
with runnev.subscribe(stream.id, cursor="head") as sub:
    for batch in sub:
        handle(batch)
        if should_stop():
            sub.close()   # breaks the loop promptly

Resume across restarts

python
last = load_cursor()   # -1 if none
for batch in runnev.subscribe(stream.id, cursor=last):
    if batch.seq <= last:      # at-least-once dedup
        continue
    handle(batch)
    last = batch.seq
    save_cursor(last)

Error handling

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                       # not retryable
except RunnevError as e:
    print(e.code, e.status, e.request_id, e.doc_url)

Exceptions: RunnevError (base), AuthenticationError, InvalidRequestError, RateLimitError, ApiError. Each exposes code, status, request_id, and doc_url.

API surface

MethodReturns
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 (iterable of Batch)

Runnable examples ship in examples/: publish.py, subscribe.py, and raw_msgpack.py.