Skip to content

Helpers

There are lots of helper functions in Phillip as it tries to have as little dependencies as possible. This will help you access requests data from osu! and process them.

osu!v v1 API

In this section we will only bring functions to help you with osu! API. (v1, obviously.)


phillip.osu.old.api.APIClient

get_api(self, endpoint, **kwargs) async

Request something based on endpoint. This function is a coroutine.

Parameters:

  • endpoint - str -- The API endpoint, reference could be found in osu!wiki.
  • **kwargs - dict | optional -- Keyword arguments that will be passed as a query string.

Raises:

  • Exception -- If API key is not assigned.

Returns

  • List[dict] -- API response.
Source code in phillip\osu\old\api.py
async def get_api(self, endpoint: str, **kwargs) -> List[dict]:
    """Request something based on endpoint. *This function is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutine).*

    **Parameters:**

    * endpoint - `str` -- The API endpoint, reference could be found in [osu!wiki](https://github.com/ppy/osu-api/wiki).
    * \*\*kwargs - `dict` | optional -- Keyword arguments that will be passed as a query string.

    **Raises:**

    * `Exception` -- If API key is not assigned.

    **Returns**

    * `List[dict]` -- API response.
    """
    kwargs["k"] = self._key

    api_args = urlencode(kwargs)
    api_url = self.BASE_URL + endpoint + "?" + api_args

    async with self._session.get(api_url) as api_res:
        return await api_res.json()

get_beatmaps(self, **kwargs) async

Get beatmapset from osu! API. This function is a coroutine.

Returns

  • List[Beatmap] -- Beatmapsets fetched from API.
Source code in phillip\osu\old\api.py
async def get_beatmaps(self, **kwargs) -> List[Beatmap]:
    """Get beatmapset from osu! API. *This function is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutine).*

    **Returns**

    * `List[Beatmap]` -- Beatmapsets fetched from API.
    """
    return [Beatmap(map) for map in await self.get_api("get_beatmaps", **kwargs)]

osu!web

In this section, we will scrape the site in order to get data from osu!

Warning

A throttler is automatically initiated to rate limit your request to 2 requests/minute in order to avoid getting IP banned. You may change this but in general this is highly discouraged.


phillip.osu.new.abstract.ABCClient

get_events(self, nominate=True, rank=True, love=True, nomination_reset=True, disqualify=True, **kwargs)

Get events of from osu!website. This function is a coroutine.

Parameters:

  • nominate - bool -- Whether to get nomination events or not. This implies qualify event.
  • rank - bool -- Whether to get ranked events or not.
  • love - bool -- Whetger to get loved events or not.
  • nomination_reset - bool -- Whether to get nomination reset events or not.
  • disqualify -- bool -- Whether to get disqualification events or not.

Yields:

  • list abstract.EventBase -- List of events resulted from fetching osu!web, with next index as next event that will be processed.
Source code in phillip\osu\new\abstract.py
async def get_events(
    self,
    nominate: bool = True,
    rank: bool = True,
    love: bool = True,
    nomination_reset: bool = True,
    disqualify: bool = True,
    **kwargs,
) -> AsyncGenerator[Type[abstract.EventBase], None]:
    """Get events of from osu!website. *This function is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutine).*

    **Parameters:**

    * nominate - `bool` -- Whether to get nomination events or not. **This implies `qualify` event.**
    * rank - `bool` -- Whether to get ranked events or not.
    * love - `bool` -- Whetger to get loved events or not.
    * nomination_reset - `bool` -- Whether to get nomination reset events or not.
    * disqualify -- `bool` -- Whether to get disqualification events or not.

    **Yields:**

    * list `abstract.EventBase` -- List of events resulted from fetching osu!web, \
         with next index as next event that will be processed.
    """
    additions = list()
    types_val = [nominate, rank, love, nomination_reset, disqualify]

    for i in range(5):
        additions.append(types_val[i] and self.TYPES[i] or str())
    if types_val[0]:
        additions.append("qualify")
    extras = urlencode(kwargs)
    url = self.events_url + "&types%5B%5D=".join(additions) + "&" + extras

    events = await self.get_json(url, "json-events")
    events.reverse()  # type: ignore

    event_cases = {
        "nominate": classes.Nominated,
        "disqualify": classes.Disqualified,
        "nomination_reset": classes.Popped,
        "rank": classes.Ranked,
        "love": classes.Loved,
    }

    for i, event in enumerate(events):
        action = event["type"]  # type: ignore
        if action == "qualify":
            continue  # Skip qualified event news

        next_map = None
        if i + 1 != len(events):
            next_map = events[i + 1]  # type: ignore

        yield event_cases[action](event, next_map, app=self._app)  # type: ignore

get_users(self, group_id) async

Get users inside of a group. This function is a coroutine.

Parameters:

  • group_id - int -- The group id.

Returns

  • List[dict] -- A dictionary containing users' data.
Source code in phillip\osu\new\abstract.py
async def get_users(self, group_id: int) -> List[GroupUser]:
    """Get users inside of a group. *This function is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutine).*

    **Parameters:**

    * group_id - `int` -- The group id.

    **Returns**

    * `List[dict]` -- A dictionary containing users' data.
    """
    uri = self.groups_url + str(group_id)
    users_json = await self.get_json(uri, "json-users")

    out = []
    for user in users_json:
        out.append(GroupUser(user))
    return out

nomination_history(self, mapid) async

Get nomination history of a beatmap. This function is a coroutine.

Parameters:

  • mapid - int -- Beatmapset ID to gather.

Returns

  • parent - List[child] -- A list containing child tuples.
    • child - Tuple[str, int] -- A tuple with a string of event type and user id of user triggering the event.
Source code in phillip\osu\new\abstract.py
async def nomination_history(self, mapid: int) -> List[Tuple[str, int]]:
    """Get nomination history of a beatmap. *This function is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutine).*

    **Parameters:**

    * mapid - `int` -- Beatmapset ID to gather.

    **Returns**

    * parent - `List[child]` -- A list containing child tuples.
        * `child` - `Tuple[str, int]` -- A tuple with a string of event type and user id of user triggering the event.
    """
    uri = f"https://osu.ppy.sh/beatmapsets/{str(mapid)}/discussion"
    set_json = await self.get_json(uri, "json-beatmapset-discussion")
    js = set_json["beatmapset"]["events"]  # type: ignore

    history = []
    for i, event in enumerate(js):
        if i + 1 != len(js):
            next_event = js[i + 1]
        if event["type"] in self.EVENTS:
            event_name = self.EVENTS[event["type"]]
            if next_event["type"] == "qualify":
                event_name = "Qualified"
            history.append((event_name, event["user_id"]))
    return history

phillip.osu.new.web.WebClient

get_html(self, uri) async

Receive html from uri with rate limit. This function is a coroutine.

Parameters:

  • uri - str -- URL of discussion page.

Returns

  • BeautifulSoup -- Soup'd html response.
Source code in phillip\osu\new\web.py
async def get_html(self, uri: str) -> BeautifulSoup:
    """Receive html from uri with rate limit. 
    *This function is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutine).*

    **Parameters:**

    * uri - `str` -- URL of discussion page.

    **Returns**

    * BeautifulSoup -- Soup'd html response.
    """
    async with self._throttler:
        async with self._session.get(uri, cookies={"locale": "en"}) as site_html:
            return BeautifulSoup(await site_html.text(), features="html.parser")