refactor: favourites repository

app/api/domains/osu.py

@@ -62,6 +62,7 @@
 from app.objects.score import Score
 from app.objects.score import SubmissionStatus
 from app.repositories import comments as comments_repo
+from app.repositories import favourites as favourites_repo
 from app.repositories import mail as mail_repo
 from app.repositories import maps as maps_repo
 from app.repositories import players as players_repo
@@ -236,12 +237,11 @@ async def osuGetBeatmapInfo(
 async def osuGetFavourites(
     player: Player = Depends(authenticate_player_session(Query, "u", "h")),
 ) -> Response:
-    rows = await app.state.services.database.fetch_all(
-        "SELECT setid FROM favourites WHERE userid = :user_id",
-        {"user_id": player.id},
-    )
+    favourites = await favourites_repo.fetch_all(userid=player.id)
 
-    return Response("\n".join([str(row["setid"]) for row in rows]).encode())
+    return Response(
+        "\n".join([str(favourite["setid"]) for favourite in favourites]).encode(),
+    )
 
 
 @router.get("/web/osu-addfavourite.php")
@@ -250,16 +250,13 @@ async def osuAddFavourite(
     map_set_id: int = Query(..., alias="a"),
 ) -> Response:
     # check if they already have this favourited.
-    if await app.state.services.database.fetch_one(
-        "SELECT 1 FROM favourites WHERE userid = :user_id AND setid = :set_id",
-        {"user_id": player.id, "set_id": map_set_id},
-    ):
+    if await favourites_repo.fetch_one(player.id, map_set_id):
         return Response(b"You've already favourited this beatmap!")
 
     # add favourite
-    await app.state.services.database.execute(
-        "INSERT INTO favourites VALUES (:user_id, :set_id, UNIX_TIMESTAMP())",
-        {"user_id": player.id, "set_id": map_set_id},
+    await favourites_repo.create(
+        userid=player.id,
+        setid=map_set_id,
     )
 
     return Response(b"Added favourite!")

app/repositories/favourites.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import textwrap
+from datetime import datetime
+from typing import Any
+from typing import TypedDict
+from typing import cast
+
+import app.state.services
+
+# +------------+------+------+-----+---------+-------+
+# | Field      | Type | Null | Key | Default | Extra |
+# +------------+------+------+-----+---------+-------+
+# | userid     | int  | NO   | PRI | NULL    |       |
+# | setid      | int  | NO   | PRI | NULL    |       |
+# | created_at | int  | NO   |     | 0       |       |
+# +------------+------+------+-----+---------+-------+
+
+READ_PARAMS = textwrap.dedent(
+    """\
+        userid, setid, created_at
+    """,
+)
+
+
+class Favourite(TypedDict):
+    userid: int
+    setid: int
+    created_at: int
+
+
+async def create(
+    userid: int,
+    setid: int,
+) -> Favourite:
+    """Create a new favourite mapset entry in the database."""
+    query = f"""\
+        INSERT INTO favourites (userid, setid, created_at)
+             VALUES (:userid, :setid, UNIX_TIMESTAMP())
+    """
+    params: dict[str, Any] = {
+        "userid": userid,
+        "setid": setid,
+    }
+    rec_id = await app.state.services.database.execute(query, params)
+
+    query = f"""\
+        SELECT {READ_PARAMS}
+          FROM favourites
+         WHERE userid = :userid
+           AND setid = :setid
+    """
+    params = {
+        "userid": userid,
+        "setid": setid,
+    }
+    favourite = await app.state.services.database.fetch_one(query, params)
+
+    assert favourite is not None
+    return cast(Favourite, dict(favourite._mapping))
+
+
+async def fetch_all(userid: int) -> list[Favourite]:
+    """Fetch all favourites from a player."""
+    query = f"""\
+        SELECT {READ_PARAMS}
+          FROM favourites
+         WHERE userid = :userid
+    """
+    params: dict[str, Any] = {
+        "userid": userid,
+    }
+
+    favourites = await app.state.services.database.fetch_all(query, params)
+    return cast(list[Favourite], [dict(f._mapping) for f in favourites])
+
+
+async def fetch_one(userid: int, setid: int) -> Favourite | None:
+    """Check if a mapset is already a favourite."""
+    query = f"""\
+        SELECT {READ_PARAMS}
+          FROM favourites
+         WHERE userid = :userid
+           AND setid = :setid
+    """
+    params: dict[str, Any] = {
+        "userid": userid,
+        "setid": setid,
+    }
+
+    favourite = await app.state.services.database.fetch_one(query, params)
+    return cast(Favourite, dict(favourite._mapping)) if favourite else None