Configuration

Client Configuration

from cb_events import EventClient, ClientConfig

config = ClientConfig(
    timeout=10,  # Server long-poll timeout (seconds)
    strict_validation=False,  # Raise on invalid events vs. skip
    retry_attempts=8,  # Total attempts (initial + retries)
    retry_backoff=1.0,  # Initial backoff (seconds)
    retry_factor=2.0,  # Backoff multiplier
    retry_max_delay=30.0,  # Max retry delay (seconds)
)

client = EventClient(events_url, config=config)

Timeout Settings

The timeout parameter controls the maximum time (in seconds) the Chaturbate server will wait before sending back a response.

config = ClientConfig(timeout=5)  # More frequent polls
config = ClientConfig(timeout=30)  # Server holds connection longer

Default: 10 seconds

Retry Configuration

config = ClientConfig(
    retry_attempts=5,  # Try 5 times total
    retry_backoff=2.0,  # Start with 2s delay
    retry_factor=1.5,  # Increase by 1.5x each retry
    retry_max_delay=60.0,  # Cap delays at 60s
)

Retries on 429, 5xx, Cloudflare 521-524. Never retries 401/403.

Validation Mode

strict_validation=False (default): skips invalid events and logs a warning.

strict_validation=True: raises pydantic.ValidationError on invalid event data.

Environment Selection

Pass the upstream URL directly to EventClient. Hostname determines production vs testbed automatically.

prod_url = "https://eventsapi.chaturbate.com/events/username/token/"
testbed_url = "https://events.testbed.cb.dev/events/username/token/"

prod_client = EventClient(prod_url)
testbed_client = EventClient(testbed_url)

Rate Limiting

Default: 2000 requests per 60 seconds per client.

Custom Rate Limiter

from aiolimiter import AsyncLimiter

limiter = AsyncLimiter(max_rate=1000, time_period=60)
client = EventClient(events_url, rate_limiter=limiter)

Shared Rate Limiter

Warning

Each client gets its own independent budget by default. Multiple clients without a shared limiter multiply the effective request rate.

limiter = AsyncLimiter(max_rate=2000, time_period=60)

client1 = EventClient(url1, rate_limiter=limiter)
client2 = EventClient(url2, rate_limiter=limiter)
client3 = EventClient(url3, rate_limiter=limiter)

Logging

Set the logger to DEBUG for verbose polling URLs and event dispatch data:

import logging

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

Example DEBUG Output:

DEBUG:cb_events.client:Polling https://eventsapi.chaturbate.com/events/user/************************/?timeout=10
DEBUG:cb_events.client:Received 1 events for user
DEBUG:cb_events.router:Dispatching chatMessage event 1775683684418-0 to 2 handlers