TikTokLive.client package

Subpackages

Submodules

TikTokLive.client.client module

Bases: AsyncIOEventEmitter

A client to connect to & read from TikTok LIVE streams

Method that can be used to register a Python function as an event listener

Parameters:

event – The event to listen to
f – The function to handle the event

Returns:

The generated pyee.Handler object

async close() → None

Discards the async sessions if you don’t intend to use the client again

Returns:: None

async connect(callback: Callable[[None], None] | Callable[[None], Coroutine[None, None, None]] | Coroutine[None, None, None] | None = None, **kwargs) → Task

Start a future-blocking connection to TikTokLive

Parameters:

callback – A callback function to run when connected
kwargs – Kwargs to pass to start

Returns:

The task, once it’s finished

property connected: bool

Whether the WebSocket client is currently connected to TikTok

Returns:: Connection status

async disconnect(close_client: bool = False) → None

Disconnect the client from the websocket.

Parameters:: close_client – Whether to also close the HTTP client if you don’t intend to reuse it
Returns:: None

property gift_info: dict | None

Information about the stream’s gifts if fetch_gift_info=True when starting the client e.g. with client.run)

Returns:: The stream gift info

Extract CustomEvent events from existing ProtoEvent events

Parameters:

response – The WebcastResponseMessage to parse for the custom event
event – The ProtoEvent to parse for the custom event

Returns:

The event, if one exists

Check whether the client is listening to a given event

Parameters:: event – The event to check listening for
Returns:: Whether it is being listened to

async is_live(unique_id: str | None = None) → bool

Check if the client is currently live on TikTok

Parameters:: unique_id – Optionally override the user to check
Returns:: Whether they are live on TikTok

property logger: Logger

The internal logger used by TikTokLive

Returns:: An instance of a logging.Logger

Decorator that can be used to register a Python function as an event listener

Parameters:

event – The event to listen to
f – The function to handle the event

Returns:

The wrapped function as a generated pyee.Handler object

classmethod parse_unique_id(unique_id: str) → str

Parse unique ID from a generic string

Parameters:: unique_id – The unique_id to parse
Returns:: The parsed unique_id

property room_id: int | None

The room ID the user is currently connected to

Returns:: Room ID or None

property room_info: dict | None

Information about the room if fetch_room_info=True when starting the client (e.g. with client.run)

Returns:: Dictionary of room info

run(**kwargs) → Task

Start a thread-blocking connection to TikTokLive

Parameters:: kwargs – Kwargs to pass to start
Returns:: The task, once it’s finished

async start(*, process_connect_events: bool = True, compress_ws_events: bool = True, fetch_room_info: bool = False, fetch_gift_info: bool = False, fetch_live_check: bool = True, room_id: int | None = None) → Task

Create a non-blocking connection to TikTok LIVE and return the task

Parameters:

process_connect_events – Whether to process initial events sent on room join
fetch_room_info – Whether to fetch room info on join
fetch_gift_info – Whether to fetch gift info on join
fetch_live_check – Whether to check if the user is live (you almost ALWAYS want this enabled)
room_id – An override to the room ID to connect directly to the livestream and skip scraping the live. Useful when trying to scale, as scraping the HTML can result in TikTok blocks.
compress_ws_events – Whether to compress the WebSocket events using gzip compression (you should probably have this on)

Returns:

Task containing the heartbeat of the client

property unique_id: str: The cleaned unique-id parameter passed to the client

property web: TikTokWebClient

The HTTP client that this client uses for requests

Returns:: A copy of the TikTokWebClient

TikTokLive.client.errors module

exception TikTokLive.client.errors.AgeRestrictedError

Bases: RuntimeError

Thrown when a LIVE is age restricted. Pass sessionid to bypass.

exception TikTokLive.client.errors.AlreadyConnectedError

Bases: RuntimeError

Thrown when attempting to connect to a user that is already connected to

exception TikTokLive.client.errors.InitialCursorMissingError

Bases: RuntimeError

Thrown when the cursor for connecting to TikTok is missing (blocked)

exception TikTokLive.client.errors.PremiumEndpointError(*args, api_message: str): Bases: SignAPIError

exception TikTokLive.client.errors.SignAPIError(reason: ErrorReason, *args: str)

Bases: RuntimeError

Thrown when a fetch to the Sign API fails for one reason or another

class ErrorReason(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Possible failure reasons

CONNECT_ERROR = 2

EMPTY_COOKIES = 5

EMPTY_PAYLOAD = 3

PREMIUM_ENDPOINT = 6

RATE_LIMIT = 1

SIGN_NOT_200 = 4

classmethod format_sign_server_message(message: str) → str: Format the sign server message

exception TikTokLive.client.errors.SignatureMissingTokensError(*args): Bases: SignAPIError

exception TikTokLive.client.errors.SignatureRateLimitError(retry_after: int, reset_time: int, api_message: str | None, *args)

Bases: SignAPIError

Thrown when a user hits the Sign API limit

property reset_time: int: The unix timestamp for when the client can request again

property retry_after: int: How long to wait until the next attempt

exception TikTokLive.client.errors.UnexpectedSignatureError(*args): Bases: SignAPIError

exception TikTokLive.client.errors.UserNotFoundError(unique_id: str, *args)

Bases: RuntimeError

Thrown when the request to check if a user is live fails because a user has no livestream account (e.g. <1000 followers)

exception TikTokLive.client.errors.UserOfflineError

Bases: RuntimeError

Thrown when the requested streamer to watch is offline

exception TikTokLive.client.errors.WebcastBlocked200Error

Bases: RuntimeError

Thrown when the webcast is blocked by TikTok with a 200 status code (detected)

exception TikTokLive.client.errors.WebsocketURLMissingError

Bases: RuntimeError

Thrown when the websocket URL to connect to TikTok is missing (blocked)

TikTokLive.client.logger module

class TikTokLive.client.logger.LogLevel(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

The level to be used with the python logging module

CRITICAL = 50

DEBUG = 10

ERROR = 40

INFO = 20

NOTSET = 0

WARNING = 30

property value: int

Return the enum item value

Returns:: Value recast (correctly) as an int

class TikTokLive.client.logger.TikTokLiveLogHandler(stream: Any | None = None, formatter: Formatter | None = None)

Bases: StreamHandler

A custom logger handler for TikTokLive

FORMAT: str = '[%(name)s] %(levelname)s from %(stack)s:%(lineno)d — %(message)s'

LOGGER: Logger | None = <Logger TikTokLive (WARNING)>

LOGGER_NAME: str = 'TikTokLive'

SPACING: Dict[int, int] = {10: 0, 20: 1, 30: 1, 40: 0}

TIME_FORMAT: str = '%H:%M:%S'

emit(record: LogRecord) → None

Handle emitting from the logger

Parameters:: record – The record to emit from
Returns:: None

classmethod format_path(record: LogRecord) → str

Take a path from a stacktrace and convert it into compressed form similar to SpringBoot.

Parameters:: record – The record to read
Returns:: The formatted path in dot-format

classmethod get_logger(level: LogLevel | None = None, stream: Any | None = None) → Logger

Create a logger or retrieve the existing one

Parameters:

stream – Where to stream to
level – The level to log above

Returns:

Instance of new logger

TikTokLive.client package

Subpackages

Submodules

TikTokLive.client.client module

TikTokLive.client.errors module

TikTokLive.client.logger module

Module contents