Custom Ranked Akatsuki Maps API

app/api/public/akatsuki.py

@@ -0,0 +1,23 @@
+"""\
+Provides an API exposing Akatsuki's beatmaps, which
+include internal state such as ranked status updates.
+"""
+
+import logging
+
+from fastapi import APIRouter
+from fastapi import Header
+from fastapi import Response
+
+from app.api.responses import JSONResponse
+from app.usecases import akatsuki_beatmaps
+
+router = APIRouter(tags=["Akatsuki Beatmaps"])
+
+
+@router.get("/api/akatsuki/v1/beatmaps/custom-ranked-beatmaps")
+async def fetch_all_custom_ranked_beatmaps() -> Response:
+    beatmaps = await akatsuki_beatmaps.fetch_all_custom_ranked_beatmaps()
+    return JSONResponse(
+        content=[AkatsukiBeatmap.model_dump() for AkatsukiBeatmap in beatmaps],
+    )

app/repositories/akatsuki_beatmaps.py

@@ -137,6 +137,43 @@ async def fetch_one_by_id(beatmap_id: int, /) -> AkatsukiBeatmap | None:
     )
 
 
+async def fetch_many_maps_with_custom_akatsuki_status() -> list[AkatsukiBeatmap]:
+    query = """\
+        SELECT * FROM beatmaps WHERE ranked_status_freezed = 1 ORDER BY lastest_updated
+    """
+    rec = await state.database.fetch_all(query)
+    akatsuki_maps_with_custom_status: list[AkatsukiBeatmap] = []
+    for akatsuki_beatmap in rec:
+        beatmap = AkatsukiBeatmap(
+            beatmap_id=akatsuki_beatmap["beatmap_id"],
+            beatmapset_id=akatsuki_beatmap["beatmapset_id"],
+            beatmap_md5=akatsuki_beatmap["beatmap_md5"],
+            song_name=akatsuki_beatmap["song_name"],
+            file_name=akatsuki_beatmap["file_name"],
+            ar=akatsuki_beatmap["ar"],
+            od=akatsuki_beatmap["od"],
+            mode=akatsuki_beatmap["mode"],
+            max_combo=akatsuki_beatmap["max_combo"],
+            hit_length=akatsuki_beatmap["hit_length"],
+            bpm=akatsuki_beatmap["bpm"],
+            ranked=akatsuki_beatmap["ranked"],
+            latest_update=akatsuki_beatmap["latest_update"],
+            ranked_status_freezed=akatsuki_beatmap["ranked_status_freezed"],
+            playcount=akatsuki_beatmap["playcount"],
+            passcount=akatsuki_beatmap["passcount"],
+            rankedby=akatsuki_beatmap["rankedby"],
+            rating=akatsuki_beatmap["rating"],
+            bancho_ranked_status=akatsuki_beatmap["bancho_ranked_status"],
+            count_circles=akatsuki_beatmap["count_circles"],
+            count_spinners=akatsuki_beatmap["count_spinners"],
+            count_sliders=akatsuki_beatmap["count_sliders"],
+            bancho_creator_id=akatsuki_beatmap["bancho_creator_id"],
+            bancho_creator_name=akatsuki_beatmap["bancho_creator_name"],
+        )
+        akatsuki_maps_with_custom_status.append(beatmap)
+    return akatsuki_maps_with_custom_status
+
+
 async def create_or_replace(beatmap: AkatsukiBeatmap) -> AkatsukiBeatmap:
     query = """\
         REPLACE INTO beatmaps (

app/usecases/akatsuki_beatmaps.py

@@ -205,3 +205,8 @@ async def fetch_one_by_md5(beatmap_md5: str) -> AkatsukiBeatmap | None:
             )
 
     return beatmap
+
+
+async def fetch_all_custom_ranked_beatmaps() -> list[AkatsukiBeatmap]:
+    beatmaps = await akatsuki_beatmaps.fetch_many_maps_with_custom_akatsuki_status()
+    return beatmaps