from __future__ import annotations
import asyncio
import email.utils
import logging
import time
from typing import Any, ClassVar, Dict, Mapping, Optional, Union
from urllib.parse import quote as _uriquote
import aiohttp
from .errors import (
BadRequest,
HTTPException,
InternalServerError,
InvalidGamemode,
NotFound,
RateLimited,
)
_log = logging.getLogger(__name__)
DEFAULT_MAX_RETRIES: int = 5
RETRY_STATUSES = {
500,
502,
504,
524,
}
async def json_or_text(
response: aiohttp.ClientResponse,
) -> Union[Dict[str, Any], str]:
text = await response.text(
encoding="utf-8"
)
content_type = response.headers.get(
"content-type",
""
)
if "application/json" in content_type:
return await response.json(
encoding="utf-8"
)
return text
[docs]
class Route:
BASE: ClassVar[str] = "https://mctiers.com/api/v2"
[docs]
def __init__(
self,
method: str,
path: str,
**parameters: Any,
) -> None:
self.method: str = method
self.path: str = path
url = self.BASE + path
if parameters:
url = url.format_map(
{
key: _uriquote(value, safe="")
if isinstance(value, str)
else value
for key, value in parameters.items()
}
)
self.url: str = url
@property
def key(self) -> str:
return f"{self.method} {self.path}"
class RouteRateLimit:
__slots__ = (
"lock",
"reset_at",
)
def __init__(self) -> None:
self.lock: asyncio.Lock = asyncio.Lock()
self.reset_at: float = 0.0
async def __aenter__(self) -> RouteRateLimit:
await self.lock.acquire()
delay = self.reset_at - time.monotonic()
if delay > 0:
_log.debug(
"Pre-emptively sleeping for %.2f seconds due to rate limit",
delay,
)
await asyncio.sleep(delay)
return self
async def __aexit__(
self,
exc_type: Any,
exc_value: Any,
traceback: Any,
) -> None:
self.lock.release()
def update(
self,
response: aiohttp.ClientResponse,
) -> None:
remaining = response.headers.get("X-RateLimit-Remaining")
if remaining != "0":
return
retry_after = parse_retry_after(response, None)
if retry_after is not None and retry_after > 0:
self.reset_at = max(
self.reset_at,
time.monotonic() + retry_after,
)
def parse_retry_after(
response: aiohttp.ClientResponse,
data: Any,
) -> Optional[float]:
if isinstance(data, dict):
value = data.get("retry_after")
if isinstance(value, (int, float)):
return max(float(value), 0.0)
if isinstance(value, str):
try:
return max(float(value), 0.0)
except ValueError:
pass
for header in (
"Retry-After",
"X-RateLimit-Reset-After",
):
value = response.headers.get(header)
if value is None:
continue
try:
return max(float(value), 0.0)
except ValueError:
try:
parsed = email.utils.parsedate_to_datetime(value)
except (TypeError, ValueError):
parsed = None
if parsed is None:
continue
return max(parsed.timestamp() - time.time(), 0.0)
return None
[docs]
class HTTPClient:
"""
Low-level async HTTP wrapper for the MCTiers API
"""
[docs]
def __init__(
self,
loop: Any,
connector: Optional[aiohttp.BaseConnector] = None,
*,
proxy_url: Optional[str] = None,
proxy_auth: Optional[aiohttp.BasicAuth] = None,
session: Optional[aiohttp.ClientSession] = None,
user_agent: Optional[str] = None,
max_retries: int = DEFAULT_MAX_RETRIES,
max_ratelimit_timeout: Optional[float] = None,
) -> None:
if max_retries < 1:
raise ValueError(
"max_retries must be greater than or equal to 1"
)
self.loop: Any = loop
self.connector: Optional[aiohttp.BaseConnector] = connector
self.proxy_url: Optional[str] = proxy_url
self.proxy_auth: Optional[aiohttp.BasicAuth] = proxy_auth
self.user_agent: str = user_agent or "mctiers.py/0.1.0"
self.max_retries: int = max_retries
self.max_ratelimit_timeout: Optional[float] = max_ratelimit_timeout
self._ratelimits: Dict[str, RouteRateLimit] = {}
self.__session: Optional[aiohttp.ClientSession] = session
self.__session_owner: bool = session is None
[docs]
async def static_login(self) -> None:
if self.__session is None or self.__session.closed:
self.__session = aiohttp.ClientSession(
connector=self.connector,
headers={
"User-Agent": self.user_agent
}
)
self.__session_owner = True
[docs]
async def close(self) -> None:
if (
self.__session_owner
and self.__session is not None
and not self.__session.closed
):
await self.__session.close()
[docs]
async def request(
self,
route: Route,
*,
params: Optional[Mapping[str, Any]] = None,
) -> Any:
if self.__session is None or self.__session.closed:
await self.static_login()
assert self.__session is not None
request_params = {
key: value
for key, value in (params or {}).items()
if value is not None
}
_log.debug(
"%s %s with %s",
route.method,
route.url,
request_params,
)
ratelimit = self._ratelimits.setdefault(
route.key,
RouteRateLimit(),
)
response: Optional[aiohttp.ClientResponse] = None
data: Any = None
async with ratelimit:
for tries in range(self.max_retries):
try:
async with self.__session.request(
route.method,
route.url,
params=request_params,
proxy=self.proxy_url,
proxy_auth=self.proxy_auth,
) as response:
data = await json_or_text(response)
ratelimit.update(response)
_log.debug(
"%s %s returned %s",
route.method,
route.url,
response.status,
)
if 200 <= response.status < 300:
return data
if response.status == 429:
retry_after = parse_retry_after(
response,
data,
)
if retry_after is None:
retry_after = 1 + tries * 2
if (
self.max_ratelimit_timeout is not None
and retry_after > self.max_ratelimit_timeout
):
raise RateLimited(
response,
data,
retry_after=retry_after,
)
if tries >= self.max_retries - 1:
break
_log.warning(
"%s %s was rate limited. Retrying in %.2f seconds.",
route.method,
route.url,
retry_after,
)
ratelimit.reset_at = max(
ratelimit.reset_at,
time.monotonic() + retry_after,
)
await asyncio.sleep(retry_after)
continue
if response.status in RETRY_STATUSES:
if tries >= self.max_retries - 1:
break
await asyncio.sleep(1 + tries * 2)
continue
self._raise_for_status(
response,
data,
)
except OSError as exc:
errno = getattr(exc, "errno", None)
if tries < self.max_retries - 1 and errno in (54, 10054):
await asyncio.sleep(1 + tries * 2)
continue
raise
if response is not None:
self._raise_for_status(
response,
data,
)
raise RuntimeError(
"Unreachable code in HTTP handling"
)
def _raise_for_status(
self,
response: aiohttp.ClientResponse,
data: Any,
) -> None:
if response.status == 400:
if isinstance(data, dict) and data.get("code") == "error.invalid_gamemode":
raise InvalidGamemode(
response,
data
)
raise BadRequest(
response,
data
)
if response.status == 404:
raise NotFound(
response,
data
)
if response.status == 429:
raise RateLimited(
response,
data,
retry_after=parse_retry_after(
response,
data,
),
)
if response.status >= 500:
raise InternalServerError(
response,
data
)
raise HTTPException(
response,
data
)
[docs]
def get_gamemodes(self) -> Any:
return self.request(
Route(
"GET",
"/mode/list"
)
)
[docs]
def get_overall_rankings(
self,
*,
count: int,
from_: int = 0,
) -> Any:
return self.request(
Route(
"GET",
"/mode/overall"
),
params={
"count": count,
"from": from_,
}
)
[docs]
def get_gamemode_rankings(
self,
gamemode: str,
*,
count: int,
from_: int = 0,
retired: bool = False,
) -> Any:
path = "/mode/{gamemode}/retired" if retired else "/mode/{gamemode}"
return self.request(
Route(
"GET",
path,
gamemode=gamemode
),
params={
"count": count,
"from": from_,
}
)
[docs]
def get_player_profile(
self,
uuid: str,
*,
tests: bool = False,
badges: bool = False,
) -> Any:
return self.request(
Route(
"GET",
"/profile/{uuid}",
uuid=uuid
),
params=self._profile_params(
tests=tests,
badges=badges
)
)
[docs]
def get_player_rankings(
self,
uuid: str,
*,
tests: bool = False,
badges: bool = False,
) -> Any:
return self.request(
Route(
"GET",
"/profile/{uuid}/rankings",
uuid=uuid
),
params=self._profile_params(
tests=tests,
badges=badges
)
)
[docs]
def get_player_profile_by_name(
self,
name: str,
*,
tests: bool = False,
badges: bool = False,
) -> Any:
return self.request(
Route(
"GET",
"/profile/by-name/{name}",
name=name
),
params=self._profile_params(
tests=tests,
badges=badges
)
)
[docs]
def get_player_profile_by_discord(
self,
discord_id: int,
*,
tests: bool = False,
badges: bool = False,
) -> Any:
return self.request(
Route(
"GET",
"/profile/by-discord/{discord_id}",
discord_id=discord_id
),
params=self._profile_params(
tests=tests,
badges=badges
)
)
[docs]
def get_tester_history(
self,
uuid: str,
*,
gamemode: Optional[str] = None,
count: int,
from_: int = 0,
) -> Any:
return self.request(
Route(
"GET",
"/tests/{uuid}",
uuid=uuid
),
params={
"gamemode": gamemode,
"count": count,
"from": from_,
}
)
[docs]
def get_recent_tests(
self,
*,
count: int,
gamemode: Optional[str] = None,
high: bool = False,
) -> Any:
path = "/tests/recent/high" if high else "/tests/recent"
return self.request(
Route(
"GET",
path
),
params={
"gamemode": gamemode,
"count": count,
}
)
@staticmethod
def _profile_params(
*,
tests: bool,
badges: bool,
) -> Dict[str, Any]:
params: Dict[str, Any] = {}
if tests:
params["tests"] = ""
if badges:
params["badges"] = ""
return params