Miwory 539a6a04f1
All checks were successful
Build And Publish Package / publish (push) Successful in 30s
Merge pull request '1.1.0' (#9) from dev into latest
Reviewed-on: #9
2026-01-16 09:56:07 +03:00
2025-12-11 15:56:56 +03:00
2025-12-11 15:56:56 +03:00

Twitch Client (Async)

An internal async Twitch API client built on top of AioHTTPX, providing typed access to the Twitch Helix API and OAuth2 endpoints with built-in rate limiting and optional caching.


What it provides

  • TwitchAPIClient — async client for Twitch Helix API
  • TwitchAuthClient — async client for Twitch OAuth2
  • Built on AioHTTPX (aiohttp transport, Redis rate limiting & caching)
  • Strongly typed responses using Pydantic models
  • Automatic request parameter cleanup (None values removed)

Installation (internal)

Configure your private package index, then:

pip install twitchclient

Authentication

OAuth helper client

from twitchclient.auth import TwitchAuthClient
from twitchclient import scopes

auth = TwitchAuthClient(
    client_id="CLIENT_ID",
    client_secret="CLIENT_SECRET",
    redirect_uri="https://your.app/callback",
)

url = await auth.create_authorization_code_grant_flow_url(
    scope=[scopes.USER_READ_EMAIL, scopes.CHAT_READ],
)

App access token

token = await auth.app_access_token()
print(token.access_token)

User access token

token = await auth.user_access_token(code="AUTH_CODE")

API Client

from twitchclient.api import TwitchAPIClient

client = TwitchAPIClient(
    client_id="CLIENT_ID",
    client_secret="CLIENT_SECRET",
    redirect_uri="https://your.app/callback",
    redis_url="redis://localhost:6379",  # optional
)
  • Base URL: https://api.twitch.tv/helix
  • Client-Id header is set automatically
  • Rate limit defaults to 700 req/min when Redis is enabled

Example usage

Get channel information

channels = await client.get_channel_information(
    access_token=token.access_token,
    broadcaster_id=123456,
)

Start a commercial

await client.start_commercial(
    access_token=token.access_token,
    broadcaster_id=123456,
)

Cached request

Caching is controlled via the X-Cache-TTL header:

streams = await client.get_streams(
    access_token=token.access_token,
    game_id=509658,
    cache_time=30,
)

Rate limiting & caching

  • Redis-backed
  • Shared key: twitch
  • Enabled automatically when redis_url is provided
  • Cache TTL is per-request (X-Cache-TTL header)

Implementation is inherited from AioHTTPX transports .


Errors

All API methods raise typed exceptions:

  • ClientError(status_code, message)
  • InternalError(status_code, message)

Defined in schema module .


Scopes

OAuth scopes are defined as constants and typed literals:

from twitchclient import scopes

scopes.CHANNEL_READ_SUBSCRIPTIONS
scopes.CHAT_READ

See scopes.py for the full list .


Typed responses

All successful responses return Pydantic models, e.g.:

  • ChannelsInformation
  • Streams
  • Users
  • EventsubSubscriptions

Models live in schema.py .


Notes & limitations

  • Async-only (no sync client)
  • Internal API, no stability guarantees
  • Some endpoints still marked TODO (e.g. Guest Star)

Description
No description provided
Readme 232 KiB
Languages
Python 100%