155 lines
4.3 KiB
Python
155 lines
4.3 KiB
Python
from typing import Literal
|
|
|
|
from aiohttpx import status as st
|
|
from aiohttpx.client import AioHTTPXClient
|
|
|
|
from . import schema as s
|
|
|
|
|
|
class osuAPIClient(AioHTTPXClient):
|
|
def __init__(
|
|
self,
|
|
redis_url: str,
|
|
client_id: str,
|
|
client_secret: str,
|
|
redirect_uri: str,
|
|
):
|
|
self.base_uri = 'https://osu.ppy.sh/api/v2'
|
|
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='osu',
|
|
limit=1200,
|
|
logger='osu! API',
|
|
)
|
|
|
|
async def get_beatmap(self, access_token: str, beatmap: int):
|
|
req = await self.get(
|
|
f'/beatmaps/{beatmap}',
|
|
headers=self.clean_dict(
|
|
{
|
|
'Authorization': f'Bearer {access_token}',
|
|
}
|
|
),
|
|
)
|
|
|
|
match req.status_code:
|
|
case st.OK:
|
|
return s.BeatmapExtended.model_validate(req.json())
|
|
|
|
case st.NOT_FOUND:
|
|
raise s.Error(404, 'Beatmap not found')
|
|
|
|
case _:
|
|
raise s.Error(500, 'Internal Server Error')
|
|
|
|
async def get_score(
|
|
self,
|
|
access_token: str,
|
|
score_id: int,
|
|
mode: Literal['fruits', 'mania', 'osu', 'taiko'] | None = None,
|
|
):
|
|
url = f'/scores/{mode}/{score_id}' if mode else f'/scores/{score_id}'
|
|
|
|
req = await self.get(
|
|
url,
|
|
headers=self.clean_dict(
|
|
{
|
|
'Authorization': f'Bearer {access_token}',
|
|
}
|
|
),
|
|
)
|
|
|
|
match req.status_code:
|
|
case st.OK:
|
|
return s.Score.model_validate(req.json())
|
|
|
|
case st.NOT_FOUND:
|
|
raise s.Error(404, 'Score not found')
|
|
|
|
case _:
|
|
raise s.Error(500, 'Internal Server Error')
|
|
|
|
async def get_user_scores(
|
|
self,
|
|
access_token: str,
|
|
user_id: int,
|
|
score_type: Literal['best', 'recent', 'firsts'],
|
|
legacy_only: Literal[0, 1] = 0,
|
|
include_fails: Literal[0, 1] = 0,
|
|
mode: Literal['fruits', 'mania', 'osu', 'taiko'] | None = None,
|
|
limit: int | None = None,
|
|
offset: str | None = None,
|
|
):
|
|
req = await self.get(
|
|
f'/users/{user_id}/scores/{score_type}',
|
|
params=self.clean_dict(
|
|
{
|
|
'legacy_only': legacy_only,
|
|
'include_fails': include_fails,
|
|
'mode': mode,
|
|
'limit': limit,
|
|
'offset': offset,
|
|
}
|
|
),
|
|
headers=self.clean_dict(
|
|
{
|
|
'Authorization': f'Bearer {access_token}',
|
|
}
|
|
),
|
|
)
|
|
|
|
match req.status_code:
|
|
case st.OK:
|
|
return [s.Score.model_validate(score) for score in req.json()]
|
|
|
|
case st.NOT_FOUND:
|
|
raise s.Error(req.status_code, 'Not Found')
|
|
|
|
case st.UNAUTHORIZED:
|
|
raise s.Error(req.status_code, 'Unauthorized')
|
|
|
|
case _:
|
|
self.logger.error(req.text)
|
|
raise s.Error(500, 'Internal Server Error')
|
|
|
|
async def get_user(
|
|
self,
|
|
access_token: str,
|
|
user_id: int,
|
|
mode: Literal['fruits', 'mania', 'osu', 'taiko'] | None = None,
|
|
key: Literal['username', 'id'] | None = None,
|
|
):
|
|
url = f'/users/{user_id}'
|
|
|
|
if mode:
|
|
url += f'/{mode}'
|
|
|
|
req = await self.get(
|
|
url,
|
|
params=self.clean_dict({'key': key}),
|
|
headers=self.clean_dict(
|
|
{
|
|
'Authorization': f'Bearer {access_token}',
|
|
}
|
|
),
|
|
)
|
|
|
|
match req.status_code:
|
|
case st.OK:
|
|
return s.GetUser.model_validate(req.json())
|
|
|
|
case st.NOT_FOUND:
|
|
raise s.Error(req.status_code, 'Not Found')
|
|
|
|
case st.UNAUTHORIZED:
|
|
raise s.Error(req.status_code, 'Unauthorized')
|
|
|
|
case _:
|
|
self.logger.error(req.text)
|
|
raise s.Error(500, 'Internal Server Error')
|