127 lines
3.5 KiB
Python
127 lines
3.5 KiB
Python
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')
|