---

███████╗ █████╗ ███████╗████████╗ █████╗ ██████╗ ██╗
██╔════╝██╔══██╗██╔════╝╚══██╔══╝██╔══██╗██╔══██╗██║
█████╗  ███████║███████╗   ██║   ███████║██████╔╝██║
██╔══╝  ██╔══██║╚════██║   ██║   ██╔══██║██╔═══╝ ██║
██║     ██║  ██║███████║   ██║   ██║  ██║██║     ██║
╚═╝     ╚═╝  ╚═╝╚══════╝   ╚═╝   ╚═╝  ╚═╝╚═╝     ╚═╝
                                                    
███████╗████████╗███████╗ █████╗ ███╗   ███╗
██╔════╝╚══██╔══╝██╔════╝██╔══██╗████╗ ████║
███████╗   ██║   █████╗  ███████║██╔████╔██║
╚════██║   ██║   ██╔══╝  ██╔══██║██║╚██╔╝██║
███████║   ██║   ███████╗██║  ██║██║ ╚═╝ ██║
╚══════╝   ╚═╝   ╚══════╝╚═╝  ╚═╝╚═╝     ╚═╝

---

📦 installation

pip install fastapi-twitch            # core (httpx + pydantic)
pip install fastapi-twitch[fastapi]   # FastAPI integration (TwitchAuth)
pip install fastapi-twitch[redis]     # RedisStateStore for the CSRF state

Full working examples are in the examples/ directory - a complete app with both session- and Redis-backed state stores and a token-storage pattern.

---

📑 quick start

from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse
from starlette.middleware.sessions import SessionMiddleware

from fastapi_twitch import TwitchAuth

twitch = TwitchAuth(
    client_id="...",          # your Twitch application credentials
    client_secret="...",
    redirect_uri="http://localhost:8000/auth/callback",
)

app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="change-me-in-prod")


@app.get("/login")
async def login(request: Request):
    return await twitch.redirect_to_login(request)  # 307 to id.twitch.tv


@app.get("/auth/callback")
async def callback(request: Request, state: str):
    token_set = await twitch.authenticate(request, expected_state=state)
    # token_set.tokens + token_set.user - persist them wherever you want
    return RedirectResponse("/me", status_code=307)

---

🧩 features

• 🔐 OAuth 2.0 done right - Authorization Code grant + PKCE S256, always on
• 🎯 single-use CSRF state - atomic read-and-delete (GETDEL in Redis, pop in the session); replaying a state answers 400
• 🧠 two state backends - SessionStateStore (zero extra deps) and RedisStateStore
• 🧩 framework-agnostic core - TwitchOAuthClient has zero web-framework dependencies
• ⚡ FastAPI integration - TwitchAuth shipped as an optional [fastapi] extra (lazy import, no hard dependency)
• 📇 account data - Helix GET /users for yourself or any user, with batching (max 100 per request)
• 🔒 token hygiene - secrets masked in repr; client_secret sent as client_secret_basic
• ✅ fully typed - py.typed marker, mypy --strict clean
• 🐍 Python 3.10+

---

⚖️ comparison with alternatives

Feature	fastapi-twitch	authlib	social-auth-core	manual (DIY)
Twitch OAuth 2.0 (Auth Code)	✅	✅	✅	🔧
PKCE S256 built-in	✅	⚠️ opt-in	❌	🔧
Single-use CSRF state (atomic)	✅	❌	❌	🔧
Async (asyncio / httpx)	✅	✅	❌ (requests)	🔧
Framework-agnostic core	✅	⚠️ needs glue/storage	⚠️ needs strategy/storage glue	-
FastAPI-native integration	✅	⚠️ manual wiring	❌	🔧
Account data via Helix /users	✅ built-in	❌	❌	🔧
Pydantic v2 models	✅	❌	❌	-
py.typed, mypy --strict clean	✅	❌	❌	-
Runtime deps (basic mode)	httpx + pydantic	httpx + more	requests + many	-
Python versions	3.10+	3.10+	3.10+	-

Each library has its own strengths - choose what fits your use case.

---

📖 usage

TwitchAuth (FastAPI integration)

TwitchAuth owns the OAuth client, the state store, and the redirect-URI contract. State + PKCE are handled internally: starting a login stashes the code_verifier (bound to the state) in the store, and authenticate() pops it atomically.

from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse
from fastapi_twitch import TwitchAuth

twitch = TwitchAuth(
    client_id="...",
    client_secret="...",
    redirect_uri="http://localhost:8000/auth/callback",
)


@app.get("/login")
async def login(request: Request):
    return await twitch.redirect_to_login(request)  # 307 to Twitch consent


@app.get("/auth/callback")
async def callback(request: Request, state: str):
    token_set = await twitch.authenticate(request, expected_state=state)
    return RedirectResponse("/me", status_code=307)

Lifecycle: TwitchAuth owns an httpx.AsyncClient. Use it as an async context manager or close it explicitly:

async with twitch:  # closes the internal client on exit
    token_set = await twitch.authenticate(request, expected_state=state)
# or, for a long-lived instance:
await twitch.aclose()

Errors are regular FastAPI HTTPExceptions, straightforward to catch:

Condition	Status code
empty / unknown / replayed state, missing code	400
Twitch rejects the code (OAuthError, e.g. invalid_grant)	400
Twitch API unavailable	502

Why are state and PKCE required?

Without a state token bound to the session, an attacker can launch a login CSRF attack: they start a Twitch login in your app and trick the victim into finishing it - logging the victim into the attacker's account. The state token (plus the PKCE code_verifier) binds the callback to the session that started the login. Replaying a consumed or unknown state is rejected because the store deletes it on the first read.

PKCE additionally protects the code exchange: even if the authorization code is intercepted, it cannot be redeemed without the code_verifier that was stashed server-side when the login started.

Framework-agnostic usage

TwitchOAuthClient is independent of any web framework:

from fastapi_twitch import TwitchOAuthClient

async def main() -> None:
    client = TwitchOAuthClient(client_id="...", client_secret="...")

    state = client.generate_state()
    verifier, challenge = client.generate_pkce_pair()

    login_url = client.login_url(
        redirect_uri="https://example.com/auth/callback",
        state=state,
        code_challenge=challenge,
        scopes=["user:read:email"],
    )
    # redirect the user's browser to `login_url`...

    tokens = await client.exchange_code(
        code=code,
        redirect_uri="https://example.com/auth/callback",
        code_verifier=verifier,
    )

    user = await client.fetch_user(access_token=tokens.access_token)
    print(user.display_name, user.email)

    await client.aclose()

The core does not import fastapi; importing it never pulls in a web framework.

Custom httpx client

A custom httpx.AsyncClient can be injected; the library will never close it:

import httpx

transport = httpx.AsyncHTTPTransport(retries=3, pool_connections=20)
async with httpx.AsyncClient(transport=transport) as http:
    twitch = TwitchAuth(
        client_id="...",
        client_secret="...",
        redirect_uri="http://localhost:8000/auth/callback",
        http_client=http,
    )

State stores

The default is the session (a cookie signed by Starlette's SessionMiddleware) - zero external services. Omit state_store entirely and add the middleware:

from fastapi import FastAPI
from starlette.middleware.sessions import SessionMiddleware
from fastapi_twitch import TwitchAuth

twitch = TwitchAuth(client_id="...", client_secret="...",
                    redirect_uri="http://localhost:8000/auth/callback")

app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="...")

Redis gives atomic GETDEL reads across instances (multi-worker / multi-node):

import redis.asyncio as aioredis
from fastapi_twitch import RedisStateStore, TwitchAuth

twitch = TwitchAuth(
    client_id="...",
    client_secret="...",
    redirect_uri="http://localhost:8000/auth/callback",
    state_store=RedisStateStore(aioredis.from_url("redis://localhost:6379")),
)

Any store implementing the two-method StateStore protocol (put/pop) works - write your own for Postgres, Memcached, or whatever fits.

---

🛡️ security

The library enforces the following on every login:

1. PKCE S256 is mandatory - generated per login, not a flag you can disable.
2. CSRF state is secrets.token_urlsafe(32) with a 10-minute TTL, bound to the session, and consumed atomically on callback (Redis GETDEL / session pop). Reuse of a state answers 400.
3. client_secret is sent as client_secret_basic (Authorization header, RFC 6749 §2.3.1) - never in the URL or request body.
4. Tokens are masked in repr() - secrets never leak into logs.
5. redirect_uri is HTTPS-only except for localhost / 127.0.0.1 / ::1 (local development).
6. Refresh-token rotation - each exchange/refresh returns a fresh access/refresh pair as a new immutable instance; store the new one and drop the old.

You should still encrypt refresh tokens at rest in your own storage.

---

⚠️ twitch's own limits

• Refresh tokens are single-use. Every /oauth2/token call that uses a refresh token rotates it: the old refresh token stops working. Handle rotation races in your store (e.g. update-on-refresh with a single writer).
• /oauth2/validate never returns email - even with user:read:email. Email comes only from Helix GET /users.
• Helix public fields need no scope (display_name, description, profile_image_url, view_count, broadcaster_type, created_at) but still require a valid token. Email needs user:read:email. Without it, the library falls back to the /validate profile (email=None) instead of crashing.
• App access tokens (server-to-server) are obtained via grant_type=client_credentials, which this library does not wrap - but you can pass such a token to fetch_user/fetch_users to read public profiles.

---

📝 models

TwitchTokens

The token set returned by exchange_code/refresh_tokens. Immutable (frozen)

• rotating produces a new instance.

Field	Type	Description
access_token	str	Twitch OAuth access token (Helix API)
refresh_token	str	refresh token (single-use; rotates on refresh)
expires_in	int	seconds until the access token expires
scope	list[str]	granted scopes
token_type	Literal["bearer"]	always bearer
obtained_at	datetime	UTC timestamp of issuance

Methods: expires_at (property), is_expired(leeway=60) -> bool.

TwitchUser

Merges /oauth2/validate (always present) with Helix GET /users (display_name, email, ... may be None when unavailable).

Field	Type	Source
id	str	/oauth2/validate user_id
login	str	account name (lowercase)
scopes	list[str]	granted scopes
display_name	str | None	Helix
email	str | None	Helix, needs user:read:email
description	str | None	Helix bio
profile_image_url / offline_image_url	str | None	Helix
view_count	int | None	Helix
type	str | None	staff / admin / ...
broadcaster_type	str | None	partner / affiliate / empty
created_at	datetime | None	Helix

TokenSet

What authenticate() returns: tokens: TwitchTokens, user: TwitchUser, validated_at: datetime.

---

🔗 API Reference

TwitchAuth(client_id, client_secret, *, redirect_uri, scopes, state_store, http_client, state_ttl, enforce_https)

FastAPI integration. Requires the [fastapi] extra.

Parameter	Type	Default	Description
client_id	str	-	Twitch application client id
client_secret	str	-	Twitch application client secret
redirect_uri	str	-	absolute URL of the /auth/callback route (HTTPS-only, except localhost)
scopes	Sequence[str] | None	("user:read:email",)	scopes for the authorize URL
state_store	StateStore | None	SessionStateStore	state/PKCE backend
http_client	httpx.AsyncClient | None	None	inject your own client (never closed)
state_ttl	timedelta	10 minutes	CSRF state lifetime
enforce_https	bool	True	reject non-https redirect_uri, except localhost

Methods:

• generate_state() → str - fresh CSRF token
• generate_pkce() → tuple[str, str] - fresh (code_verifier, code_challenge)
• await login_url(request, *, scopes=None, state=None, code_verifier=None, code_challenge=None, extra=None) → str - stashes state + verifier, returns Twitch authorize URL
• await redirect_to_login(request, *, scopes=None, extra=None) → RedirectResponse - 307 to Twitch
• await authenticate(request, *, expected_state) → TokenSet - validates callback + exchanges code (raises HTTPException 400/502)
• await fetch_user(*, access_token, user_id=None, login=None) → TwitchUser
• await fetch_users(*, access_token, ids=None, logins=None) → list[TwitchUser] - batches to max 100 ids/logins per request
• await aclose() / async with twitch: - lifecycle

TwitchOAuthClient(client_id, client_secret, *, http_client, timeout)

Framework-agnostic core. No web-framework dependency.

Parameter	Type	Default	Description
client_id	str	-	Twitch application client id
client_secret	str	-	Twitch application client secret
http_client	httpx.AsyncClient | None	None	inject your own client (never closed)
timeout	float	10.0	request timeout

Methods:

• generate_state() → str (static)
• generate_pkce_pair() → tuple[str, str] (static)
• login_url(*, redirect_uri, state, code_challenge, scopes) → str
• await exchange_code(*, code, redirect_uri, code_verifier) → TwitchTokens
• await refresh_tokens(*, refresh_token) → TwitchTokens
• await validate(access_token) → TwitchUser
• await revoke(*, token) → None
• await fetch_user(*, access_token, user_id=None, login=None) → TwitchUser
• await fetch_users(*, access_token, ids=None, logins=None) → list[TwitchUser]
• await complete_login(*, code, redirect_uri, code_verifier) → TokenSet - exchange + validate + best-effort profile merge
• await aclose()

Exceptions

Exception	Description
TwitchAuthError	base class for all library errors
TwitchAPIError	non-2xx response (carries .status_code)
OAuthError	RFC 6749 error envelope (invalid_grant, ...); carries .code, .description
InvalidStateError	missing / expired / replayed CSRF state
InvalidPKCEError	missing or mismatched code_verifier
ScopeMissingError	requested data needs a scope the token lacks
StateStoreError	state backend failure / misconfiguration
ConfigurationError	bad library config (e.g. http:// redirect_uri)

---

📜 license

MIT