Low Level APIs#

This page is an API reference for some of the more low-level APIs contained within TwitchIO.

Disclaimer: Some of these are unstable APIs, meaning that they have no version guarantees, and can break at any time. Those APIs will be marked as such, and are only here for reference purposes.

If you are using these APIs, it is expected that you are proficient in Python, and can teach yourself complex topics.

Shard Managers#

This is just an API reference. For a detailed explanation, please see Sharding.

class twitchio.BaseShardManager(**kwargs)#

This is an advanced subclassable object that can be passed to your Client or Bot.

New in version 3.0.

token_handler#

Your Client’s token handler.

Type

TokenHandler

client#

The client for this shard manager.

Type

Client

shards#

The shards this shard manager has control over.

Type

dict[str: Shard]

initial_channels#

The initial channels passed to the Client or Bot. Callables passed to the Client/Bot will have been transformed into a list of channels by this point. If you haven’t passed any initial channels, this will be None.

Type

list[str]

add_shard(shard_id: str, authorized_user: str | None, *, initial_channels: Optional[Iterable[str]] = None) → Shard#

Adds a shard to the manager. This method does not start the shard, to do that, use Shard.start()

Parameters

  • shard_id (str) – The unique ID to refer to this shard by. It must be unique to this manager.

  • authorized_user (str | None) – The user to sign in as. You must have a user token for this username.

  • initial_channels (list[str] | None) – The initial channels this websocket should connect to.

Returns

The shard instance that was created.

Return type

Shard

coroutine assign_shard(channel_name: str, is_initial_channel: bool) → None#

This function is a coroutine.

Method to be overriden in a subclass.

This needs to be implemented to assign a channel to a shard (otherwise channels cannot be joined). This will be called in-between setup() and start().

Note

The channel should be assigned inside of this callback.

Parameters

  • channel_name (str) – The channel that needs to be joined.

  • is_initial_channel (bool) – Indicates if the channel is part of the initial channel list. This can be useful to ignore channels from the initial channel list when you’ve already assigned them by passing initial_channels to Shard.

coroutine get_sender_shard(channel_name: str) → Shard#

This function is a coroutine.

Method to be overriden in a subclass.

This method is called whenever the library needs to know what shard to use to send a message with. Unless you have different users logged in on different shards, you can typically return any shard.

Note

You do not need to have joined a channel’s chat to send messages to it.

Parameters

channel_name (str) – The channel to send a message to.

coroutine setup(**kwargs) → None#

This function is a coroutine.

Method to be overriden in a subclass.

This is a setup hook that should be used to create shards, but not start them.

Parameters

**kwargs (dict[str: Any]) – The kwargs passed to the Client/Bot.

coroutine start() → None#

This function is a coroutine.

Method to be overriden in a subclass.

This method is called when Client/Bot.start is called. It should be used to start shards, and should wait until they’ve all exited before returning. You can use await self.wait_until_exit to do this easily.

coroutine start_all_shards() → None#

This function is a coroutine.

A helper method to start all shards that have not yet been started.

coroutine stop() → None#

This function is a coroutine.

Method to be overriden in a subclass.

This method is called when Client/Bot.stop is called.

coroutine stop_all_shards() → None#

This function is a coroutine.

A helper method to stop all shards.

coroutine wait_until_exit() → None#

This function is a coroutine.

Helper function that blocks until all shards exit.

class twitchio.DefaultShardManager(**kwargs)#

The default shard manager. This will be used by default if no other shard manager is passed to Client.

The documentation for each overriden function below will contain it’s implementation for reference purposes.

New in version 3.0.

main_shard#

The sole shard in use.

Type

Shard

coroutine assign_shard(channel_name: str, is_initial_channel: bool) → None#

This function is a coroutine.

Assigns a channel to a shard. Because this implementation only uses one shard, it is quite simple.

The implementation looks like this:

if is_initial_channel:
    return

await self.main_shard.add_channels((channel_name,))
coroutine get_sender_shard(channel_name: str) → Shard#

This function is a coroutine.

Returns the only shard available. Nothing interesting.

Parameters

channel_name (str) – The channel to send a message to.

coroutine setup(**kwargs) → None#

This function is a coroutine.

Sets up the DefaultShardManager. Calls TokenHandler.get_irc_token(None) to get a token to log in with.

The implementation looks like this:

_, user = await self.token_handler._client_get_irc_login(self.client, None)
self.main_shard = self.add_shard("main-shard", user.name)
coroutine start() → None#

This function is a coroutine.

Starts the shard, enabling it to send and receive messages.

The implementation looks like this:

await self.main_shard.start(block=True)
coroutine stop() → None#

This function is a coroutine.

Stops the shard, closing its connection to twitch.

The implementation looks like this:

await self.main_shard.stop()
class twitchio.DistributedShardManager(**kwargs)#

A distributed shard manager. This shard manager will distribute channels across multiple shards, allowing for more channels to be read from concurrently. This shard manager still uses one user as the login user.

The documentation for each overriden function can be viewed on GitHub.

New in version 3.0.

The following parameters are passed to Client:

Parameters

  • channels_per_shard (int) – The maximum amount of channels per shard. Defaults to 25.

  • max_shard_count (int | None) – The maxiumum amount of shards that the manager is allowed to create. Defaults to 5.

  • initial_shard_count (int) – The amount of initial shards that the manager should create and distribute initial channels across. Defaults to 1.

channels_per_shard#

The amount of channels that can be put on each shard.

Type

int

max_shard_count#

The maximum amount of shards that the manager is allowed to create.

Type

int | None

coroutine assign_shard(channel_name: str, is_initial_channel: bool) → None#

This function is a coroutine.

Assigns a channel to a shard. To distribute evenly, this method checks all existing shard levels, and finds the one with the least amount of shards. If all the shards are at their limits, it will create a new shard instead.

The implementation looks like this:

if is_initial_channel:
    return

ideal_shard = min(self.shards.values(), key=lambda shard: len(shard.websocket._channels))

if len(ideal_shard.websocket._channels) >= self.channels_per_shard:
    if len(self.shards) >= self.max_shard_count:
        raise RuntimeError("All shards are full, and the shard count limit has been reached.")

    new_shard_id = f"extended-shard-{self.next_shard_id}"
    self.next_shard_id += 1

    ideal_shard = self.add_shard(new_shard_id, self.authorized_name)
    await ideal_shard.start(block=False)

await ideal_shard.add_channels((channel_name,))
coroutine get_sender_shard(channel_name: str) → Shard#

This function is a coroutine.

Fetches the first shard available to send from.

The implementation looks like this:

return next(iter(self.shards.values()))
coroutine setup(**kwargs) → None#

This function is a coroutine.

Sets up the initial state of the shard manager. creates initial_shard_count shards, and splits the initial channels (if any) into them. If more initial channels are provided than the initial shard count can handle, it will create more.

The implementation of this is quite long, please view it on GitHub.

coroutine start() → None#

This function is a coroutine.

Starts all shards, enabling them to send and receive messages.

The implementation looks like this:

await self.start_all_shards()
await self.wait_until_exit()
coroutine stop() → None#

This function is a coroutine.

Stops all shards, closing their connections to twitch.

The implementation looks like this:

await self.stop_all_shards()