cb-events 4.7.1

Async Python client for the Chaturbate Events API.

CB Events

Async Python client for the Chaturbate Events API.

Installation

uv pip install cb-events

Quick Start

import asyncio
from cb_events import EventClient, EventRouter, EventType, Event

router = EventRouter()

@router.on(EventType.TIP)
async def handle_tip(event: Event) -> None:
    if event.user and event.tip:
        print(f"{event.user.username} tipped {event.tip.tokens} tokens")

async def main():
    async with EventClient(username, token) as client:
        async for event in client:
            await router.dispatch(event)

asyncio.run(main())

Event Types

TIP · FANCLUB_JOIN · MEDIA_PURCHASE · CHAT_MESSAGE · PRIVATE_MESSAGE · USER_ENTER · USER_LEAVE · FOLLOW · UNFOLLOW · BROADCAST_START · BROADCAST_STOP · ROOM_SUBJECT_CHANGE

Configuration

from cb_events import EventClientConfig

config = EventClientConfig(
    timeout=10,              # Request timeout (seconds)
    use_testbed=False,       # Use testbed endpoint
    retry_attempts=8,        # Max retry attempts
    retry_backoff=1.0,       # Initial backoff (seconds)
    retry_factor=2.0,        # Backoff multiplier
    retry_max_delay=30.0,    # Max retry delay (seconds)
)

client = EventClient(username, token, config=config)

Note: Config is immutable after creation. config parameter must be passed as keyword argument.

Rate Limiting

Default: 2000 requests/60s per client. Share rate limiter across clients:

from aiolimiter import AsyncLimiter

limiter = AsyncLimiter(max_rate=2000, time_period=60)
client1 = EventClient(username1, token1, rate_limiter=limiter)
client2 = EventClient(username2, token2, rate_limiter=limiter)

Event Properties

Properties return None on incompatible event types:

event.user          # User object (most events)
event.tip           # Tip object (TIP only)
event.message       # Message object (CHAT_MESSAGE, PRIVATE_MESSAGE)
event.room_subject  # RoomSubject object (ROOM_SUBJECT_CHANGE)
event.broadcaster   # Broadcaster username string

Error Handling

from cb_events import AuthError, EventsError

try:
    async with EventClient(username, token) as client:
        async for event in client:
            await router.dispatch(event)
except AuthError:
    # Authentication failed (401/403)
    pass
except EventsError as e:
    # API/network errors - e.status_code, e.response_text
    pass

Retries: Automatic on 429, 5xx, Cloudflare errors. No retry on auth errors.

Handlers: Execute sequentially. If a handler raises an exception, it propagates immediately and stops subsequent handlers.

Logging

import logging

## Error Handling

```python
from cb_events import AuthError, EventsError

try:
    async with EventClient(username, token) as client:
        async for event in client:
            await router.dispatch(event)
except AuthError:
    # Authentication failed (401/403)
    pass
except EventsError as e:
    # API/network errors - e.status_code, e.response_text
    pass

Retries: Automatic on 429, 5xx, Cloudflare errors. No retry on auth errors.

Handlers: Execute sequentially. If a handler raises an exception, it propagates immediately and stops subsequent handlers.

Logging

import logging

logging.getLogger('cb_events').setLevel(logging.DEBUG)

Requirements

Python ≥3.12 - Dependencies

License

MIT - See LICENSE

---

Not affiliated with Chaturbate.