Flows
DeviceCodeFlow
¶
Bases: BaseFlow
OAuth2 Device Code Flow.
This flow is designed for devices that either lack a browser or have limited input capabilities. The user authorizes the application on a separate device with a browser.
???+ tip
If you already have a token, you can prevent creating another one::
await flow.authorize(token=client.get_token(client.http.client_id)[0])
Examples:
Context Manager::
async with DeviceCodeFlow("CLIENT_ID", "CLIENT_SECRET") as flow:
device_code = await flow.request_device_code({"user:read:email"})
print(f"Go to: {device_code.verification_uri}")
print(f"Enter code: {device_code.user_code}")
token = await flow.wait_for_device_token(device_code.device_code)
Manual::
flow = DeviceCodeFlow("CLIENT_ID", "CLIENT_SECRET")
await flow.authorize()
device_code = await flow.request_device_code({"user:read:email"})
print(f"Go to: {device_code.verification_uri}")
print(f"Enter code: {device_code.user_code}")
token = await flow.wait_for_device_token(device_code.device_code)
await flow.close()
From existing App::
app = App("CLIENT_ID", "CLIENT_SECRET")
await app.authorize()
flow = await DeviceCodeFlow.from_app(app)
# Flow is now ready to use with the app's existing token
application
property
¶
tokens
property
¶
tokens_validity
property
¶
Get token validity timestamps for all users.
Returns:
| Type | Description |
|---|---|
MappingProxyType[str, Tuple[float, float]]
|
Read-only mapping of user IDs to (last_validated, expires_in) tuples. Timestamps are in Unix epoch format. |
users_scopes
property
¶
Get the scopes for all authorized users.
Returns:
| Type | Description |
|---|---|
MappingProxyType[str, Tuple[str, ...]]
|
Read-only mapping of user IDs to their granted OAuth scopes. |
__aenter__()
async
¶
Async context manager entry.
__aexit__(exc_type, exc_value, traceback)
async
¶
Async context manager exit.
add_token(user_id, access_token, refresh_token)
¶
Add or update tokens for a user without validation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
user_id
|
str
|
The user identifier. |
required |
access_token
|
str
|
The access token for API requests. |
required |
refresh_token
|
Optional[str]
|
The refresh token for token renewal. Can be None. |
required |
Returns:
| Type | Description |
|---|---|
UserAPI
|
A UserAPI instance for the user with stored tokens. |
add_user(access_token, refresh_token)
async
¶
Add and validate a new user with provided tokens.
Validates the provided tokens with Twitch's API and stores them if valid. This method should be preferred over add_token() when token validation is needed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
access_token
|
str
|
The user's access token. |
required |
refresh_token
|
Optional[str]
|
The user's refresh token for automatic renewal. |
required |
Returns:
| Type | Description |
|---|---|
UserAPI
|
A UserAPI instance for the validated and stored user. |
Raises:
| Type | Description |
|---|---|
AuthFailure
|
If the tokens are invalid or authorization fails. |
authorize(*, token=None)
async
¶
Authorize the application for API access.
Note
This method is called automatically when using the async context manager. The application property will be available after successful authorization.
Obtains an application access token using the client credentials flow, enabling access to application-level APIs.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
Optional[str]
|
An optional pre-existing app access token. If None, automatically generates one using client credentials flow. |
None
|
Raises:
| Type | Description |
|---|---|
BadRequest
|
If invalid client id. |
Forbidden
|
If invalid client secret. |
Unauthorized
|
If the client credentials are invalid or expired. |
clear()
¶
Reset the client to its initial state.
Clears all internal state including tasks and events. After calling this method, the client can be reused.
close()
async
¶
Closes the connection to Twitch.
credentials_grant_flow()
async
¶
Perform OAuth2 client credentials grant flow.
Info
Obtains an app access token using the client credentials grant flow. This token can be used for app-only requests that don't require user authorization.
Returns:
| Type | Description |
|---|---|
ClientCredentials
|
App access token information. |
Raises:
| Type | Description |
|---|---|
BadRequest
|
If invalid client ID. |
Forbidden
|
If invalid client secret. |
Unauthorized
|
If the client credentials are invalid or expired. |
from_app(app)
classmethod
¶
Create a DeviceCodeFlow instance from an existing App with authorization.
Danger
The provided App instance must already be authenticated before calling this method.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
app
|
App
|
An existing App instance that should already be authorized. The flow will inherit the app's client credentials and current token. |
required |
Returns:
| Type | Description |
|---|---|
DeviceCodeFlow
|
A new DeviceCodeFlow instance that's already authorized with the app's token. |
Raises:
| Type | Description |
|---|---|
IndexError
|
If the app has no tokens available. |
get_device_token(device_code)
async
¶
Poll for an access token using the device code.
Attempts to exchange a device code for an access token. This should be called repeatedly until the user completes authorization or the device code expires.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
device_code
|
str
|
The device code obtained from request_device_code(). |
required |
Returns:
| Type | Description |
|---|---|
Token
|
Access token and refresh token if authorization is complete. |
Raises:
| Type | Description |
|---|---|
BadRequest
|
If the device code is invalid, expired, or authorization is still pending. |
get_token(user_id)
¶
Get stored tokens for a specific user.
Retrieves the access token and refresh token (if available) for the specified user from local storage.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
user_id
|
str
|
The user identifier to retrieve tokens for. |
required |
Returns:
| Type | Description |
|---|---|
Optional[Tuple[str, Optional[str]]]
|
The stored (access_token, refresh_token) tuple, or None if not found. |
get_token_quota(user_id)
¶
get_user_by_id(user_id)
¶
Get a UserAPI instance for an already authorized user.
Warning
This method does not validate stored tokens. If tokens are invalid or expired, API calls through the returned UserAPI may fail.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
user_id
|
str
|
The user identifier to get API access for. |
required |
Returns:
| Type | Description |
|---|---|
UserAPI
|
A UserAPI instance for the specified user. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the user_id matches the client_id (use application property instead). |
TokenError
|
If no tokens are stored for the specified user_id. |
refresh_token(token)
async
¶
Refresh an expired or expiring access token.
Uses a refresh token to obtain a new access token and refresh token pair. The old tokens will be invalidated after successful refresh.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
The refresh token to use for obtaining new tokens. |
required |
Returns:
| Type | Description |
|---|---|
Token
|
New access and refresh tokens. |
Raises:
| Type | Description |
|---|---|
BadRequest
|
If the refresh token is invalid. |
NotFound
|
If the client ID does not exist. |
Unauthorized
|
If the refresh token is expired or invalid. |
remove_token(user_id)
¶
Remove stored tokens for a user.
Warning
This method only removes tokens from local storage and does NOT revoke them on Twitch's servers. The tokens will remain valid until they expire naturally.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
user_id
|
str
|
The user identifier whose tokens should be removed. |
required |
request_device_code(scopes=frozenset())
async
¶
Request a device code and user code for device flow authentication.
Initiates the device flow by requesting a device code, user code, and verification URI from Twitch. The user will need to visit the verification URI and enter the user code to authorize the application.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scopes
|
Set[str]
|
Set of OAuth scopes to request for the token. If empty, default scopes will be used. |
frozenset()
|
Returns:
| Type | Description |
|---|---|
DeviceCode
|
Contains device_code, user_code, verification_uri, expires_in, and interval for polling. |
Raises:
| Type | Description |
|---|---|
BadRequest
|
If invalid client ID or scopes are provided. |
Forbidden
|
If the client is not authorized for device flow. |
revoke_token(token)
async
¶
Revoke an access token, invalidating it immediately.
Revokes the specified access token, making it invalid for future API requests. This is useful for logout functionality or security purposes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
The access token to revoke. |
required |
Raises:
| Type | Description |
|---|---|
BadRequest
|
If the access token is invalid. |
NotFound
|
If the client ID does not exist. |
validate_token(token)
async
¶
Validate a Twitch access token.
Verifies that the provided access token is valid and returns information about the token including its expiration time, scopes, and associated user.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
The access token to validate. |
required |
Returns:
| Type | Description |
|---|---|
ValidateToken
|
Token validation information. |
Raises:
| Type | Description |
|---|---|
Unauthorized
|
If the token is invalid. |
wait_for_device_token(device_code, interval=5, timeout=300)
async
¶
Wait for user authorization and automatically retrieve the token.
Continuously polls for an access token until the user completes authorization, the request times out, or an error occurs. This method handles the polling logic automatically, including backing off when requested by the server.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
device_code
|
str
|
The device code obtained from request_device_code(). |
required |
interval
|
int
|
Base polling interval in seconds. The actual interval may be adjusted based on server responses. |
5
|
timeout
|
int
|
Maximum time to wait for authorization in seconds. |
300
|
Returns:
| Type | Description |
|---|---|
Token
|
Access token and refresh token once authorization is complete. |
Raises:
| Type | Description |
|---|---|
TimeoutError
|
If the timeout is reached before authorization completes. |
BadRequest
|
If the device code is invalid, expired, or the user denied the authorization request. |
AuthorizationCodeGrantFlow
¶
Bases: BaseFlow
OAuth2 Authorization Code Grant Flow.
This flow is designed for web applications that can securely store a client secret and handle HTTP redirects. Users are redirected to Twitch to authorize the application, then redirected back with an authorization code that can be exchanged for tokens.
Tip
This flow is ideal for web applications with server-side components that can handle HTTP redirects and store secrets securely.
Examples:
With FastAPI Server::
from twitch.oauth import AuthorizationCodeGrantFlow
from fastapi.responses import RedirectResponse
from contextlib import asynccontextmanager
from fastapi import FastAPI
import uvicorn
flow = AuthorizationCodeGrantFlow('CLIENT_ID', 'CLIENT_SECRET')
@asynccontextmanager
async def lifespan(_):
await flow.authorize()
yield
await flow.close()
app = FastAPI(lifespan=lifespan)
@app.get('/login')
def login():
auth_url = flow.get_authorization_url(
redirect_uri='http://localhost:3000/callback',
scopes={'user:read:email'}
)
return RedirectResponse(url=auth_url)
@app.get('/callback')
async def callback(code: str):
token = await flow.get_authorization_token(code=code, redirect_uri='http://localhost:3000/callback')
return token.raw
uvicorn.run(app, host='localhost', port=3000)
Manual::
flow = AuthorizationCodeGrantFlow('CLIENT_ID', 'CLIENT_SECRET')
await flow.authorize()
# Generate auth URL for user
auth_url = flow.get_authorization_url(
redirect_uri='http://localhost:3000/callback',
scopes={'user:read:email'}
)
print(f'Visit: {auth_url}')
# After getting code from callback, exchange for token
code = input("Enter the code from the callback URL: ")
token = await flow.get_authorization_token(code, 'http://localhost:3000/callback')
await flow.close()
application
property
¶
tokens
property
¶
tokens_validity
property
¶
Get token validity timestamps for all users.
Returns:
| Type | Description |
|---|---|
MappingProxyType[str, Tuple[float, float]]
|
Read-only mapping of user IDs to (last_validated, expires_in) tuples. Timestamps are in Unix epoch format. |
users_scopes
property
¶
Get the scopes for all authorized users.
Returns:
| Type | Description |
|---|---|
MappingProxyType[str, Tuple[str, ...]]
|
Read-only mapping of user IDs to their granted OAuth scopes. |
__aenter__()
async
¶
Async context manager entry.
__aexit__(exc_type, exc_value, traceback)
async
¶
Async context manager exit.
add_token(user_id, access_token, refresh_token)
¶
Add or update tokens for a user without validation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
user_id
|
str
|
The user identifier. |
required |
access_token
|
str
|
The access token for API requests. |
required |
refresh_token
|
Optional[str]
|
The refresh token for token renewal. Can be None. |
required |
Returns:
| Type | Description |
|---|---|
UserAPI
|
A UserAPI instance for the user with stored tokens. |
add_user(access_token, refresh_token)
async
¶
Add and validate a new user with provided tokens.
Validates the provided tokens with Twitch's API and stores them if valid. This method should be preferred over add_token() when token validation is needed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
access_token
|
str
|
The user's access token. |
required |
refresh_token
|
Optional[str]
|
The user's refresh token for automatic renewal. |
required |
Returns:
| Type | Description |
|---|---|
UserAPI
|
A UserAPI instance for the validated and stored user. |
Raises:
| Type | Description |
|---|---|
AuthFailure
|
If the tokens are invalid or authorization fails. |
authorize(*, token=None)
async
¶
Authorize the application for API access.
Note
This method is called automatically when using the async context manager. The application property will be available after successful authorization.
Obtains an application access token using the client credentials flow, enabling access to application-level APIs.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
Optional[str]
|
An optional pre-existing app access token. If None, automatically generates one using client credentials flow. |
None
|
Raises:
| Type | Description |
|---|---|
BadRequest
|
If invalid client id. |
Forbidden
|
If invalid client secret. |
Unauthorized
|
If the client credentials are invalid or expired. |
clear()
¶
Reset the client to its initial state.
Clears all internal state including tasks and events. After calling this method, the client can be reused.
close()
async
¶
Closes the connection to Twitch.
credentials_grant_flow()
async
¶
Perform OAuth2 client credentials grant flow.
Info
Obtains an app access token using the client credentials grant flow. This token can be used for app-only requests that don't require user authorization.
Returns:
| Type | Description |
|---|---|
ClientCredentials
|
App access token information. |
Raises:
| Type | Description |
|---|---|
BadRequest
|
If invalid client ID. |
Forbidden
|
If invalid client secret. |
Unauthorized
|
If the client credentials are invalid or expired. |
from_app(app)
classmethod
¶
Create a AuthorizationCodeGrantFlow instance from an existing App with authorization.
Danger
The provided App instance must already be authenticated before calling this method.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
app
|
App
|
An existing App instance that should already be authorized. The flow will inherit the app's client credentials and current token. |
required |
Returns:
| Type | Description |
|---|---|
AuthorizationCodeGrantFlow
|
A new AuthorizationCodeGrantFlow instance that's already authorized with the app's token. |
Raises:
| Type | Description |
|---|---|
IndexError
|
If the app has no tokens available. |
get_authorization_token(code, redirect_uri)
async
¶
Exchange an authorization code for an access token and refresh token.
After the user authorizes your application, Twitch redirects them to your redirect URI with an authorization code. Use this method to exchange that code for usable access and refresh tokens.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
str
|
The authorization code returned from the authorization step. This is provided in the 'code' query parameter of the callback URL. |
required |
redirect_uri
|
str
|
Your app's registered redirect URI. This must exactly match the redirect_uri used in get_authorization_url(). |
required |
Returns:
| Type | Description |
|---|---|
Token
|
Contains access_token, refresh_token, expires_in, scopes, and token_type information. |
Raises:
| Type | Description |
|---|---|
BadRequest
|
If the authorization code is invalid, expired, or the redirect_uri doesn't match. |
Unauthorized
|
If the client credentials are invalid. |
Forbidden
|
If the client is not authorized for this grant type. |
get_authorization_url(redirect_uri, scopes=frozenset(), state=None, force_verify=False)
¶
Generate the authorization URL for the authorization code grant flow.
Creates a URL that users should visit to authorize your application. After authorization, Twitch will redirect them to your redirect_uri with an authorization code.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
redirect_uri
|
str
|
Your app's registered redirect URI. The authorization code will be sent to this URI after user authorization. |
required |
scopes
|
Set[str]
|
Set of OAuth scopes to request for the token. If empty, default scopes will be used. |
frozenset()
|
state
|
Optional[str]
|
A randomly generated string to help prevent CSRF attacks. Strongly recommended for security. |
None
|
force_verify
|
bool
|
Whether to force the user to re-authorize your app's access to their resources. Default is False. |
False
|
Returns:
| Type | Description |
|---|---|
str
|
The authorization URL that users should visit to authorize your application. |
get_token(user_id)
¶
Get stored tokens for a specific user.
Retrieves the access token and refresh token (if available) for the specified user from local storage.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
user_id
|
str
|
The user identifier to retrieve tokens for. |
required |
Returns:
| Type | Description |
|---|---|
Optional[Tuple[str, Optional[str]]]
|
The stored (access_token, refresh_token) tuple, or None if not found. |
get_token_quota(user_id)
¶
get_user_by_id(user_id)
¶
Get a UserAPI instance for an already authorized user.
Warning
This method does not validate stored tokens. If tokens are invalid or expired, API calls through the returned UserAPI may fail.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
user_id
|
str
|
The user identifier to get API access for. |
required |
Returns:
| Type | Description |
|---|---|
UserAPI
|
A UserAPI instance for the specified user. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the user_id matches the client_id (use application property instead). |
TokenError
|
If no tokens are stored for the specified user_id. |
refresh_token(token)
async
¶
Refresh an expired or expiring access token.
Uses a refresh token to obtain a new access token and refresh token pair. The old tokens will be invalidated after successful refresh.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
The refresh token to use for obtaining new tokens. |
required |
Returns:
| Type | Description |
|---|---|
Token
|
New access and refresh tokens. |
Raises:
| Type | Description |
|---|---|
BadRequest
|
If the refresh token is invalid. |
NotFound
|
If the client ID does not exist. |
Unauthorized
|
If the refresh token is expired or invalid. |
remove_token(user_id)
¶
Remove stored tokens for a user.
Warning
This method only removes tokens from local storage and does NOT revoke them on Twitch's servers. The tokens will remain valid until they expire naturally.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
user_id
|
str
|
The user identifier whose tokens should be removed. |
required |
revoke_token(token)
async
¶
Revoke an access token, invalidating it immediately.
Revokes the specified access token, making it invalid for future API requests. This is useful for logout functionality or security purposes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
The access token to revoke. |
required |
Raises:
| Type | Description |
|---|---|
BadRequest
|
If the access token is invalid. |
NotFound
|
If the client ID does not exist. |
validate_token(token)
async
¶
Validate a Twitch access token.
Verifies that the provided access token is valid and returns information about the token including its expiration time, scopes, and associated user.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
The access token to validate. |
required |
Returns:
| Type | Description |
|---|---|
ValidateToken
|
Token validation information. |
Raises:
| Type | Description |
|---|---|
Unauthorized
|
If the token is invalid. |