Релиз

src/twitchclient/auth.py

@@ -0,0 +1,126 @@
+from urllib.parse import urlencode
+
+from aiohttpx import status as st
+from aiohttpx.client import AioHTTPXClient
+
+from . import schema as s
+from . import scopes
+
+
+class TwitchAuthClient(AioHTTPXClient):
+    def __init__(
+        self,
+        redis_url: str,
+        client_id: str,
+        client_secret: str,
+        redirect_uri: str,
+    ):
+        self.base_uri = 'https://id.twitch.tv/oauth2'
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.redirect_uri = redirect_uri
+
+        super().__init__(
+            base_url=self.base_uri,
+            redis_url=redis_url,
+            key='twitch',
+            limit=700,
+            logger='Twitch Auth',
+        )
+
+    async def create_authorization_code_grant_flow_url(
+        self,
+        scope: list[scopes.Any],
+        *,
+        force_verify: bool = False,
+        state: str | None = None,
+    ):
+        url = 'https://id.twitch.tv/oauth2/authorize?'
+        query = self.clean_dict(
+            {
+                'client_id': self.client_id,
+                'redirect_uri': self.redirect_uri,
+                'response_type': 'code',
+                'scope': ' '.join(scope),
+                'state': state,
+                'force_verify': force_verify,
+            }
+        )
+
+        return url + urlencode(query)
+
+    async def app_access_token(self):
+        req = await self.post(
+            '/token',
+            json={
+                'client_id': self.client_id,
+                'client_secret': self.client_secret,
+                'grant_type': 'client_credentials',
+            },
+        )
+
+        match req.status_code:
+            case st.OK:
+                return s.AppAccessToken.model_validate(req.json())
+
+            case _:
+                raise s.InternalError(req.status_code, 'Internal Server Error')
+
+    async def user_access_token(self, code: str):
+        req = await self.post(
+            '/token',
+            json={
+                'client_id': self.client_id,
+                'client_secret': self.client_secret,
+                'redirect_uri': self.redirect_uri,
+                'grant_type': 'authorization_code',
+                'code': code,
+            },
+        )
+
+        match req.status_code:
+            case st.OK:
+                return s.UserAccessToken.model_validate(req.json())
+
+            case st.BAD_REQUEST:
+                return None
+
+            case _:
+                raise s.InternalError(req.status_code, 'Internal Server Error')
+
+    async def validate_access_token(self, access_token: str):
+        req = await self.get(
+            '/validate',
+            headers={'Authorization': f'OAuth {access_token}'},
+        )
+
+        match req.status_code:
+            case st.OK:
+                return s.AccessTokenValidation.model_validate(req.json())
+
+            case st.UNAUTHORIZED:
+                return None
+
+            case _:
+                raise s.InternalError(req.status_code, 'Internal Server Error')
+
+    async def refresh_access_token(self, refresh_token: str):
+        req = await self.post(
+            '/token',
+            json={
+                'client_id': self.client_id,
+                'client_secret': self.client_secret,
+                'grant_type': 'refresh_token',
+                'refresh_token': refresh_token,
+            },
+        )
+
+        match req.status_code:
+            case st.OK:
+                return s.UserAccessToken.model_validate(req.json())
+
+            case st.BAD_REQUEST:
+                return None
+
+            case _:
+                raise s.InternalError(req.status_code, 'Internal Server Error')