Add auth methods

twitchio/authentication/oauth.py

@@ -23,22 +23,29 @@
 """
 from __future__ import annotations
 
+import urllib.parse
 from typing import TYPE_CHECKING
 
 from ..http import HTTPClient
 from .payloads import *
 
 
 if TYPE_CHECKING:
-    from ..types_.responses import RefreshTokenResponse, ValidateTokenResponse
+    from ..types_.responses import (
+        ClientCredentialsResponse,
+        RefreshTokenResponse,
+        UserTokenResponse,
+        ValidateTokenResponse,
+    )
 
 
 class OAuth(HTTPClient):
-    def __init__(self, *, client_id: str, client_secret: str) -> None:
+    def __init__(self, *, client_id: str, client_secret: str, redirect_uri: str | None) -> None:
         super().__init__()
 
         self.client_id = client_id
         self.client_secret = client_secret
+        self.redirect_uri = redirect_uri
 
     async def validate_token(self, token: str, /) -> ValidateTokenPayload:
         token = token.removeprefix("Bearer ").removeprefix("OAuth ")
@@ -64,5 +71,64 @@ async def refresh_token(self, refresh_token: str, /) -> RefreshTokenPayload:
 
         return RefreshTokenPayload(data)
 
-    async def revoke_token(self, token: str, /) -> ...:
-        raise NotImplementedError
+    async def user_access_token(self, code: str, /) -> UserTokenPayload:
+        if not self.redirect_uri:
+            raise ValueError("Missing redirect_uri")
+
+        headers: dict[str, str] = {"Content-Type": "application/x-www-form-urlencoded"}
+
+        params: dict[str, str] = {
+            "client_id": self.client_id,
+            "client_secret": self.client_secret,
+            "code": code,
+            "grant_type": "authorization_code",
+            "redirect_uri": self.redirect_uri,
+            # "scope": " ".join(SCOPES), #TODO
+            # "state": #TODO
+        }
+
+        data: UserTokenResponse = await self.request_json(
+            "POST", "/oauth2/token", use_id=True, headers=headers, params=params
+        )
+
+        return UserTokenPayload(data)
+
+    async def revoke_token(self, token: str, /) -> None:
+        headers: dict[str, str] = {"Content-Type": "application/x-www-form-urlencoded"}
+
+        params: dict[str, str] = {
+            "client_id": self.client_id,
+            "token": token,
+        }
+
+        await self.request_json("POST", "/oauth2/revoke", use_id=True, headers=headers, params=params)
+
+    async def client_credentials_token(self) -> ClientCredentialsPayload:
+        headers: dict[str, str] = {"Content-Type": "application/x-www-form-urlencoded"}
+
+        params = {
+            "client_id": self.client_id,
+            "client_secret": self.client_secret,
+            "grant_type": "client_credentials",
+        }
+
+        data: ClientCredentialsResponse = await self.request_json(
+            "POST", "/oauth2/token", use_id=True, headers=headers, params=params
+        )
+
+        return ClientCredentialsPayload(data)
+
+    def get_authorization_url(self, scopes: list[str], state: str = "") -> str:
+        if not self.redirect_uri:
+            raise ValueError("Missing redirect_uri")
+
+        base_url = "https://id.twitch.tv/oauth2/authorize"
+        params = {
+            "client_id": self.client_id,
+            "redirect_uri": urllib.parse.quote(self.redirect_uri),
+            "response_type": "code",
+            "scope": "+".join(urllib.parse.quote(scope) for scope in scopes),
+            "state": state,
+        }
+        query_string = "&".join(f"{key}={value}" for key, value in params.items())
+        return f"{base_url}?{query_string}"

twitchio/authentication/payloads.py

@@ -34,6 +34,8 @@
 __all__ = (
     "RefreshTokenPayload",
     "ValidateTokenPayload",
+    "ClientCredentialsPayload",
+    "UserTokenPayload",
 )
 
 
@@ -77,3 +79,27 @@ def __init__(self, raw: ValidateTokenResponse, /) -> None:
         self.scopes: list[str] = raw["scopes"]
         self.user_id: str = raw["user_id"]
         self.expires_in: int = raw["expires_in"]
+
+
+class UserTokenPayload(BasePayload):
+    __slots__ = ("access_token", "refresh_token", "expires_in", "scope", "token_type")
+
+    def __init__(self, raw: UserTokenResponse, /) -> None:
+        super().__init__(raw)
+
+        self.access_token: str = raw["access_token"]
+        self.refresh_token: str = raw["refresh_token"]
+        self.expires_in: int = raw["expires_in"]
+        self.scope: str | list[str] = raw["scope"]
+        self.token_type: str = raw["token_type"]
+
+
+class ClientCredentialsPayload(BasePayload):
+    __slots__ = ("access_token", "expires_in", "token_type")
+
+    def __init__(self, raw: ClientCredentialsResponse, /) -> None:
+        super().__init__(raw)
+
+        self.access_token: str = raw["access_token"]
+        self.expires_in: int = raw["expires_in"]
+        self.token_type: str = raw["token_type"]

twitchio/types_/responses.py

@@ -27,7 +27,9 @@
 __all__ = (
     "RefreshTokenResponse",
     "ValidateTokenResponse",
+    "ClientCredentialsResponse",
     "OAuthResponses",
+    "UserTokenResponse",
 )
 
 
@@ -38,6 +40,13 @@ class RefreshTokenResponse(TypedDict):
     scope: str | list[str]
     token_type: str
 
+class UserTokenResponse(TypedDict):
+    access_token: str
+    refresh_token: str
+    expires_in: int
+    scope: str | list[str]
+    token_type: str
+
 
 class ValidateTokenResponse(TypedDict):
     client_id: str
@@ -46,5 +55,10 @@ class ValidateTokenResponse(TypedDict):
     user_id: str
     expires_in: int
 
+class ClientCredentialsResponse(TypedDict):
+    access_token: str
+    expires_in: int
+    token_type: str
+
 
-OAuthResponses: TypeAlias = RefreshTokenResponse | ValidateTokenResponse
+OAuthResponses: TypeAlias = RefreshTokenResponse | ValidateTokenResponse | ClientCredentialsResponse | UserTokenResponse