cb_events¶

Async client for the Chaturbate Events API.

Provides automatic retries, rate limiting, and type-safe event handling.

Key Components:: EventClient: Async context manager for API polling. Router: Decorator-based event dispatcher. Event: Type-safe event model with nested data accessors. ClientConfig: Immutable configuration for client behavior.

Example

Basic usage with event routing:

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

router = Router()

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

async def main() -> None:
    async with EventClient(
        "https://eventsapi.chaturbate.com/events/username/token/"
    ) as client:
        async for event in client:
            await router.dispatch(event)

asyncio.run(main())

Submodules¶

Attributes¶

HandlerFunc

Async handler signature accepted by Router decorators.

Exceptions¶

`AuthError`	Authentication failure from the Events API.
`ClientRequestError`	HTTP 4xx request failure (excluding auth and rate limiting).
`EventsError`	Base exception for API failures with optional HTTP metadata.
`HttpStatusError`	Base error for non-authentication HTTP status code failures.
`RateLimitError`	HTTP 429 rate-limiting failure.
`ServerError`	HTTP 5xx server-side failure.

Classes¶

`ClientConfig`	Immutable configuration for EventClient behavior.
`Event`	Event from the Chaturbate Events API.
`EventClient`	Async client for polling the Chaturbate Events API.
`EventType`	Event types from the Chaturbate Events API.
`Media`	Media purchase transaction details.
`Message`	Chat or private message content.
`RoomSubject`	Room subject or title information.
`Router`	Routes events to registered async handlers.
`Tip`	Tip transaction details.
`User`	User information attached to events.

Package Contents¶

exception cb_events.AuthError(message: str, *, status_code: int | None = None, response_text: str | None = None)[source]¶

Bases: EventsError

Authentication failure from the Events API.

Raised when the API returns HTTP 401 (Unauthorized) or 403 (Forbidden), typically indicating invalid credentials or an expired token.

Also raised during client initialization if username or token is empty or contains invalid whitespace.

Example

Handling authentication errors:

try:
    async with EventClient(
        "https://eventsapi.chaturbate.com/events/user/invalid_token/"
    ) as client:
        await client.poll()
except AuthError:
    print("Invalid credentials - regenerate token")

Initialize error with message and optional HTTP details.

Parameters:

message – Human-readable description of the failure.
status_code – Optional HTTP status code returned by the API.
response_text – Optional raw response body. Truncated to 200 characters to reduce PII exposure in structured exception loggers and error-reporting tools.

exception cb_events.ClientRequestError(message: str, *, status_code: int | None = None, response_text: str | None = None)[source]¶

Bases: HttpStatusError

HTTP 4xx request failure (excluding auth and rate limiting).

Initialize error with message and optional HTTP details.

Parameters:

message – Human-readable description of the failure.
status_code – Optional HTTP status code returned by the API.
response_text – Optional raw response body. Truncated to 200 characters to reduce PII exposure in structured exception loggers and error-reporting tools.

exception cb_events.EventsError(message: str, *, status_code: int | None = None, response_text: str | None = None)[source]¶

Bases: Exception

Base exception for API failures with optional HTTP metadata.

Raised for network errors, invalid responses, rate limiting, and other non-authentication failures. Includes HTTP status code and response body when available.

Example

Inspecting error details:

try:
    events = await client.poll()
except EventsError as e:
    if e.status_code == 429:
        print("Rate limited, backing off...")
    print(f"Response: {e.response_text}")

Initialize error with message and optional HTTP details.

Parameters:

message – Human-readable description of the failure.
status_code – Optional HTTP status code returned by the API.
response_text – Optional raw response body. Truncated to 200 characters to reduce PII exposure in structured exception loggers and error-reporting tools.

response_text: str | None¶: Raw response body, truncated to 200 characters to limit PII in logs.

status_code: int | None¶: HTTP status code if available.

exception cb_events.HttpStatusError(message: str, *, status_code: int | None = None, response_text: str | None = None)[source]¶

Bases: EventsError

Base error for non-authentication HTTP status code failures.

Initialize error with message and optional HTTP details.

Parameters:

message – Human-readable description of the failure.
status_code – Optional HTTP status code returned by the API.
response_text – Optional raw response body. Truncated to 200 characters to reduce PII exposure in structured exception loggers and error-reporting tools.

exception cb_events.RateLimitError(message: str, *, status_code: int | None = None, response_text: str | None = None)[source]¶

Bases: HttpStatusError

HTTP 429 rate-limiting failure.

Only raised after all retry attempts are exhausted; the client retries 429 responses automatically per ClientConfig.retry_attempts.

Initialize error with message and optional HTTP details.

Parameters:

message – Human-readable description of the failure.
status_code – Optional HTTP status code returned by the API.
response_text – Optional raw response body. Truncated to 200 characters to reduce PII exposure in structured exception loggers and error-reporting tools.

exception cb_events.ServerError(message: str, *, status_code: int | None = None, response_text: str | None = None)[source]¶

Bases: HttpStatusError

HTTP 5xx server-side failure.

Initialize error with message and optional HTTP details.

Parameters:

message – Human-readable description of the failure.
status_code – Optional HTTP status code returned by the API.
response_text – Optional raw response body. Truncated to 200 characters to reduce PII exposure in structured exception loggers and error-reporting tools.

class cb_events.ClientConfig(/, **data: Any)[source]¶

Bases: pydantic.BaseModel

Immutable configuration for EventClient behavior.

Controls timeouts, retry logic, and validation strictness.

Example

Lenient configuration for development:

config = ClientConfig(
    strict_validation=False,
    retry_attempts=3,
)

Note

This class is immutable (frozen). Create a new instance to change configuration values.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

validate_delays() → Self[source]¶

Validate retry delay configuration.

Returns:: Validated configuration instance.
Return type:: Self
Raises:: ValueError – If retry_max_delay is less than retry_backoff.

model_config¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

retry_attempts: int = None¶: Total attempts including the initial request (must be >= 1).

retry_backoff: float = None¶: Initial retry delay in seconds.

retry_factor: float = None¶: Backoff multiplier applied after each retry.

retry_max_delay: float = None¶: Maximum delay between retries in seconds.

strict_validation: bool = False¶: Raise on invalid events vs. skip and log.

timeout: int = None¶: Server long-poll timeout in seconds.

class cb_events.Event(/, **data: Any)[source]¶

Bases: BaseEventModel

Event from the Chaturbate Events API.

The main event container that wraps all event types. Use the typed properties to access nested data safely—they return None if data is missing or invalid for the event type.

Example

Safe access to nested data:

if event.type == EventType.TIP:
    if tip := event.tip:
        print(f"Tip: {tip.tokens} tokens")
    if user := event.user:
        print(f"From: {user.username}")

Note

Failures to parse nested model data are logged as a warning and return None. Scalar convenience accessors, such as broadcaster, return None quietly when the value is missing or invalid.

Warning

All string fields in event data (e.g. message.message, user.username, tip.message) originate from untrusted user input and are not sanitized by this library. Escape or validate them before use in HTML, SQL, or shell contexts.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

property broadcaster: str | None¶: Broadcaster username, or None if missing or invalid.

data: dict[str, object] = None¶: Event data payload.

id: str¶: Unique identifier for the event.

property media: Media | None¶: Media purchase data if present and valid (MEDIA_PURCHASE only).

property message: Message | None¶: Message data if present and valid.

property room_subject: RoomSubject | None¶: Room subject if present and valid (ROOM_SUBJECT_CHANGE only).

property tip: Tip | None¶: Tip data if present and valid (TIP events only).

type: EventType = None¶: Type of the event.

property user: User | None¶: User data if present and valid.

class cb_events.EventClient(events_url: str, *, config: cb_events.config.ClientConfig | None = None, rate_limiter: aiolimiter.AsyncLimiter | None = None)[source]¶

Async client for polling the Chaturbate Events API.

Streams events with automatic retries, rate limiting, and credential handling. Use as an async context manager and async iterator.

This client implements long-polling with stateful URL tracking, where the API returns a nextUrl to continue from the last position.

Variables:

username – Chaturbate username parsed from the Events URL.
token – API token parsed from the Events URL.
config – Client configuration instance.
base_url – API endpoint base URL.
session – Active HTTP session (None until context entry).

Example

Basic polling loop:

async with EventClient(
    "https://eventsapi.chaturbate.com/events/username/token/"
) as client:
    async for event in client:
        print(f"Received {event.type}: {event.id}")

Shared rate limiting across multiple clients:

from aiolimiter import AsyncLimiter

limiter = AsyncLimiter(max_rate=2000, time_period=60)
async with (
    EventClient(
        "https://eventsapi.chaturbate.com/events/user1/token1/",
        rate_limiter=limiter,
    ) as c1,
    EventClient(
        "https://eventsapi.chaturbate.com/events/user2/token2/",
        rate_limiter=limiter,
    ) as c2,
):
    # Both clients share the same rate limit pool
    pass

Note

Not thread-safe. Must be used as an async context manager.

Initialize event client with credentials and configuration.

Parameters:

events_url – Full Events API URL provided by upstream, for example https://eventsapi.chaturbate.com/events/<username>/<token>/.
config – Optional client configuration overrides.
rate_limiter – Optional shared rate limiter to coordinate calls across multiple clients.

async close() → None[source]¶

Close the HTTP session and reset internal state.

Releases network resources and clears the stored nextUrl. Safe to call multiple times. Called automatically when exiting the async context manager.

Note

After calling close(), the client must be re-entered via async with before making further requests.

async poll() → list[cb_events.models.Event][source]¶

Poll the API for new events.

Makes a single request to the Events API and returns any available events. Updates _next_url from each response for subsequent polls.

Returns:: List of events received, or an empty list on timeout.

base_url: str¶

config: cb_events.config.ClientConfig¶

session: aiohttp.ClientSession | None = None¶

token: str¶

username: str¶

class cb_events.EventType[source]¶

Bases: str, enum.Enum

Event types from the Chaturbate Events API.

Each member represents a distinct event category that can be received from the API. Use with Router.on() to register type-specific handlers.

Example

Filtering events by type:

@router.on(EventType.TIP)
async def handle_tip(event: Event) -> None:
    print(f"Tip received: {event.tip.tokens}")

Initialize self. See help(type(self)) for accurate signature.

BROADCAST_START = 'broadcastStart'¶: Broadcaster has started streaming.

BROADCAST_STOP = 'broadcastStop'¶: Broadcaster has stopped streaming.

CHAT_MESSAGE = 'chatMessage'¶: Chat message has been sent.

FANCLUB_JOIN = 'fanclubJoin'¶: User has joined the fan club.

FOLLOW = 'follow'¶: User has followed the broadcaster.

MEDIA_PURCHASE = 'mediaPurchase'¶: User has purchased media.

PRIVATE_MESSAGE = 'privateMessage'¶: Private message has been sent.

ROOM_SUBJECT_CHANGE = 'roomSubjectChange'¶: Room subject or title has changed.

TIP = 'tip'¶: User has sent a tip.

UNFOLLOW = 'unfollow'¶: User has unfollowed the broadcaster.

USER_ENTER = 'userEnter'¶: User has entered the room.

USER_LEAVE = 'userLeave'¶: User has left the room.

class cb_events.Media(/, **data: Any)[source]¶

Bases: BaseEventModel

Media purchase transaction details.

Contains information about a media purchase event.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

id: str¶: Identifier of the purchased media.

name: str¶: Name of the purchased media.

tokens: int¶: Number of tokens spent on the media purchase.

type: Literal['video', 'photos']¶: Type of the purchased media.

class cb_events.Message(/, **data: Any)[source]¶

Bases: BaseEventModel

Chat or private message content.

Represents message data from chatMessage and privateMessage events.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

bg_color: str | None = None¶: Background color of the message.

color: str | None = None¶: Text color of the message.

font: str | None = None¶: Font style of the message.

from_user: str | None = None¶: Username of the sender.

property is_private: bool¶: True if this is a private message.

message: str¶: Content of the message.

orig: str | None = None¶: Original message content.

to_user: str | None = None¶: Username of the recipient.

class cb_events.RoomSubject(/, **data: Any)[source]¶

Bases: BaseEventModel

Room subject or title information.

Contains the updated room subject from roomSubjectChange events.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

subject: str¶: The room subject or title.

class cb_events.Router[source]¶

Routes events to registered async handlers.

Provides decorator-based registration for event handlers with support for both type-specific and wildcard (catch-all) handlers. Handlers execute sequentially in registration order; errors are logged but do not prevent subsequent handlers from running.

Example

router = Router()

@router.on(EventType.TIP)
async def handle_tip(event: Event) -> None:
    print(f"Tip: {event.tip.tokens if event.tip else 0}")

@router.on_any()
async def log_event(event: Event) -> None:
    print(f"Event received: {event.type}")

await router.dispatch(event)

Note

Wildcard handlers (registered via on_any()) execute before type-specific handlers for each event to support logging or preprocessing regardless of event type.

Initialize router with an empty handler registry.

async dispatch(event: cb_events.models.Event) → None[source]¶

Dispatch an event to all matching handlers.

Executes wildcard handlers first, then type-specific handlers, in registration order. Handler exceptions are caught, logged, and do not propagate or prevent other handlers from executing.

Parameters:: event – Event instance to dispatch to registered handlers.

on(event_type: cb_events.models.EventType) → collections.abc.Callable[[HandlerFunc], HandlerFunc][source]¶

Parameters:: event_type – Event category to associate with the handler.
Returns:: Decorator that registers the handler and returns it unchanged.

on_any() → collections.abc.Callable[[HandlerFunc], HandlerFunc][source]¶

Returns:: Decorator that registers the handler and returns it unchanged.

class cb_events.Tip(/, **data: Any)[source]¶

Bases: BaseEventModel

Tip transaction details.

Contains information about a tip event including the amount and optional message.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

is_anon: bool = False¶: Whether the tip is anonymous.

message: str | None = None¶: Optional message attached to the tip.

tokens: int¶: Number of tokens tipped.

class cb_events.User(/, **data: Any)[source]¶

Bases: BaseEventModel

User information attached to events.

Contains details about the user who triggered the event, including display name, membership status, and various flags.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

color_group: str | None = None¶

Color group of the user.

Known values: "o" (owner), "m" (moderator), "f" (fanclub), "l" (dark purple), "p" (light purple), "tr" (dark blue), "t" (light blue), "g" (grey).

fc_auto_renew: bool = False¶: Whether the user’s fanclub membership is a recurring subscription.

gender: str | None = None¶

Gender of the user.

Known values: "m" (male), "f" (female), "c" (couple), "t" (trans).

has_darkmode: bool = False¶: Whether the user has dark mode enabled.

has_tokens: bool = False¶: Whether the user has at least 1 token.

in_fanclub: bool = False¶: Whether the user is in the fan club.

in_private_show: bool = False¶: Whether the user is in the broadcaster’s private show.

is_broadcasting: bool = False¶: Whether the user is currently broadcasting.

is_follower: bool = False¶: Whether the user is a follower.

is_mod: bool = False¶: Whether the user is a moderator.

is_owner: bool = False¶: Whether the user is the room owner.

is_silenced: bool = False¶: Whether the user is silenced.

is_spying: bool = False¶: Whether the user is spying on a private show.

language: str | None = None¶

User’s preferred language.

Known values: "de" (German), "en" (English), "es" (Spanish), "fr" (French), "it" (Italian), "ja" (Japanese), "ko" (Korean), "pl" (Polish), "pt" (Portuguese), "ru" (Russian), "zh" (Chinese).

recent_tips: Literal['none', 'few', 'some', 'lots', 'tons'] | None = None¶

How much the user has tipped recently.

Possible values: "none" (no recent tips, no tokens — grey username), "few" (few or no recent tips, has tokens — light blue), "some" (some recent tips — dark blue), "lots" (lots — purple), "tons" (tons — dark purple).

Note

The value "none" is truthy. Compare explicitly with recent_tips is None (field absent) versus recent_tips == "none" (present but no tips).

subgender: Literal['tf', 'tm', 'tn'] | None = None¶

Subgender of the user (only set when gender is "t" / trans).

Possible values: "tf" (transfemme), "tm" (transmasc), "tn" (non-binary). None when the user is not trans.

username: str¶: Display name of the user.

type cb_events.HandlerFunc = Callable[[Event], Awaitable[None]][source]¶: Async handler signature accepted by Router decorators.