from __future__ import annotations
import asyncio
from typing import Any, Dict, List, Optional, Type, TYPE_CHECKING
import aiohttp
from .http import HTTPClient
from .models import (
Gamemode,
GamemodePlayer,
OverallPlayer,
Player,
Ranking,
Test,
)
if TYPE_CHECKING:
from types import TracebackType
from typing_extensions import Self
class _LoopSentinel:
__slots__ = ()
def __getattr__(
self,
attr: str,
) -> None:
raise AttributeError(
"loop attribute cannot be accessed in non-async contexts"
)
_loop: Any = _LoopSentinel()
[docs]
class Client:
"""
High-level async client for the MCTiers API
"""
[docs]
def __init__(
self,
*,
proxy_url: Optional[str] = None,
proxy_auth: Optional[aiohttp.BasicAuth] = None,
connector: Optional[aiohttp.BaseConnector] = None,
session: Optional[aiohttp.ClientSession] = None,
user_agent: Optional[str] = None,
max_retries: int = 5,
max_ratelimit_timeout: Optional[float] = None,
) -> None:
self.loop: Any = _loop
self.http: HTTPClient = HTTPClient(
self.loop,
connector,
proxy_url=proxy_url,
proxy_auth=proxy_auth,
session=session,
user_agent=user_agent,
max_retries=max_retries,
max_ratelimit_timeout=max_ratelimit_timeout,
)
[docs]
async def __aenter__(self) -> Self:
await self._async_setup_hook()
return self
[docs]
async def __aexit__(
self,
exc_type: Optional[Type[BaseException]],
exc_value: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
await self.close()
async def _async_setup_hook(self) -> None:
self.loop = asyncio.get_running_loop()
self.http.loop = self.loop
await self.http.static_login()
[docs]
async def close(self) -> None:
await self.http.close()
[docs]
async def list_gamemodes(self) -> Dict[str, Gamemode]:
data = await self.http.get_gamemodes()
return {
slug: Gamemode(
slug=slug,
data=payload
)
for slug, payload in data.items()
}
[docs]
async def fetch_overall_rankings(
self,
*,
count: int,
from_: int = 0,
) -> List[OverallPlayer]:
self._validate_page(
count=count,
from_=from_,
max_count=50
)
data = await self.http.get_overall_rankings(
count=count,
from_=from_
)
return [
OverallPlayer(player)
for player in data
]
[docs]
async def fetch_gamemode_rankings(
self,
gamemode: str,
*,
count: int,
from_: int = 0,
) -> Dict[int, List[GamemodePlayer]]:
return await self._fetch_gamemode_rankings(
gamemode,
count=count,
from_=from_,
retired=False
)
[docs]
async def fetch_retired_gamemode_rankings(
self,
gamemode: str,
*,
count: int,
from_: int = 0,
) -> Dict[int, List[GamemodePlayer]]:
return await self._fetch_gamemode_rankings(
gamemode,
count=count,
from_=from_,
retired=True
)
[docs]
async def fetch_player_profile(
self,
uuid: str,
*,
tests: bool = False,
badges: bool = False,
) -> Player:
data = await self.http.get_player_profile(
uuid,
tests=tests,
badges=badges
)
return Player(data)
[docs]
async def fetch_player_rankings(
self,
uuid: str,
*,
tests: bool = False,
badges: bool = False,
) -> Dict[str, Ranking]:
data = await self.http.get_player_rankings(
uuid,
tests=tests,
badges=badges
)
return {
gamemode: Ranking(
gamemode=gamemode,
data=ranking
)
for gamemode, ranking in data.items()
}
[docs]
async def fetch_player_profile_by_name(
self,
name: str,
*,
tests: bool = False,
badges: bool = False,
) -> Player:
data = await self.http.get_player_profile_by_name(
name,
tests=tests,
badges=badges
)
return Player(data)
[docs]
async def fetch_player_profile_by_discord(
self,
discord_id: int,
*,
tests: bool = False,
badges: bool = False,
) -> Player:
data = await self.http.get_player_profile_by_discord(
discord_id,
tests=tests,
badges=badges
)
return Player(data)
[docs]
async def fetch_tester_history(
self,
uuid: str,
*,
gamemode: Optional[str] = None,
count: int,
from_: int = 0,
) -> List[Test]:
self._validate_page(
count=count,
from_=from_,
max_count=50
)
data = await self.http.get_tester_history(
uuid,
gamemode=gamemode,
count=count,
from_=from_
)
return [
Test(test)
for test in data
]
[docs]
async def fetch_recent_tests(
self,
*,
count: int,
gamemode: Optional[str] = None,
) -> List[Test]:
return await self._fetch_recent_tests(
count=count,
gamemode=gamemode,
high=False
)
[docs]
async def fetch_recent_high_tests(
self,
*,
count: int,
gamemode: Optional[str] = None,
) -> List[Test]:
return await self._fetch_recent_tests(
count=count,
gamemode=gamemode,
high=True
)
async def _fetch_gamemode_rankings(
self,
gamemode: str,
*,
count: int,
from_: int,
retired: bool,
) -> Dict[int, List[GamemodePlayer]]:
self._validate_page(
count=count,
from_=from_,
max_count=50
)
data = await self.http.get_gamemode_rankings(
gamemode,
count=count,
from_=from_,
retired=retired
)
return {
int(tier): [
GamemodePlayer(player)
for player in players
]
for tier, players in data.items()
}
async def _fetch_recent_tests(
self,
*,
count: int,
gamemode: Optional[str],
high: bool,
) -> List[Test]:
self._validate_count(
count=count,
max_count=20
)
data = await self.http.get_recent_tests(
count=count,
gamemode=gamemode,
high=high
)
return [
Test(test)
for test in data
]
@staticmethod
def _validate_count(
*,
count: int,
max_count: int,
) -> None:
if count < 0 or count > max_count:
raise ValueError(
f"count must be between 0 and {max_count}"
)
@classmethod
def _validate_page(
cls,
*,
count: int,
from_: int,
max_count: int,
) -> None:
cls._validate_count(
count=count,
max_count=max_count
)
if from_ < 0:
raise ValueError(
"from_ must be greater than or equal to 0"
)