diff --git a/src/pychasing/client.py b/src/pychasing/client.py index 9277b02..bd5a9e8 100644 --- a/src/pychasing/client.py +++ b/src/pychasing/client.py @@ -16,7 +16,6 @@ import urllib.parse import rlim import io -import re try: from typing import Literal @@ -30,13 +29,10 @@ ) -cont_pat = re.compile(r"(?<=\after=)[^\&]*") - - def _print_error(response: requests.Response) -> None: """Print out an error code from a `requests.Response` if an HTTP error is encountered. - + """ error_side = ("Client" if 400 <= response.status_code < 500 else "Server" if 500 <= response.status_code < 600 else None) @@ -55,21 +51,35 @@ def _print_error(response: requests.Response) -> None: error_description = "" if response_json and "error" in response_json: error_description = "(" + response_json["error"] + ") " - + print(f"\033[93m{response.status_code} {error_side} Error: {reason} " f"{error_description}for url: {response.url}\033[0m") def p(v): """Return `v` if `v` is `...` or a `str`, else return `v.value`. - + """ return v if v == ... or isinstance(v, str) else v.value +def _extract_query_parameter(url: str, parameter: str): + parsed_params = urllib.parse.parse_qs(urllib.parse.urlparse(url).query) + return parsed_params[parameter][0] + + +def _extract_after_query_parameter(next_url: str = ...): + if next_url == ...: + return next_url + try: + return _extract_query_parameter(next_url, "after") + except Exception as e: + raise ValueError(f"'next_url' string has an unknown structure {e}") + + class Client: """The main class used to interact with the Ballchasing API. - + """ def __init__(self, token: str, auto_rate_limit: bool = True, patreon_tier: Union[str, enums.PatreonTier] = enums.PatreonTier.none, @@ -95,12 +105,12 @@ def __init__(self, token: str, auto_rate_limit: bool = True, patreon_tier = enums.PatreonTier[patreon_tier] except KeyError as exc: raise ValueError(f"{patreon_tier!r} is not a valid PatreonTier") from exc - + if auto_rate_limit: for k, v in patreon_tier.value.items(): rlim.set_rate_limiter(getattr(self, k.name), rlim.RateLimiter(*v, safestart=rate_limit_safe_start)) - + def ping(self, *, print_error: bool = True) -> requests.Response: """Ping the https://ballchasing.com servers. @@ -109,12 +119,12 @@ def ping(self, *, print_error: bool = True) -> requests.Response: print_error : bool, optional, default=True Prints an error message (that contains information about the error) if the request resulted in an HTTP error (i.e. status codes 400 through 599). - + Returns ------- requests.Response The `requests.Response` object returned from the HTTP request. - + """ # prepare URL prepped_url = httpprep.URL(protocol="https", domain="ballchasing", top_level_domain="com", @@ -123,7 +133,7 @@ def ping(self, *, print_error: bool = True) -> requests.Response: # prepare headers prepped_headers = httpprep.Headers() prepped_headers.Authorization = self._token - + # make request, print error, and return response response = requests.get(prepped_url.build(), headers=prepped_headers.format_dict()) if print_error: @@ -146,12 +156,12 @@ def upload_replay(self, file: io.BufferedReader, print_error : bool, optional, default=True Prints an error message (that contains information about the error) if the request resulted in an HTTP error (i.e. status codes 400 through 599). - + Returns ------- requests.Response The `requests.Response` object returned from the HTTP request. - + """ # prepare URL prepped_url = httpprep.URL(protocol="https", domain="ballchasing", top_level_domain="com", @@ -161,7 +171,7 @@ def upload_replay(self, file: io.BufferedReader, # prepare headers prepped_headers = httpprep.Headers() prepped_headers.Authorization = self._token - + # make request, print error, and return response response = requests.post(prepped_url.build(query_check=...), headers=prepped_headers.format_dict(), files={"file":file}) @@ -170,7 +180,7 @@ def upload_replay(self, file: io.BufferedReader, return response @rlim.placeholder - def list_replays(self, *, next: str = ..., title: str = ..., player_names: Iterable[str] = ..., + def list_replays(self, *, next_url: str = ..., title: str = ..., player_names: Iterable[str] = ..., player_ids: Iterable[Tuple[Union[enums.Platform, str], Union[int, str]]] = ..., playlists: Iterable[Union[enums.Playlist, str]] = ..., season: Union[str, enums.Season] = ..., @@ -190,7 +200,7 @@ def list_replays(self, *, next: str = ..., title: str = ..., player_names: Itera Parameters ---------- - next : str, optional + next_url : str, optional A continuation URL (which can be acquired with `.json()["next"]`). If defined, the original parameters are still required to get the expected result. @@ -239,7 +249,7 @@ def list_replays(self, *, next: str = ..., title: str = ..., player_names: Itera Only include replays played after a given date, formatted as an RFC3339 datetime string. count : int, optional, default=150 - The number of replays returned. Must be between 1 and 200 + The number of replays returned. Must be between 1 and 200 (inclusive) if defined. sort_by : str or ReplaySortBy, optional, default= ReplaySortBy.upload_date @@ -249,16 +259,16 @@ def list_replays(self, *, next: str = ..., title: str = ..., player_names: Itera print_error : bool, optional, default=True Prints an error message (that contains information about the error) if the request resulted in an HTTP error (i.e. status codes 400 through 599). - + Returns ------- requests.Response The `requests.Response` object returned from the HTTP request. - + Raises: ValueError: `count` is defined and is less than 0 or greater than 200. - + """ if count != ... and 1 > count > 200: raise ValueError("\"count\" must be between 1 and 200") @@ -267,13 +277,6 @@ def list_replays(self, *, next: str = ..., title: str = ..., player_names: Itera prepped_headers = httpprep.Headers() prepped_headers.Authorization = self._token - # prepare url - if next != ...: - try: - next = urllib.parse.unquote(re.search(r"(?<=after=)[^\&]*", next).group()) - except Exception: - raise ValueError("'next' string has an unknown structure") - prepped_url = httpprep.URL(protocol="https", domain="ballchasing", top_level_domain="com", path_segments=["api", "replays"]) prepped_url.components.queries[ @@ -295,7 +298,7 @@ def list_replays(self, *, next: str = ..., title: str = ..., player_names: Itera "sort-by", "sort-dir" ] = [ - next, + _extract_after_query_parameter(next_url), title, p(season), p(match_result), @@ -329,7 +332,7 @@ def list_replays(self, *, next: str = ..., title: str = ..., player_names: Itera if print_error: _print_error(response) return response - + @rlim.placeholder def get_replay(self, replay_id: str, *, print_error: bool = True) -> requests.Response: """Get more in-depth information for a specific replay. @@ -341,12 +344,12 @@ def get_replay(self, replay_id: str, *, print_error: bool = True) -> requests.Re print_error : bool, optional, default=True Prints an error message (that contains information about the error) if the request resulted in an HTTP error (i.e. status codes 400 through 599). - + Returns ------- requests.Response The `requests.Response` object returned from the HTTP request. - + """ # prepare url prepped_url = httpprep.URL(protocol="https", domain="ballchasing", top_level_domain="com", @@ -361,7 +364,7 @@ def get_replay(self, replay_id: str, *, print_error: bool = True) -> requests.Re if print_error: _print_error(response) return response - + @rlim.placeholder def delete_replay(self, replay_id: str, *, print_error: bool = True) -> requests.Response: """Delete the given replay from https://ballchasing.com, so long as the @@ -374,12 +377,12 @@ def delete_replay(self, replay_id: str, *, print_error: bool = True) -> requests print_error : bool, optional, default=True Prints an error message (that contains information about the error) if the request resulted in an HTTP error (i.e. status codes 400 through 599). - + Returns ------- requests.Response The `requests.Response` object returned from the HTTP request. - + """ # prepare url prepped_url = httpprep.URL(protocol="https", domain="ballchasing", top_level_domain="com", @@ -394,7 +397,7 @@ def delete_replay(self, replay_id: str, *, print_error: bool = True) -> requests if print_error: _print_error(response) return response - + @rlim.placeholder def patch_replay(self, replay_id: str, *, title: str = ..., visibility: Union[str, enums.Visibility] = ..., group: str = ..., @@ -417,12 +420,12 @@ def patch_replay(self, replay_id: str, *, title: str = ..., print_error : bool, optional, default=True Prints an error message (that contains information about the error) if the request resulted in an HTTP error (i.e. status codes 400 through 599). - + Returns ------- requests.Response The `requests.Response` object returned from the HTTP request. - + """ # prepare url prepped_url = httpprep.URL(protocol="https", domain="ballchasing", top_level_domain="com", @@ -454,7 +457,7 @@ def download_replay(self, replay_id: str, *, print_error: bool = True) -> reques print_error : bool, optional, default=True Prints an error message (that contains information about the error) if the request resulted in an HTTP error (i.e. status codes 400 through 599). - + Warnings -------- Replay files can be rather large (up to around 1.5mb). The HTTP request @@ -465,7 +468,7 @@ def download_replay(self, replay_id: str, *, print_error: bool = True) -> reques ------- requests.Response The `requests.Response` object returned from the HTTP request. - + """ # prepare url prepped_url = httpprep.URL(protocol="https", domain="ballchasing", top_level_domain="com", @@ -509,7 +512,7 @@ def create_group(self, name: str, player_identification: Union[str, enums.Player ------- requests.Response The `requests.Response` object returned from the HTTP request. - + """ # prepare url prepped_url = httpprep.URL(protocol="https", domain="ballchasing", top_level_domain="com", @@ -532,9 +535,9 @@ def create_group(self, name: str, player_identification: Union[str, enums.Player if print_error: _print_error(response) return response - + @rlim.placeholder - def list_groups(self, *, next: str = ..., name: str = ..., creator: Union[str, int] = ..., + def list_groups(self, *, next_url: str = ..., name: str = ..., creator: Union[str, int] = ..., group: str = ..., created_before: Union[models.Date, str] = ..., created_after: Union[models.Date, str] = ..., count: int = ..., sort_by: Union[str, enums.GroupSortBy] = ..., @@ -545,9 +548,9 @@ def list_groups(self, *, next: str = ..., name: str = ..., creator: Union[str, i Parameters ---------- - next : str, optional + next_url : str, optional A continuation URL (which can be acquired with - `.json()["next"]`). If defined, the original parameters are + `.json()["next_url"]`). If defined, the original parameters are still required to get the expected result. name : str, optional Only include groups whose title contains the given text. @@ -580,26 +583,19 @@ def list_groups(self, *, next: str = ..., name: str = ..., creator: Union[str, i ------- requests.Response The `requests.Response` object returned from the HTTP request. - + Raises: ValueError: `count` is defined and is less than 0 or greater than 200. - + """ if count != ... and 1 > count > 200: raise ValueError("\"count\" must be between 1 and 200") - + # prepare headers prepped_headers = httpprep.Headers() prepped_headers.Authorization = self._token - # prepare url - if next != ...: - try: - next = urllib.parse.unquote(re.search(r"(?<=after=)[^\&]*", next).group()) - except Exception: - raise ValueError("'next' string has an unknown structure") - # prepare url prepped_url = httpprep.URL(protocol="https", domain="ballchasing", top_level_domain="com", path_segments=["api", "groups"]) @@ -614,13 +610,13 @@ def list_groups(self, *, next: str = ..., name: str = ..., creator: Union[str, i "sort-by", "sort-dir" ] = [ - next, + _extract_after_query_parameter(next_url), name, creator, group, created_before, created_after, - count, + count, p(sort_by), p(sort_dir) ] @@ -644,12 +640,12 @@ def get_group(self, group_id: str, *, print_error: bool = True) -> requests.Resp print_error : bool, optional, default=True Prints an error message (that contains information about the error) if the request resulted in an HTTP error (i.e. status codes 400 through 599). - + Returns ------- requests.Response The `requests.Response` object returned from the HTTP request. - + """ # prepare url prepped_url = httpprep.URL(protocol="https", domain="ballchasing", top_level_domain="com", @@ -664,7 +660,7 @@ def get_group(self, group_id: str, *, print_error: bool = True) -> requests.Resp if print_error: _print_error(response) return response - + def delete_group(self, group_id: str, *, print_error: bool = True) -> requests.Response: """Delete a specific group (and all children groups) from https://ballchasing.com, so long as it is owned by the token holder. @@ -676,12 +672,12 @@ def delete_group(self, group_id: str, *, print_error: bool = True) -> requests.R print_error : bool, optional, default=True Prints an error message (that contains information about the error) if the request resulted in an HTTP error (i.e. status codes 400 through 599). - + Returns ------- requests.Response The `requests.Response` object returned from the HTTP request. - + """ # prepare url prepped_url = httpprep.URL(protocol="https", domain="ballchasing", top_level_domain="com", @@ -696,7 +692,7 @@ def delete_group(self, group_id: str, *, print_error: bool = True) -> requests.R if print_error: _print_error(response) return response - + @rlim.placeholder def patch_group(self, group_id: str, *, player_identification: Union[str, enums.PlayerIdentification] = ..., @@ -726,12 +722,12 @@ def patch_group(self, group_id: str, *, print_error : bool, optional, default=True Prints an error message (that contains information about the error) if the request resulted in an HTTP error (i.e. status codes 400 through 599). - + Returns ------- requests.Response The `requests.Response` object returned from the HTTP request. - + """ # prepare url prepped_url = httpprep.URL(protocol="https", domain="ballchasing", top_level_domain="com", @@ -754,16 +750,16 @@ def patch_group(self, group_id: str, *, if print_error: _print_error(response) return response - + def maps(self, *, print_error: bool = True) -> requests.Response: """Get a list of current maps. - + Parameters ---------- print_error : bool, optional, default=True Prints an error message (that contains information about the error) if the request resulted in an HTTP error (i.e. status codes 400 through 599). - + Returns ------- requests.Response diff --git a/src/pychasing/enums.py b/src/pychasing/enums.py index 7870e92..c090ee6 100644 --- a/src/pychasing/enums.py +++ b/src/pychasing/enums.py @@ -154,17 +154,17 @@ class Map(enum.Enum): arc_p="arc_p" Starbase_ARC="arc_p" arc_standard_p="arc_standard_p" - Starbase_ARC_Standard="arc_standard_p" + Starbase_ARC_Standard="arc_standard_p" bb_p="bb_p" Champions_Field_NFL="bb_p" beach_night_p="beach_night_p" - Salty_Shores_Night="beach_night_p" + Salty_Shores_Night="beach_night_p" beach_p="beach_p" Salty_Shores="beach_p" beachvolley="beachvolley" - Salty_Shores_Volley="beachvolley" - chn_stadium_day_p="chn_stadium_day_p" - Forbidden_Temple_Day="chn_stadium_day_p" + Salty_Shores_Volley="beachvolley" + chn_stadium_day_p="chn_stadium_day_p" + Forbidden_Temple_Day="chn_stadium_day_p" chn_stadium_p="chn_stadium_p" Forbidden_Temple="chn_stadium_p" cs_day_p="cs_day_p" @@ -174,7 +174,7 @@ class Map(enum.Enum): cs_p="cs_p" Champions_Field="cs_p" eurostadium_night_p="eurostadium_night_p" - Mannfield_Night="eurostadium_night_p" + Mannfield_Night="eurostadium_night_p" eurostadium_p="eurostadium_p" Mannfield="eurostadium_p" eurostadium_rainy_p="eurostadium_rainy_p" @@ -310,7 +310,7 @@ class Season(enum.Enum): f2p_28="f28" f2p_29="f29" f2p_30="f30" - + class PlayerIdentification(enum.Enum): by_id="by-id" by_name="by-name" diff --git a/src/pychasing/models.py b/src/pychasing/models.py index 6bbb941..c55e928 100644 --- a/src/pychasing/models.py +++ b/src/pychasing/models.py @@ -27,7 +27,7 @@ def __init__(self, year: int, month: int, day: int, hour: int = ..., hour : int, optional, default=0 minute : int, optional, default=0 second : int, optional, default=0 - + """ def __new__(cls: "Date", year: int, month: int, day: int, hour: int = ..., minute: int = ..., second: int = ...) -> "Date": @@ -38,7 +38,7 @@ def __new__(cls: "Date", year: int, month: int, day: int, hour: int = ..., class ReplayBuffer(io.BufferedReader): """An object that can be used to store a replay file in-memory before uploading. - + """ def __init__(self, name: str, raw: bytes = ..., buffer_size: int = ...) -> None: self._name = name @@ -50,7 +50,7 @@ def __init__(self, name: str, raw: bytes = ..., buffer_size: int = ...) -> None: super().__init__(self._buffer) else: super().__init__(self._buffer, buffer_size) - + @property def name(self) -> str: return self._name