diff --git a/aiosu/helpers.py b/aiosu/helpers.py index 32c9919..e732a8d 100644 --- a/aiosu/helpers.py +++ b/aiosu/helpers.py @@ -10,6 +10,8 @@ from typing import Callable from typing import Optional from typing import TypeVar + from collections.abc import Mapping + from collections.abc import MutableMapping T = TypeVar("T") @@ -19,13 +21,13 @@ ) -def from_list(f: Callable[[Any], T], x: Any) -> list[T]: +def from_list(f: Callable[[Any], T], x: list[object]) -> list[T]: r"""Applies a function to all elements in a list. :param f: Function to apply on list elements :type f: Callable[[Any], T] :param x: List of objects - :type x: Any + :type x: list[object] :raises TypeError: If x is not a list :return: New list :rtype: list[T] @@ -36,8 +38,8 @@ def from_list(f: Callable[[Any], T], x: Any) -> list[T]: def add_param( - params: dict[str, Any], - kwargs: dict[str, Any], + params: MutableMapping[str, Any], + kwargs: Mapping[str, Any], key: str, param_name: Optional[str] = None, converter: Optional[Callable[[Any], T]] = None, @@ -45,9 +47,9 @@ def add_param( r"""Adds a parameter to a dictionary if it exists in kwargs. :param params: Dictionary to add parameter to - :type params: dict[str, Any] + :type params: Mapping[str, Any] :param kwargs: Dictionary to get parameter from - :type kwargs: dict[str, Any] + :type kwargs: Mapping[str, Any] :param key: Key to get parameter from :type key: str :param param_name: Name of parameter to add to dictionary, defaults to None @@ -57,10 +59,12 @@ def add_param( :return: True if parameter was added, False otherwise :rtype: bool """ - if key in kwargs: - value = kwargs[key] - if converter: - value = converter(value) - params[param_name or key] = value - return True - return False + if key not in kwargs: + return False + + value = kwargs[key] + if converter: + value = converter(value) + + params[param_name or key] = value + return True diff --git a/aiosu/models/base.py b/aiosu/models/base.py index 1e98bd1..d51db38 100644 --- a/aiosu/models/base.py +++ b/aiosu/models/base.py @@ -3,6 +3,9 @@ """ from __future__ import annotations +from typing import SupportsFloat +from typing import SupportsInt + import pydantic from pydantic import ConfigDict @@ -35,3 +38,21 @@ class FrozenModel(BaseModel): populate_by_name=True, frozen=True, ) + + +def cast_int(v: object) -> int: + if v is None: + return 0 + if isinstance(v, (SupportsInt, str)): + return int(v) + + raise ValueError(f"{v} is not a valid value.") + + +def cast_float(v: object) -> float: + if v is None: + return 0.0 + if isinstance(v, (SupportsFloat, str)): + return float(v) + + raise ValueError(f"{v} is not a valid value.") diff --git a/aiosu/models/beatmap.py b/aiosu/models/beatmap.py index 5610d59..9762bf3 100644 --- a/aiosu/models/beatmap.py +++ b/aiosu/models/beatmap.py @@ -3,11 +3,11 @@ """ from __future__ import annotations +from collections.abc import Mapping from datetime import datetime from enum import Enum from enum import unique from functools import cached_property -from typing import Any from typing import Literal from typing import Optional @@ -16,6 +16,7 @@ from pydantic import model_validator from .base import BaseModel +from .base import cast_int from .common import CurrentUserAttributes from .common import CursorModel from .gamemode import Gamemode @@ -133,7 +134,7 @@ def __str__(self) -> str: return self.name_api @classmethod - def _missing_(cls, query: object) -> Any: + def _missing_(cls, query: object) -> BeatmapRankStatus: if isinstance(query, int): for status in list(BeatmapRankStatus): if status.id == query: @@ -165,7 +166,7 @@ class BeatmapAvailability(BaseModel): download_disabled: Optional[bool] = None @classmethod - def _from_api_v1(cls, data: Any) -> BeatmapAvailability: + def _from_api_v1(cls, data: Mapping[str, object]) -> BeatmapAvailability: return cls.model_validate({"download_disabled": data["download_unavailable"]}) @@ -208,8 +209,8 @@ def from_beatmapset_id(cls, beatmapset_id: int) -> BeatmapCovers: ) @classmethod - def _from_api_v1(cls, data: Any) -> BeatmapCovers: - return cls.from_beatmapset_id(data["beatmapset_id"]) + def _from_api_v1(cls, data: Mapping[str, object]) -> BeatmapCovers: + return cls.from_beatmapset_id(cast_int(data["beatmapset_id"])) class BeatmapHype(BaseModel): @@ -294,7 +295,7 @@ def count_objects(self) -> Optional[int]: @model_validator(mode="before") @classmethod - def _set_url(cls, values: dict[str, Any]) -> dict[str, Any]: + def _set_url(cls, values: dict[str, object]) -> dict[str, object]: if values.get("url") is None: id = values["id"] beatmapset_id = values["beatmapset_id"] @@ -305,14 +306,14 @@ def _set_url(cls, values: dict[str, Any]) -> dict[str, Any]: return values @classmethod - def _from_api_v1(cls, data: Any) -> Beatmap: + def _from_api_v1(cls, data: Mapping[str, object]) -> Beatmap: return cls.model_validate( { "beatmapset_id": data["beatmapset_id"], "difficulty_rating": data["difficultyrating"], "id": data["beatmap_id"], - "mode": int(data["mode"]), - "status": int(data["approved"]), + "mode": cast_int(data["mode"]), + "status": cast_int(data["approved"]), "total_length": data["total_length"], "hit_length": data["total_length"], "user_id": data["creator_id"], @@ -388,7 +389,7 @@ def discussion_url(self) -> str: return f"https://osu.ppy.sh/beatmapsets/{self.id}/discussion" @classmethod - def _from_api_v1(cls, data: Any) -> Beatmapset: + def _from_api_v1(cls, data: Mapping[str, object]) -> Beatmapset: return cls.model_validate( { "id": data["beatmapset_id"], @@ -400,7 +401,7 @@ def _from_api_v1(cls, data: Any) -> Beatmapset: "play_count": data["playcount"], "preview_url": f"//b.ppy.sh/preview/{data['beatmapset_id']}.mp3", "source": data["source"], - "status": int(data["approved"]), + "status": cast_int(data["approved"]), "title": data["title"], "title_unicode": data["title"], "user_id": data["creator_id"], @@ -505,8 +506,10 @@ class BeatmapsetDiscussionResponse(CursorModel): @model_validator(mode="before") @classmethod - def _set_max_blocks(cls, values: dict[str, Any]) -> dict[str, Any]: - values["max_blocks"] = values["reviews_config"]["max_blocks"] + def _set_max_blocks(cls, values: dict[str, object]) -> dict[str, object]: + if isinstance(values["reviews_config"], Mapping): + values["max_blocks"] = values["reviews_config"]["max_blocks"] + return values diff --git a/aiosu/models/common.py b/aiosu/models/common.py index 8f083cd..a52a77e 100644 --- a/aiosu/models/common.py +++ b/aiosu/models/common.py @@ -7,7 +7,6 @@ from datetime import datetime from functools import cached_property from functools import partial -from typing import Any from typing import Literal from typing import Optional @@ -53,10 +52,13 @@ class TimestampedCount(BaseModel): @field_validator("start_date", mode="before") @classmethod - def _date_validate(cls, v: Any) -> Any: + def _date_validate(cls, v: object) -> datetime: if isinstance(v, str): return datetime.strptime(v, "%Y-%m-%d") - return v + if isinstance(v, datetime): + return v + + raise ValueError(f"{v} is not a valid value.") class Achievement(BaseModel): @@ -131,7 +133,7 @@ class CursorModel(BaseModel): """ cursor_string: Optional[str] = None - next: Optional[partial[Coroutine[Any, Any, CursorModel]]] = Field( + next: Optional[partial[Coroutine[object, object, CursorModel]]] = Field( default=None, exclude=True, ) diff --git a/aiosu/models/gamemode.py b/aiosu/models/gamemode.py index 9a98f83..bc3f2d6 100644 --- a/aiosu/models/gamemode.py +++ b/aiosu/models/gamemode.py @@ -5,10 +5,6 @@ from enum import Enum from enum import unique -from typing import TYPE_CHECKING - -if TYPE_CHECKING: - from typing import Any __all__ = ("Gamemode",) @@ -88,5 +84,5 @@ def from_type(cls, __o: object) -> Gamemode: raise ValueError(f"Gamemode {__o} does not exist.") @classmethod - def _missing_(cls, query: object) -> Any: + def _missing_(cls, query: object) -> Gamemode: return cls.from_type(query) diff --git a/aiosu/models/lazer.py b/aiosu/models/lazer.py index c9f9b70..5553991 100644 --- a/aiosu/models/lazer.py +++ b/aiosu/models/lazer.py @@ -5,7 +5,6 @@ from datetime import datetime from functools import cached_property -from typing import Any from typing import Optional from pydantic import computed_field @@ -63,7 +62,7 @@ class LazerMod(BaseModel): """Temporary model for lazer mods.""" acronym: str - settings: dict[str, Any] = Field(default_factory=dict) + settings: dict[str, object] = Field(default_factory=dict) def __str__(self) -> str: return self.acronym @@ -216,7 +215,7 @@ def mods_str(self) -> str: @model_validator(mode="before") @classmethod - def _fail_rank(cls, values: dict[str, Any]) -> dict[str, Any]: + def _fail_rank(cls, values: dict[str, object]) -> dict[str, object]: if not values["passed"]: values["rank"] = "F" return values diff --git a/aiosu/models/legacy/match.py b/aiosu/models/legacy/match.py index c3317d9..408b40f 100644 --- a/aiosu/models/legacy/match.py +++ b/aiosu/models/legacy/match.py @@ -1,22 +1,21 @@ from __future__ import annotations +from collections.abc import Mapping from datetime import datetime from enum import IntEnum from enum import unique from typing import Optional -from typing import TYPE_CHECKING from pydantic import Field from pydantic import field_validator from pydantic import model_validator from ..base import BaseModel +from ..base import cast_int from ..gamemode import Gamemode from ..mods import Mods from ..score import ScoreStatistics -if TYPE_CHECKING: - from typing import Any __all__ = ( "MatchTeam", @@ -74,21 +73,19 @@ def get_full_mods(self, game: MatchGame) -> Mods: @model_validator(mode="before") @classmethod - def _set_statistics(cls, values: dict[str, Any]) -> dict[str, Any]: + def _set_statistics(cls, values: dict[str, object]) -> dict[str, object]: values["statistics"] = ScoreStatistics._from_api_v1(values) return values @field_validator("enabled_mods", mode="before") @classmethod - def _set_enabled_mods(cls, v: Any) -> int: - if v is not None: - return int(v) - return 0 + def _set_enabled_mods(cls, v: object) -> int: + return cast_int(v) @field_validator("team", mode="before") @classmethod - def _set_team(cls, v: Any) -> int: - return int(v) + def _set_team(cls, v: object) -> int: + return cast_int(v) class MatchGame(BaseModel): @@ -108,25 +105,23 @@ class MatchGame(BaseModel): @field_validator("mode", mode="before") @classmethod - def _set_mode(cls, v: Any) -> int: - return int(v) + def _set_mode(cls, v: object) -> int: + return cast_int(v) @field_validator("mods", mode="before") @classmethod - def _set_mods(cls, v: Any) -> int: - if v is not None: - return int(v) - return 0 + def _set_mods(cls, v: object) -> int: + return cast_int(v) @field_validator("scoring_type", mode="before") @classmethod - def _set_scoring_type(cls, v: Any) -> int: - return int(v) + def _set_scoring_type(cls, v: object) -> int: + return cast_int(v) @field_validator("team_type", mode="before") @classmethod - def _set_team_type(cls, v: Any) -> int: - return int(v) + def _set_team_type(cls, v: object) -> int: + return cast_int(v) class Match(BaseModel): @@ -141,5 +136,9 @@ class Match(BaseModel): @model_validator(mode="before") @classmethod - def _format_values(cls, values: dict[str, Any]) -> dict[str, Any]: - return {**values["match"], "games": values["games"]} + def _format_values(cls, values: dict[str, object]) -> dict[str, object]: + match = values["match"] + if not isinstance(match, Mapping): + raise ValueError(f"Invalid match type: {type(match)}") + + return {**match, "games": values["games"]} diff --git a/aiosu/models/mods.py b/aiosu/models/mods.py index 4498372..a46a594 100644 --- a/aiosu/models/mods.py +++ b/aiosu/models/mods.py @@ -15,7 +15,6 @@ if TYPE_CHECKING: - from typing import Any from typing import Union __all__ = ( @@ -139,7 +138,7 @@ def from_type(cls, __o: object) -> Mod: raise ValueError(f"Mod {__o!r} does not exist.") @classmethod - def _missing_(cls, query: object) -> Any: + def _missing_(cls, query: object) -> Mod: return cls.from_type(query) @@ -207,7 +206,7 @@ def __or__(self, __o: object) -> int: @classmethod def __get_pydantic_core_schema__( cls, - source_type: type[Any], + source_type: type[object], handler: GetCoreSchemaHandler, ) -> CoreSchema: return core_schema.no_info_before_validator_function( diff --git a/aiosu/models/multiplayer.py b/aiosu/models/multiplayer.py index 374555f..c61869f 100644 --- a/aiosu/models/multiplayer.py +++ b/aiosu/models/multiplayer.py @@ -3,8 +3,8 @@ """ from __future__ import annotations +from collections.abc import Mapping from datetime import datetime -from typing import Any from typing import Literal from typing import Optional @@ -105,8 +105,8 @@ class MultiplayerEvent(BaseModel): @model_validator(mode="before") @classmethod - def _set_type(cls, values: dict[str, Any]) -> dict[str, Any]: - if "detail" in values: + def _set_type(cls, values: dict[str, object]) -> dict[str, object]: + if "detail" in values and isinstance(values["detail"], Mapping): values["type"] = values["detail"]["type"] return values diff --git a/aiosu/models/oauthtoken.py b/aiosu/models/oauthtoken.py index 154440c..47c2242 100644 --- a/aiosu/models/oauthtoken.py +++ b/aiosu/models/oauthtoken.py @@ -6,7 +6,6 @@ from datetime import datetime from datetime import timedelta from functools import cached_property -from typing import TYPE_CHECKING import jwt from pydantic import computed_field @@ -15,8 +14,6 @@ from .base import FrozenModel from .scopes import Scopes -if TYPE_CHECKING: - from typing import Any __all__ = ("OAuthToken",) @@ -55,9 +52,10 @@ def can_refresh(self) -> bool: @model_validator(mode="before") @classmethod - def _set_expires_on(cls, values: dict[str, Any]) -> dict[str, Any]: - if isinstance(values.get("expires_in"), int): + def _set_expires_on(cls, values: dict[str, object]) -> dict[str, object]: + expires_in = values.get("expires_in") + if isinstance(expires_in, int): values["expires_on"] = datetime.utcnow() + timedelta( - seconds=values["expires_in"], + seconds=expires_in, ) return values diff --git a/aiosu/models/score.py b/aiosu/models/score.py index 0412524..f01e6ab 100644 --- a/aiosu/models/score.py +++ b/aiosu/models/score.py @@ -16,6 +16,7 @@ from ..utils.accuracy import OsuAccuracyCalculator from ..utils.accuracy import TaikoAccuracyCalculator from .base import BaseModel +from .base import cast_int from .beatmap import Beatmap from .beatmap import Beatmapset from .common import CurrentUserAttributes @@ -24,7 +25,7 @@ from .user import User if TYPE_CHECKING: - from typing import Any + from collections.abc import Mapping from .. import v1 __all__ = ( @@ -113,7 +114,7 @@ class ScoreStatistics(BaseModel): @model_validator(mode="before") @classmethod - def _convert_none_to_zero(cls, values: dict[str, Any]) -> dict[str, Any]: + def _convert_none_to_zero(cls, values: dict[str, object]) -> dict[str, object]: # Lazer API returns null for some statistics for key in values: if values[key] is None: @@ -121,7 +122,7 @@ def _convert_none_to_zero(cls, values: dict[str, Any]) -> dict[str, Any]: return values @classmethod - def _from_api_v1(cls, data: Any) -> ScoreStatistics: + def _from_api_v1(cls, data: Mapping[str, object]) -> ScoreStatistics: return cls.model_validate( { "count_50": data["count50"], @@ -211,7 +212,7 @@ def completion(self) -> Optional[float]: @model_validator(mode="before") @classmethod - def _fail_rank(cls, values: dict[str, Any]) -> dict[str, Any]: + def _fail_rank(cls, values: dict[str, object]) -> dict[str, object]: if not values["passed"]: values["rank"] = "F" return values @@ -235,7 +236,7 @@ async def request_beatmap(self, client: v1.Client) -> None: @classmethod def _from_api_v1( cls, - data: Any, + data: Mapping[str, object], mode: Gamemode, ) -> Score: statistics = ScoreStatistics._from_api_v1(data) @@ -244,7 +245,7 @@ def _from_api_v1( "id": data["score_id"], "user_id": data["user_id"], "accuracy": 0.0, - "mods": int(data["enabled_mods"]), + "mods": cast_int(data["enabled_mods"]), "score": data["score"], "pp": data.get("pp", 0.0), "max_combo": data["maxcombo"], diff --git a/aiosu/models/user.py b/aiosu/models/user.py index 2744a9f..8686b87 100644 --- a/aiosu/models/user.py +++ b/aiosu/models/user.py @@ -3,26 +3,25 @@ """ from __future__ import annotations +from collections.abc import Mapping from datetime import datetime from enum import Enum from enum import unique from functools import cached_property -from typing import Any from typing import Literal from typing import Optional -from typing import TYPE_CHECKING from pydantic import computed_field from pydantic import Field from .base import BaseModel +from .base import cast_float +from .base import cast_int from .common import Country from .common import HTMLBody from .common import TimestampedCount from .gamemode import Gamemode -if TYPE_CHECKING: - from typing import Callable __all__ = ( "User", @@ -44,9 +43,6 @@ ) -cast_int: Callable[..., int] = lambda x: int(x or 0) -cast_float: Callable[..., float] = lambda x: float(x or 0) - UserAccountHistoryType = Literal[ "note", "restriction", @@ -81,7 +77,7 @@ def new_api_name(self) -> str: return self.value @classmethod - def _missing_(cls, query: object) -> Any: + def _missing_(cls, query: object) -> UserQueryType: if isinstance(query, str): query = query.lower() for q in list(UserQueryType): @@ -95,7 +91,7 @@ class UserLevel(BaseModel): progress: int @classmethod - def _from_api_v1(cls, data: Any) -> UserLevel: + def _from_api_v1(cls, data: Mapping[str, object]) -> UserLevel: level = cast_float(data["level"]) current = int(level) progress = (level - current) * 100 @@ -172,7 +168,7 @@ class UserGradeCounts(BaseModel): """Number of A ranks achieved.""" @classmethod - def _from_api_v1(cls, data: Any) -> UserGradeCounts: + def _from_api_v1(cls, data: Mapping[str, object]) -> UserGradeCounts: return cls.model_validate( { "ss": cast_int(data["count_rank_ss"]), @@ -244,7 +240,7 @@ def pp_per_playtime(self) -> float: return self.pp / self.play_time * 3600 @classmethod - def _from_api_v1(cls, data: Any) -> UserStats: + def _from_api_v1(cls, data: Mapping[str, object]) -> UserStats: """Some fields can be None, we want to force them to cast to a value.""" return cls.model_validate( { @@ -337,7 +333,7 @@ def url(self) -> str: return f"https://osu.ppy.sh/users/{self.id}" @classmethod - def _from_api_v1(cls, data: Any) -> User: + def _from_api_v1(cls, data: Mapping[str, object]) -> User: return cls.model_validate( { "avatar_url": f"https://s.ppy.sh/a/{data['user_id']}", diff --git a/aiosu/utils/accuracy.py b/aiosu/utils/accuracy.py index ef04380..b1b0734 100644 --- a/aiosu/utils/accuracy.py +++ b/aiosu/utils/accuracy.py @@ -6,11 +6,11 @@ import abc from typing import TYPE_CHECKING +from ..models.base import cast_int from ..models.gamemode import Gamemode from ..models.mods import Mod if TYPE_CHECKING: - from typing import Callable from ..models.score import Score __all__ = [ @@ -20,8 +20,6 @@ "CatchAccuracyCalculator", ] -cast_int: Callable[..., int] = lambda x: int(x or 0) - def get_calculator(mode: Gamemode) -> type[AbstractAccuracyCalculator]: r"""Returns the accuracy calculator for the given gamemode. diff --git a/aiosu/utils/binary.py b/aiosu/utils/binary.py index e33560d..fd3c513 100644 --- a/aiosu/utils/binary.py +++ b/aiosu/utils/binary.py @@ -10,7 +10,6 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: - from typing import Any from typing import BinaryIO _lzma_format = lzma.FORMAT_ALONE @@ -195,7 +194,7 @@ def unpack_replay_data(file: BinaryIO) -> str: return data.decode("ascii") -def pack(file: BinaryIO, fmt: str, value: Any) -> None: +def pack(file: BinaryIO, fmt: str, value: object) -> None: r"""Pack a value into a file. :param file: The file to pack into. @@ -203,7 +202,7 @@ def pack(file: BinaryIO, fmt: str, value: Any) -> None: :param fmt: The format to pack with. :type fmt: str :param value: The value to pack. - :type value: Any + :type value: object """ file.write(struct.pack(fmt, value)) diff --git a/aiosu/v1/client.py b/aiosu/v1/client.py index 7b54a8c..61565cc 100644 --- a/aiosu/v1/client.py +++ b/aiosu/v1/client.py @@ -30,6 +30,7 @@ if TYPE_CHECKING: from types import TracebackType from typing import Any + from collections.abc import MutableMapping from typing import Optional from typing import Union @@ -44,7 +45,11 @@ def get_content_type(content_type: str) -> str: return content_type.split(";")[0] -def _beatmap_score_conv(data: Any, mode: Gamemode, beatmap_id: int) -> Score: +def _beatmap_score_conv( + data: MutableMapping[str, object], + mode: Gamemode, + beatmap_id: int, +) -> Score: data["beatmap_id"] = beatmap_id return Score._from_api_v1(data, mode) diff --git a/aiosu/v2/client.py b/aiosu/v2/client.py index c113094..f6c901f 100644 --- a/aiosu/v2/client.py +++ b/aiosu/v2/client.py @@ -6,10 +6,10 @@ from __future__ import annotations import functools +from collections.abc import Awaitable from datetime import datetime from functools import partial from io import BytesIO -from typing import Any from typing import Callable from typing import cast from typing import Literal @@ -81,16 +81,17 @@ if TYPE_CHECKING: from types import TracebackType + from typing import Any from typing import Optional from typing import Union __all__ = ("Client",) -F = TypeVar("F", bound=Callable[..., Any]) +F = TypeVar("F", bound=Callable[..., Awaitable[object]]) ClientRequestType = Literal["GET", "POST", "DELETE", "PUT", "PATCH"] -def to_lower_str(value: Any) -> str: +def to_lower_str(value: object) -> str: """Converts a value to a lowercase string.""" return str(value).lower() @@ -106,7 +107,7 @@ def prepare_token(func: F) -> F: """ @functools.wraps(func) - async def _prepare_token(self: Client, *args: Any, **kwargs: Any) -> Any: + async def _prepare_token(self: Client, *args: Any, **kwargs: Any) -> object: await self._prepare_token() return await func(self, *args, **kwargs) @@ -121,7 +122,7 @@ def check_token(func: F) -> F: """ @functools.wraps(func) - async def _check_token(self: Client, *args: Any, **kwargs: Any) -> Any: + async def _check_token(self: Client, *args: Any, **kwargs: Any) -> object: token = await self.get_current_token() if datetime.utcnow().timestamp() > token.expires_on.timestamp(): await self._refresh() @@ -143,7 +144,7 @@ def _requires_scope( func: F, ) -> F: @functools.wraps(func) - async def _wrap(self: Client, *args: Any, **kwargs: Any) -> Any: + async def _wrap(self: Client, *args: Any, **kwargs: Any) -> object: token = await self.get_current_token() if any_scope: if not (required_scopes & token.scopes): @@ -250,7 +251,7 @@ async def func(event: ClientUpdateEvent) self._register_listener(func, ClientUpdateEvent) @functools.wraps(func) - async def _on_client_update(*args: Any, **kwargs: Any) -> Any: + async def _on_client_update(*args: Any, **kwargs: Any) -> object: return await func(*args, **kwargs) return cast(F, _on_client_update) @@ -420,7 +421,7 @@ async def get_featured_artists(self, **kwargs: Any) -> ArtistResponse: :rtype: aiosu.models.artist.ArtistResponse """ url = f"{self.base_url}/beatmaps/artists/tracks" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="limit") add_param(params, kwargs, key="album") add_param(params, kwargs, key="artist") @@ -476,7 +477,7 @@ async def get_changelog_listing(self, **kwargs: Any) -> ChangelogListing: :rtype: aiosu.models.changelog.ChangelogListing """ url = f"{self.base_url}/api/v2/changelog" - params: dict[str, Any] = { + params: dict[str, object] = { "message_formats": kwargs.pop("message_formats", ["html", "markdown"]), } add_param(params, kwargs, key="from") @@ -529,7 +530,7 @@ async def lookup_changelog_build( :rtype: aiosu.models.changelog.Build """ url = f"{self.base_url}/api/v2/changelog/{changelog_query}" - params: dict[str, Any] = { + params: dict[str, object] = { "message_formats": kwargs.pop("message_formats", ["html", "markdown"]), } if "is_id" in kwargs or isinstance(changelog_query, int): @@ -559,7 +560,7 @@ async def get_news_listing(self, **kwargs: Any) -> NewsListing: url = f"{self.base_url}/api/v2/news" if not 1 <= (limit := kwargs.pop("limit", 12)) <= 21: raise ValueError("Invalid limit specified. Limit must be between 1 and 21") - params: dict[str, Any] = { + params: dict[str, object] = { "limit": limit, } add_param(params, kwargs, key="year") @@ -593,7 +594,7 @@ async def get_news_post( :rtype: aiosu.models.news.NewsPost """ url = f"{self.base_url}/api/v2/news/{news_query}" - params: dict[str, Any] = { + params: dict[str, object] = { "message_formats": kwargs.pop("message_formats", ["html", "markdown"]), } if "is_id" in kwargs or isinstance(news_query, int): @@ -635,7 +636,7 @@ async def get_comment(self, comment_id: int, **kwargs: Any) -> CommentBundle: :rtype: aiosu.models.comment.CommentBundle """ url = f"{self.base_url}/api/v2/comments/{comment_id}" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="cursor_string") json = await self._request("GET", url, params=params) resp = CommentBundle.model_validate(json) @@ -668,7 +669,7 @@ async def get_comments(self, **kwargs: Any) -> CommentBundle: :rtype: aiosu.models.comment.CommentBundle """ url = f"{self.base_url}/api/v2/comments" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="commentable_type") add_param(params, kwargs, key="commentable_id") add_param(params, kwargs, key="parent_id") @@ -703,7 +704,7 @@ async def search(self, query: str, **kwargs: Any) -> SearchResponse: :rtype: aiosu.models.search.SearchResponse """ url = f"{self.base_url}/api/v2/search" - params: dict[str, Any] = { + params: dict[str, object] = { "query": query, "mode": kwargs.pop("mode", "all"), } @@ -772,7 +773,7 @@ async def get_user(self, user_query: Union[str, int], **kwargs: Any) -> User: :rtype: aiosu.models.user.User """ url = f"{self.base_url}/api/v2/users/{user_query}" - params: dict[str, Any] = {} + params: dict[str, object] = {} if "mode" in kwargs: mode = Gamemode(kwargs.pop("mode")) url += f"/{mode}" @@ -799,7 +800,7 @@ async def get_users(self, user_ids: list[int]) -> list[User]: :rtype: list[aiosu.models.user.User] """ url = f"{self.base_url}/api/v2/users" - params: dict[str, Any] = { + params: dict[str, object] = { "ids": user_ids, } json = await self._request("GET", url, params=params) @@ -827,7 +828,7 @@ async def get_user_kudosu(self, user_id: int, **kwargs: Any) -> list[KudosuHisto :rtype: list[aiosu.models.kudosu.KudosuHistory] """ url = f"{self.base_url}/api/v2/users/{user_id}/kudosu" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="limit") add_param(params, kwargs, key="offset") json = await self._request("GET", url, params=params) @@ -876,7 +877,7 @@ async def __get_type_scores( f'"{request_type}" is not a valid request_type. Valid options are: "recent", "best", "firsts"', ) url = f"{self.base_url}/api/v2/users/{user_id}/scores/{request_type}" - params: dict[str, Any] = { + params: dict[str, object] = { "include_fails": int(kwargs.pop("include_fails", False)), "limit": limit, "offset": kwargs.pop("offset", 0), @@ -1032,7 +1033,7 @@ async def get_user_beatmap_scores( :rtype: list[aiosu.models.score.Score] """ url = f"{self.base_url}/api/v2/beatmaps/{beatmap_id}/scores/users/{user_id}/all" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="mode", converter=lambda x: str(Gamemode(x))) json = await self._request("GET", url, params=params) return from_list(Score.model_validate, json.get("scores", [])) @@ -1066,7 +1067,7 @@ async def get_user_beatmaps( :rtype: list[aiosu.models.beatmap.Beatmap] """ url = f"{self.base_url}/api/v2/users/{user_id}/beatmapsets/{type}" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="limit") add_param(params, kwargs, key="offset") json = await self._request("GET", url, params=params) @@ -1098,7 +1099,7 @@ async def get_user_most_played( :rtype: list[aiosu.models.beatmap.BeatmapUserPlaycount] """ url = f"{self.base_url}/api/v2/users/{user_id}/beatmapsets/most_played" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="limit") add_param(params, kwargs, key="offset") json = await self._request("GET", url, params=params) @@ -1130,7 +1131,7 @@ async def get_user_recent_activity( :rtype: list[aiosu.models.event.Event] """ url = f"{self.base_url}/api/v2/users/{user_id}/recent_activity" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="limit") add_param(params, kwargs, key="offset") json = await self._request("GET", url, params=params) @@ -1174,7 +1175,7 @@ async def get_beatmap_scores(self, beatmap_id: int, **kwargs: Any) -> list[Score :rtype: list[aiosu.models.score.Score] """ url = f"{self.base_url}/api/v2/beatmaps/{beatmap_id}/scores" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="mode", converter=lambda x: str(Gamemode(x))) add_param( params, @@ -1215,7 +1216,7 @@ async def get_beatmaps(self, beatmap_ids: list[int]) -> list[Beatmap]: :rtype: list[aiosu.models.beatmap.Beatmap] """ url = f"{self.base_url}/api/v2/beatmaps" - params: dict[str, Any] = { + params: dict[str, object] = { "ids": beatmap_ids, } json = await self._request("GET", url, params=params) @@ -1244,7 +1245,7 @@ async def lookup_beatmap(self, **kwargs: Any) -> Beatmap: :rtype: aiosu.models.beatmap.Beatmap """ url = f"{self.base_url}/api/v2/beatmaps/lookup" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="checksum") add_param(params, kwargs, key="filename") add_param(params, kwargs, key="id") @@ -1279,7 +1280,7 @@ async def get_beatmap_attributes( :rtype: aiosu.models.beatmap.BeatmapDifficultyAttributes """ url = f"{self.base_url}/api/v2/beatmaps/{beatmap_id}/attributes" - data: dict[str, Any] = {} + data: dict[str, object] = {} add_param( data, kwargs, @@ -1321,7 +1322,7 @@ async def lookup_beatmapset(self, beatmap_id: int) -> Beatmapset: :rtype: aiosu.models.beatmap.Beatmapset """ url = f"{self.base_url}/api/v2/beatmapsets/lookup" - params: dict[str, Any] = { + params: dict[str, object] = { "beatmap_id": beatmap_id, } json = await self._request("GET", url, params=params) @@ -1351,7 +1352,7 @@ async def search_beatmapsets( :rtype: list[aiosu.models.beatmap.BeatmapsetSearchResponse] """ url = f"{self.base_url}/api/v2/beatmapsets/search/{search_filter}" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="cursor_string") json = await self._request("GET", url) resp = BeatmapsetSearchResponse.model_validate(json) @@ -1390,7 +1391,7 @@ async def get_beatmapset_events(self, **kwargs: Any) -> list[BeatmapsetEvent]: :rtype: list[aiosu.models.event.Event] """ url = f"{self.base_url}/api/v2/beatmapsets/events" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="limit") add_param(params, kwargs, key="page") add_param(params, kwargs, key="beatmapset_id") @@ -1442,7 +1443,7 @@ async def get_beatmapset_discussions( :rtype: aiosu.models.beatmap.BeatmapsetDiscussionResponse """ url = f"{self.base_url}/api/v2/beatmapsets/discussions" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="beatmap_id") add_param(params, kwargs, key="beatmapset_id") add_param(params, kwargs, key="beatmapset_status") @@ -1496,7 +1497,7 @@ async def get_beatmapset_discussion_posts( :rtype: aiosu.models.beatmap.BeatmapsetDiscussionPostResponse """ url = f"{self.base_url}/api/v2/beatmapsets/discussions/posts" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="beatmapset_discussion_id") add_param(params, kwargs, key="limit") add_param(params, kwargs, key="page") @@ -1549,7 +1550,7 @@ async def get_beatmapset_discussion_votes( :rtype: aiosu.models.beatmap.BeatmapsetDiscussionVoteResponse """ url = f"{self.base_url}/api/v2/beatmapsets/discussions/votes" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="beatmapset_discussion_id") add_param(params, kwargs, key="limit") add_param(params, kwargs, key="page") @@ -1647,7 +1648,7 @@ async def get_rankings( :rtype: aiosu.models.rankings.Rankings """ url = f"{self.base_url}/api/v2/rankings/{mode}/{type}" - params: dict[str, Any] = {} + params: dict[str, object] = {} add_param(params, kwargs, key="country") add_param(params, kwargs, key="filter") add_param(params, kwargs, key="spotlight") @@ -1678,7 +1679,7 @@ async def get_rankings_kudosu(self, **kwargs: Any) -> Rankings: :rtype: aiosu.models.rankings.Rankings """ url = f"{self.base_url}/api/v2/rankings/kudosu" - params: dict[str, Any] = {} + params: dict[str, int] = {} add_param(params, kwargs, key="page_id", param_name="page") json = await self._request("GET", url, params=params) resp = Rankings.model_validate(json) @@ -1730,7 +1731,7 @@ async def get_forum_topic(self, topic_id: int, **kwargs: Any) -> ForumTopicRespo if not 1 <= (limit := kwargs.pop("limit", 20)) <= 50: raise ValueError("Invalid limit specified. Limit must be between 1 and 50") url = f"{self.base_url}/api/v2/forums/topics/{topic_id}" - params: dict[str, Any] = { + params: dict[str, object] = { "limit": limit, } add_param(params, kwargs, key="sort") @@ -1787,14 +1788,14 @@ async def create_forum_topic( :rtype: aiosu.models.forum.ForumCreateTopicResponse """ url = f"{self.base_url}/api/v2/forums/topics" - data: dict[str, Any] = { + data: dict[str, object] = { "forum_id": forum_id, "title": title, "body": content, } add_param(data, kwargs, key="with_poll") if data.get("with_poll"): - forum_topic_poll: dict[str, Any] = { + forum_topic_poll: dict[str, object] = { "title": kwargs["poll_title"], "length_days": kwargs.pop("poll_length_days", 0), "vote_change": kwargs.pop("poll_vote_change", False), @@ -1898,7 +1899,7 @@ async def get_chat_ack(self, **kwargs: Any) -> list[ChatUserSilence]: :rtype: list[aiosu.models.chat.ChatUserSilence] """ url = f"{self.base_url}/api/v2/chat/ack" - data: dict[str, Any] = {} + data: dict[str, object] = {} add_param(data, kwargs, key="since") add_param(data, kwargs, key="silence_id_since", param_name="history_since") json = await self._request("POST", url, json=data) @@ -1934,7 +1935,7 @@ async def get_chat_updates(self, since: int, **kwargs: Any) -> ChatUpdateRespons if not 1 <= (limit := kwargs.get("limit", 50)) <= 50: raise ValueError("limit must be between 1 and 50") url = f"{self.base_url}/api/v2/chat/updates" - params: dict[str, Any] = { + params: dict[str, object] = { "since": since, "limit:": limit, } @@ -2008,7 +2009,7 @@ async def get_channel_messages( if not 1 <= (limit := kwargs.get("limit", 50)) <= 50: raise ValueError("limit must be between 1 and 50") url = f"{self.base_url}/api/v2/chat/channels/{channel_id}/messages" - params: dict[str, Any] = { + params: dict[str, object] = { "limit:": limit, } add_param(params, kwargs, key="since") @@ -2050,7 +2051,7 @@ async def create_chat_channel( :rtype: aiosu.models.chat.ChatChannel """ url = f"{self.base_url}/api/v2/chat/channels" - data: dict[str, Any] = { + data: dict[str, object] = { "type": type, } add_param(data, kwargs, key="message") @@ -2144,7 +2145,7 @@ async def send_message( :rtype: aiosu.models.chat.ChatMessage """ url = f"{self.base_url}/api/v2/chat/channels/{channel_id}/messages" - data: dict[str, Any] = { + data: dict[str, object] = { "message": message, "is_action": is_action, } @@ -2182,7 +2183,7 @@ async def send_private_message( :rtype: aiosu.models.chat.ChatMessageCreateResponse """ url = f"{self.base_url}/api/v2/chat/new" - data: dict[str, Any] = { + data: dict[str, object] = { "target_id": user_id, "message": message, "is_action": is_action, @@ -2219,7 +2220,7 @@ async def get_multiplayer_matches( if not 1 <= (limit := kwargs.pop("limit", 1)) <= 50: raise ValueError("Limit must be between 1 and 50") url = f"{self.base_url}/api/v2/matches" - params: dict[str, Any] = { + params: dict[str, object] = { "limit": limit, } add_param(params, kwargs, key="sort") @@ -2261,7 +2262,7 @@ async def get_multiplayer_match( if not 1 <= (limit := kwargs.pop("limit", 1)) <= 100: raise ValueError("Limit must be between 1 and 100") url = f"{self.base_url}/api/v2/matches/{match_id}" - params: dict[str, Any] = { + params: dict[str, object] = { "limit": limit, } add_param(params, kwargs, key="before") @@ -2302,7 +2303,7 @@ async def get_multiplayer_rooms(self, **kwargs: Any) -> list[MultiplayerRoom]: if "mode" in kwargs: mode: MultiplayerRoomMode = kwargs.pop("mode") url += f"/{mode}" - params: dict[str, Any] = { + params: dict[str, object] = { "limit": limit, } add_param(params, kwargs, key="sort") @@ -2356,7 +2357,7 @@ async def get_multiplayer_leaderboard( if not 1 <= (limit := kwargs.pop("limit", 50)) <= 50: raise ValueError("Limit must be between 1 and 50") url = f"{self.base_url}/api/v2/rooms/{room_id}/leaderboard" - params: dict[str, Any] = { + params: dict[str, object] = { "limit": limit, } json = await self._request("GET", url, params=params) @@ -2396,7 +2397,7 @@ async def get_multiplayer_scores( if not 1 <= (limit := kwargs.pop("limit", 1)) <= 100: raise ValueError("Limit must be between 1 and 100") url = f"{self.base_url}/api/v2/rooms/{room_id}/playlist/{playlist_id}/scores" - params: dict[str, Any] = { + params: dict[str, object] = { "limit": limit, } add_param(params, kwargs, key="sort") diff --git a/aiosu/v2/clientstorage.py b/aiosu/v2/clientstorage.py index f0fdd49..03af029 100644 --- a/aiosu/v2/clientstorage.py +++ b/aiosu/v2/clientstorage.py @@ -4,7 +4,7 @@ from __future__ import annotations import functools -from typing import Any +from collections.abc import Awaitable from typing import Callable from typing import cast from typing import TYPE_CHECKING @@ -21,12 +21,13 @@ if TYPE_CHECKING: from types import TracebackType + from typing import Any from typing import Optional from typing import Union __all__ = ("ClientStorage",) -F = TypeVar("F", bound=Callable[..., Any]) +F = TypeVar("F", bound=Callable[..., Awaitable[object]]) class ClientStorage(Eventable): @@ -91,7 +92,7 @@ async def func(event: ClientAddEvent) self._register_listener(func, ClientAddEvent) @functools.wraps(func) - async def _on_client_add(*args: Any, **kwargs: Any) -> Any: + async def _on_client_add(*args: Any, **kwargs: Any) -> object: return await func(*args, **kwargs) return cast(F, _on_client_add) @@ -107,7 +108,7 @@ async def func(event: ClientUpdateEvent) self._register_listener(func, ClientUpdateEvent) @functools.wraps(func) - async def _on_client_update(*args: Any, **kwargs: Any) -> Any: + async def _on_client_update(*args: Any, **kwargs: Any) -> object: return await func(*args, **kwargs) return cast(F, _on_client_update) diff --git a/tests/test_v1/test_client.py b/tests/test_v1/test_client.py index d393f0d..6dc5128 100644 --- a/tests/test_v1/test_client.py +++ b/tests/test_v1/test_client.py @@ -1,10 +1,15 @@ from __future__ import annotations +from typing import TYPE_CHECKING + import pytest import aiosu from ..classes import mock_request +if TYPE_CHECKING: + from collections.abc import Mapping + STATUS_CAN_200 = { 200: "application/json", } @@ -25,8 +30,8 @@ def get_data(func_name: str, status_code: int, extension: str = "json") -> bytes def generate_test( func, - status_codes: dict[int, str], - func_kwargs: dict = {}, + status_codes: Mapping[int, str], + func_kwargs: Mapping[str, object] = {}, ): @pytest.mark.asyncio @pytest.mark.parametrize("status_code, content_type", status_codes.items()) diff --git a/tests/test_v2/test_client.py b/tests/test_v2/test_client.py index d5bbf53..97ec1ec 100644 --- a/tests/test_v2/test_client.py +++ b/tests/test_v2/test_client.py @@ -1,10 +1,15 @@ from __future__ import annotations +from typing import TYPE_CHECKING + import pytest import aiosu from ..classes import mock_request +if TYPE_CHECKING: + from collections.abc import Mapping + STATUS_CAN_200 = { 200: "application/json", } @@ -40,8 +45,8 @@ def get_data(func_name: str, status_code: int, extension: str = "json") -> bytes def generate_test( func, - status_codes: dict[int, str], - func_kwargs: dict = {}, + status_codes: Mapping[int, str], + func_kwargs: Mapping = {}, ): @pytest.mark.asyncio @pytest.mark.parametrize("status_code, content_type", status_codes.items())