減少代碼重複，利用裝飾器來處理網路請求與回傳

Readme.md

使用方法

/api/get_user

>>> api = Client(token=...)
>>> user = await api.get_user('840') # List[User]
>>> user[0]
User(user_id=6008293, username='840', join_date=datetime.datetime(2015, 2, 16, 14, 19, 4), count300=18896911, count100=1456061, count50=130196, playcount=70506, ranked_score=44995986434, total_score=169312471943, pp_rank=15969, pp_raw=6928.38, level=101.424, accuracy=99.12299346923828, count_rank_ss=146, count_rank_ssh=17, count_rank_s=1196, count_rank_sh=198, count_rank_a=2715, country='TW', total_seconds_played=5284370, pp_country_rank=282, events=[])

/api/get_beatmaps

>>> api = Client(token=...)
>>> beatmaps = await api.get_beatmaps(s=1283387) # List[Beatmap]
>>> len(beatmaps)
3
>>> beatmaps[0]
Beatmap(beatmapset_id=1283387, beatmap_id=2665294, approved=1, total_length=226, hit_length=225, version='Reincarnation', file_md5='da2c9e1e0a56dc4d80b087eaca594767', diff_size=4.0, diff_overall=9.0, diff_approach=9.3, diff_drain=5.0, mode=0, count_normal=752, count_slider=311, count_spinner=3, submit_date=datetime.datetime(2020, 10, 22, 2, 17), approved_date=datetime.datetime(2021, 2, 22, 8, 48, 20), last_update=datetime.datetime(2021, 2, 12, 12, 27, 59), artist='Akatsuki Records', artist_unicode='暁Records', title='KARMANATIONS', title_unicode='KARMANATIONS', creator='-Rik-', creator_id=4624788, bpm=192.0, source='東方夢時空\u3000～ Pantasmagria of Dim.Dream', tags=['stack', 'japanese', 'touhou', 'vocal', 'video', 'game', 'pop', '蓬莱人形', '～', 'dolls', 'in', 'pseudo', 'paradise', 'zun', '東方project', 'mima', 'phantasmagoria', 'th3', 'blooddark/karmanations', 'c98', 'リーインカーネイション'], genre_id=2, language_id=3, favourite_count=1126, rating=9.66891, storyboard=False, video=False, download_unavailable=False, audio_unavailable=False, playcount=323434, passcount=31477, packs='S989', max_combo=1404, diff_aim=3.34499, diff_speed=2.7843, difficultyrating=6.42693)

osu_api_v1_wrapper.py

from datetime import date, datetime
from typing import Any, Dict, List, Literal, Tuple, Union, Callable, TypeVar
from typing_extensions import ParamSpec
from pydantic import BaseModel, validator
import typing
import httpx

from osu.objects.v1 import *


T = TypeVar("T")
P = ParamSpec("P")
BASE_URL = "https://osu.ppy.sh/api"


country = {
    "AF": "Afghanistan",
    "AX": "Aland Islands",
    "AL": "Albania",
    "DZ": "Algeria",
    "AS": "American Samoa",
    "AD": "Andorra",
    "AO": "Angola",
    "AI": "Anguilla",
    "AQ": "Antarctica",
    "AG": "Antigua And Barbuda",
    "AR": "Argentina",
    "AM": "Armenia",
    "AW": "Aruba",
    "AU": "Australia",
    "AT": "Austria",
    "AZ": "Azerbaijan",
    "BS": "Bahamas",
    "BH": "Bahrain",
    "BD": "Bangladesh",
    "BB": "Barbados",
    "BY": "Belarus",
    "BE": "Belgium",
    "BZ": "Belize",
    "BJ": "Benin",
    "BM": "Bermuda",
    "BT": "Bhutan",
    "BO": "Bolivia",
    "BA": "Bosnia And Herzegovina",
    "BW": "Botswana",
    "BV": "Bouvet Island",
    "BR": "Brazil",
    "IO": "British Indian Ocean Territory",
    "BN": "Brunei Darussalam",
    "BG": "Bulgaria",
    "BF": "Burkina Faso",
    "BI": "Burundi",
    "KH": "Cambodia",
    "CM": "Cameroon",
    "CA": "Canada",
    "CV": "Cape Verde",
    "KY": "Cayman Islands",
    "CF": "Central African Republic",
    "TD": "Chad",
    "CL": "Chile",
    "CN": "China",
    "CX": "Christmas Island",
    "CC": "Cocos (Keeling) Islands",
    "CO": "Colombia",
    "KM": "Comoros",
    "CG": "Congo",
    "CD": "Congo, Democratic Republic",
    "CK": "Cook Islands",
    "CR": "Costa Rica",
    "CI": 'Cote D"Ivoire',
    "HR": "Croatia",
    "CU": "Cuba",
    "CY": "Cyprus",
    "CZ": "Czech Republic",
    "DK": "Denmark",
    "DJ": "Djibouti",
    "DM": "Dominica",
    "DO": "Dominican Republic",
    "EC": "Ecuador",
    "EG": "Egypt",
    "SV": "El Salvador",
    "GQ": "Equatorial Guinea",
    "ER": "Eritrea",
    "EE": "Estonia",
    "ET": "Ethiopia",
    "FK": "Falkland Islands (Malvinas)",
    "FO": "Faroe Islands",
    "FJ": "Fiji",
    "FI": "Finland",
    "FR": "France",
    "GF": "French Guiana",
    "PF": "French Polynesia",
    "TF": "French Southern Territories",
    "GA": "Gabon",
    "GM": "Gambia",
    "GE": "Georgia",
    "DE": "Germany",
    "GH": "Ghana",
    "GI": "Gibraltar",
    "GR": "Greece",
    "GL": "Greenland",
    "GD": "Grenada",
    "GP": "Guadeloupe",
    "GU": "Guam",
    "GT": "Guatemala",
    "GG": "Guernsey",
    "GN": "Guinea",
    "GW": "Guinea-Bissau",
    "GY": "Guyana",
    "HT": "Haiti",
    "HM": "Heard Island & Mcdonald Islands",
    "VA": "Holy See (Vatican City State)",
    "HN": "Honduras",
    "HK": "Hong Kong",
    "HU": "Hungary",
    "IS": "Iceland",
    "IN": "India",
    "ID": "Indonesia",
    "IR": "Iran, Islamic Republic Of",
    "IQ": "Iraq",
    "IE": "Ireland",
    "IM": "Isle Of Man",
    "IL": "Israel",
    "IT": "Italy",
    "JM": "Jamaica",
    "JP": "Japan",
    "JE": "Jersey",
    "JO": "Jordan",
    "KZ": "Kazakhstan",
    "KE": "Kenya",
    "KI": "Kiribati",
    "KR": "Korea",
    "KP": "North Korea",
    "KW": "Kuwait",
    "KG": "Kyrgyzstan",
    "LA": 'Lao People"s Democratic Republic',
    "LV": "Latvia",
    "LB": "Lebanon",
    "LS": "Lesotho",
    "LR": "Liberia",
    "LY": "Libyan Arab Jamahiriya",
    "LI": "Liechtenstein",
    "LT": "Lithuania",
    "LU": "Luxembourg",
    "MO": "Macao",
    "MK": "Macedonia",
    "MG": "Madagascar",
    "MW": "Malawi",
    "MY": "Malaysia",
    "MV": "Maldives",
    "ML": "Mali",
    "MT": "Malta",
    "MH": "Marshall Islands",
    "MQ": "Martinique",
    "MR": "Mauritania",
    "MU": "Mauritius",
    "YT": "Mayotte",
    "MX": "Mexico",
    "FM": "Micronesia, Federated States Of",
    "MD": "Moldova",
    "MC": "Monaco",
    "MN": "Mongolia",
    "ME": "Montenegro",
    "MS": "Montserrat",
    "MA": "Morocco",
    "MZ": "Mozambique",
    "MM": "Myanmar",
    "NA": "Namibia",
    "NR": "Nauru",
    "NP": "Nepal",
    "NL": "Netherlands",
    "AN": "Netherlands Antilles",
    "NC": "New Caledonia",
    "NZ": "New Zealand",
    "NI": "Nicaragua",
    "NE": "Niger",
    "NG": "Nigeria",
    "NU": "Niue",
    "NF": "Norfolk Island",
    "MP": "Northern Mariana Islands",
    "NO": "Norway",
    "OM": "Oman",
    "PK": "Pakistan",
    "PW": "Palau",
    "PS": "Palestinian Territory, Occupied",
    "PA": "Panama",
    "PG": "Papua New Guinea",
    "PY": "Paraguay",
    "PE": "Peru",
    "PH": "Philippines",
    "PN": "Pitcairn",
    "PL": "Poland",
    "PT": "Portugal",
    "PR": "Puerto Rico",
    "QA": "Qatar",
    "RE": "Reunion",
    "RO": "Romania",
    "RU": "Russian Federation",
    "RW": "Rwanda",
    "BL": "Saint Barthelemy",
    "SH": "Saint Helena",
    "KN": "Saint Kitts And Nevis",
    "LC": "Saint Lucia",
    "MF": "Saint Martin",
    "PM": "Saint Pierre And Miquelon",
    "VC": "Saint Vincent And Grenadines",
    "WS": "Samoa",
    "SM": "San Marino",
    "ST": "Sao Tome And Principe",
    "SA": "Saudi Arabia",
    "SN": "Senegal",
    "RS": "Serbia",
    "SC": "Seychelles",
    "SL": "Sierra Leone",
    "SG": "Singapore",
    "SK": "Slovakia",
    "SI": "Slovenia",
    "SB": "Solomon Islands",
    "SO": "Somalia",
    "ZA": "South Africa",
    "GS": "South Georgia And Sandwich Isl.",
    "ES": "Spain",
    "LK": "Sri Lanka",
    "SD": "Sudan",
    "SR": "Suriname",
    "SJ": "Svalbard And Jan Mayen",
    "SZ": "Swaziland",
    "SE": "Sweden",
    "CH": "Switzerland",
    "SY": "Syrian Arab Republic",
    "TW": "Taiwan",
    "TJ": "Tajikistan",
    "TZ": "Tanzania",
    "TH": "Thailand",
    "TL": "Timor-Leste",
    "TG": "Togo",
    "TK": "Tokelau",
    "TO": "Tonga",
    "TT": "Trinidad And Tobago",
    "TN": "Tunisia",
    "TR": "Turkey",
    "TM": "Turkmenistan",
    "TC": "Turks And Caicos Islands",
    "TV": "Tuvalu",
    "UG": "Uganda",
    "UA": "Ukraine",
    "AE": "United Arab Emirates",
    "GB": "United Kingdom",
    "US": "United States",
    "UM": "United States Outlying Islands",
    "UY": "Uruguay",
    "UZ": "Uzbekistan",
    "VU": "Vanuatu",
    "VE": "Venezuela",
    "VN": "Vietnam",
    "VG": "Virgin Islands, British",
    "VI": "Virgin Islands, U.S.",
    "WF": "Wallis And Futuna",
    "EH": "Western Sahara",
    "YE": "Yemen",
    "ZM": "Zambia",
    "ZW": "Zimbabwe",
}


class Country(str):
    def get_fullname(self):
        return country.get(self)


class UserEvent(BaseModel):
    display_html: str
    beatmap_id: int = None
    beatmapset_id: int = None
    date: datetime
    epicfactor: int = None


class User(BaseModel):
    user_id: int
    username: str
    join_date: datetime
    count300: int = None
    count100: int = None
    count50: int = None
    playcount: int = None
    ranked_score: int = None
    total_score: int = None
    pp_rank: int = None
    pp_raw: float = None
    level: float = None
    accuracy: float = None
    count_rank_ss: int = None
    count_rank_ssh: int = None
    count_rank_s: int = None
    count_rank_sh: int = None
    count_rank_a: int = None
    country: Country
    total_seconds_played: int = None
    pp_country_rank: int = None
    events: List[UserEvent] = []


class Beatmap(BaseModel):
    beatmapset_id: int
    beatmap_id: int
    approved: int
    total_length: int
    hit_length: int
    version: str
    file_md5: str
    diff_size: float
    diff_overall: float
    diff_approach: float
    diff_drain: float
    mode: int
    count_normal: int
    count_slider: int
    count_spinner: int
    submit_date: datetime
    approved_date: datetime = None
    last_update: datetime
    artist: str
    artist_unicode: str
    title: str
    title_unicode: str
    creator: str
    creator_id: int
    bpm: float
    source: str
    tags: str
    genre_id: int
    language_id: int
    favourite_count: int
    rating: float
    storyboard: bool
    video: bool
    download_unavailable: bool
    audio_unavailable: bool
    playcount: int
    passcount: int
    packs: str = None
    max_combo: int = None
    diff_aim: float = None
    diff_speed: float = None
    difficultyrating: float

    @validator("tags")
    def tags_list(cls, tags: str) -> List[str]:
        return tags.split()


class ScoreBase(BaseModel):
    score: int
    count300: int
    count100: int
    count50: int
    countmiss: int
    maxcombo: int
    countkatu: int
    countgeki: int
    perfect: bool
    enabled_mods: int = None
    user_id: int
    rank: str


class ScoreWithDate(ScoreBase):
    date: datetime


class ScoreWithUsername(ScoreBase):
    username: str


class BeatmapScore(ScoreWithUsername, ScoreWithDate):
    score_id: int
    pp: float
    replay_available: bool


class RecentScore(ScoreWithDate):
    beatmap_id: int


class MultiplayerScore(ScoreBase):
    slot: int
    team: int
    passed: bool = Field(alias="pass")


class UserTopScore(RecentScore):
    score_id: int
    pp: float
    replay_available: bool


class MatchInfo(BaseModel):
    match_id: int
    name: str
    start_time: datetime
    end_time: datetime = None


class MatchGame(BaseModel):
    game_id: int
    start_time: datetime
    end_time: datetime = None
    beatmap_id: int
    play_mode: int
    match_type: int
    scoring_type: int
    team_type: int
    mods: int
    scores: List[MultiplayerScore]


class Match(BaseModel):
    match: MatchInfo
    games: List[MatchGame] = []


class Replay(BaseModel):
    content: str


def args2kwargs(args: Tuple[Any], type_hints: Dict[str, Any]):
    kwargs = {}
    for a, k in zip(args, type_hints):
        kwargs[k] = a
    return kwargs


def requester(func: Callable[P, T]) -> Callable[P, T]:
    # 獲取函式的類型提示
    type_hints = typing.get_type_hints(func)

    # 獲取函式回傳的類型提示
    return_type = type_hints["return"]

    # 獲取函式回傳的類型提示的原類型，如 List[] -> list
    return_type_origin = typing.get_origin(return_type)
    return_type_args = typing.get_args(return_type)

    async def wrapper(client: "Client", *args: P.args, **kwargs: P.kwargs) -> T:
        kwargs |= args2kwargs(args, type_hints)
        data: Union[list, dict, None] = (
            await client.request(f"/{func.__name__}", **kwargs)
        ).json()
        if return_type_origin:
            if isinstance(data, list):
                return_model = return_type_args[0]
                return list(map(lambda i: return_model(**i), data))
        if isinstance(data, dict):
            return return_type(**data)
        if data is not None:
            return return_type(data)
        else:
            return None

    return wrapper


class Client:
    def __init__(self, token: str) -> None:
        self._client = lambda: httpx.AsyncClient(base_url=BASE_URL, params={"k": token})

    async def request(self, uri: str, **params):
        async with self._client() as client:
            response = await client.get(uri, params=params)
            response.raise_for_status()
        return response

    @typing.overload
    async def get_beatmaps(self) -> List[Beatmap]:
        ...

    @typing.overload
    async def get_beatmaps(
        self,
        m: int = None,
        a: int = None,
        limit: int = None,
        mods: int = None,
    ) -> List[Beatmap]:
        ...

    @typing.overload
    async def get_beatmaps(
        self,
        s: int,
        *,
        m: int = None,
        a: int = None,
        limit: int = None,
        mods: int = None,
    ) -> List[Beatmap]:
        ...

    @typing.overload
    async def get_beatmaps(
        self,
        since: Union[date, str],
        *,
        m: int = None,
        a: int = None,
        limit: int = None,
        mods: int = None,
    ) -> List[Beatmap]:
        ...

    @typing.overload
    async def get_beatmaps(
        self,
        b: int,
        *,
        m: int = None,
        a: int = None,
        limit: int = None,
        mods: int = None,
    ) -> List[Beatmap]:
        ...

    @typing.overload
    async def get_beatmaps(
        self,
        u: int,
        *,
        type: Literal["string", "id"] = None,
        m: int = None,
        limit: int = None,
        mods: int = None,
    ) -> List[Beatmap]:
        ...

    @typing.overload
    async def get_beatmaps(
        self,
        h: str,
        *,
        m: int = None,
        a: int = None,
        mods: int = None,
    ) -> List[Beatmap]:
        ...

    @requester
    async def get_beatmaps(
        self,
        *,
        since: Union[date, str] = None,
        s: int = None,
        b: int = None,
        u: Union[str, int] = None,
        type: Literal["string", "id"] = None,
        m: int = None,
        a: int = None,
        h: str = None,
        limit: int = None,
        mods: int = None,
    ) -> List[Beatmap]:
        """### Retrieve general beatmap information.

        Args:
            since (Union[date, str], optional): return all beatmaps ranked or loved since this date. Must be a MySQL date. In UTC
            s (int, optional): specify a beatmapset_id to return metadata from.
            b (int, optional): specify a beatmap_id to return metadata from.
            u (Union[str, int], optional): specify a user_id or a username to return metadata from.
            type (Literal['string', 'id'], optional): specify if `u` is a user_id or a username. Use `string` for usernames or `id` for user_ids. Optional, default behaviour is automatic recognition (may be problematic for usernames made up of digits only).
            m (int, optional): mode (0 = osu!, 1 = Taiko, 2 = CtB, 3 = osu!mania). Optional, maps of all modes are returned by default.
            a (int, optional): specify whether converted beatmaps are included (0 = not included, 1 = included). Only has an effect if `m` is chosen and not 0. Converted maps show their converted difficulty rating. Optional, default is 0.
            h (str, optional): the beatmap hash. It can be used, for instance, if you're trying to get what beatmap has a replay played in, as .osr replays only provide beatmap hashes (example of hash: a5b99395a42bd55bc5eb1d2411cbdf8b). Optional, by default all beatmaps are returned independently from the hash.
            limit (int, optional): the amount of results. Optional, default and maximum are 500.
            mods (int, optional): mods that applies to the beatmap requested. Optional, default is 0. (Refer to the Mods section below, note that requesting multiple mods is supported, but it should not contain any non-difficulty-increasing mods or the return value will be invalid.)

        Returns:
            List[Beatmap]
        """
        pass

    @requester
    async def get_user(
        self,
        u: str,
        *,
        m: int = None,
        type: Literal["string", "id"] = None,
        event_days: int = None,
    ) -> List[User]:
        """### Retrieve general user information.

        Args:
            u (str): specify a user_id or a username to return metadata from (required).
            m (int, optional): mode (0 = osu!, 1 = Taiko, 2 = CtB, 3 = osu!mania). Optional, default value is 0.
            type (Literal['string', 'id'], optional): specify if `u` is a user_id or a username. Use `string` for usernames or id for user_ids. Optional, default behaviour is automatic recognition (may be problematic for usernames made up of digits only).
            event_days (int, optional): max number of days between now and last event date. Range of 1-31. Optional, default value is 1.

        Returns:
            List[User]
        """
        pass

    @requester
    async def get_scores(
        self,
        b: int,
        *,
        u: str = None,
        m: int = None,
        mods: int = None,
        type: Literal["string", "id"] = None,
        limit: int = None,
    ) -> List[BeatmapScore]:
        """### Retrieve information about the top 100 scores of a specified beatmap.

        Args:
            b (int): specify a beatmap_id to return score information from.
            u (str, optional): specify a user_id or a username to return score information for.
            m (int, optional): mode (0 = osu!, 1 = Taiko, 2 = CtB, 3 = osu!mania). Optional, default value is 0.
            mods (int, optional): specify a mod or mod combination (See the bitwise enum)
            type (Literal['string', 'id'], optional): specify if `u` is a user_id or a username. Use `string` for usernames or `id` for user_ids. Optional, default behaviour is automatic recognition (may be problematic for usernames made up of digits only).
            limit (int, optional):  the number of results from the top (range between 1 and 100 - defaults to 50).

        Returns:
            List[BeatmapScore]
        """
        pass

    @requester
    async def get_user_best(
        self,
        u: str,
        *,
        m: int = None,
        limit: int = None,
        type: Literal["string", "id"] = None,
    ) -> List[UserTopScore]:
        """### Get the top scores for the specified user.

        Args:
            u (str): specify a user_id or a username to return best scores from.
            m (int, optional): mode (0 = osu!, 1 = Taiko, 2 = CtB, 3 = osu!mania). Optional, default value is 0.
            limit (int, optional): amount of results (range between 1 and 100 - defaults to 10).
            type (Literal['string', 'id'], optional): specify if `u` is a user_id or a username. Use `string` for usernames or `id` for user_ids. Optional, default behaviour is automatic recognition (may be problematic for usernames made up of digits only).

        Returns:
            List[UserTopScore]
        """
        pass

    @requester
    async def get_user_recent(
        self,
        u: str,
        *,
        m: int = None,
        limit: int = None,
        type: Literal["string", "id"] = None,
    ) -> List[RecentScore]:
        """### Gets the user's ten most recent plays over the last 24 hours.

        Args:
            u (str): specify a user_id or a username to return best scores from.
            m (int, optional): mode (0 = osu!, 1 = Taiko, 2 = CtB, 3 = osu!mania). Optional, default value is 0.
            limit (int, optional): amount of results (range between 1 and 100 - defaults to 10).
            type (Literal['string', 'id'], optional): specify if `u` is a user_id or a username. Use `string` for usernames or `id` for user_ids. Optional, default behaviour is automatic recognition (may be problematic for usernames made up of digits only).

        Returns:
            List[RecentScore]
        """
        pass

    @requester
    async def get_match(self, *, mp: int) -> Match:
        """Retrieve information about a multiplayer match.

        Args:
            mp (int): match id to get information from

        Returns:
            Match
        """
        pass

    @requester
    async def get_replay(
        self,
        b: int,
        u: str,
        *,
        m: int = None,
        s: int = None,
        limit: int = None,
        type: Literal["string", "id"] = None,
    ) -> Replay:
        """Get the replay data of a user's score on a map.

        Rate limiting:
            As this is quite a load-heavy request, it has special rules about rate limiting. You are only allowed to do 10 requests per minute. Also, please note that this request is **not** intended for batch retrievals.

        Args:
            b (int): the beatmap ID (not beatmap set ID!) in which the replay was played.
            u (str): the user that has played the beatmap.
            m (int, optional): the mode the score was played in.
            s (str, optional): specify a score id to retrieve the replay data for. May be passed instead of b and u
            limit (int, optional): _description_. Defaults to None.
            type (Literal['string', 'id'], optional): specify if `u` is a user_id or a username. Use `string` for usernames or `id` for user_ids. Optional, default behaviour is automatic recognition (may be problematic for usernames made up of digits only).

        Returns:
            Replay: Note that the binary data you get when you decode above base64-string, is not the contents of an .osr-file. It is the LZMA stream referred to by the osu-wiki here: https://osu.ppy.sh/help/wiki/osu!_File_Formats/Osr_(file_format)
        """
        pass