Merge pull request #2023 from rommapp/misc/mobygames-service-adapter

misc: Create MobyGames service adapter

backend/adapters/services/mobygames.py

@@ -0,0 +1,176 @@
+import asyncio
+import http
+from collections.abc import Collection
+from typing import Literal, overload
+
+import aiohttp
+import yarl
+from adapters.services.mobygames_types import MobyGame, MobyGameBrief, MobyOutputFormat
+from aiohttp.client import ClientTimeout
+from config import MOBYGAMES_API_KEY
+from fastapi import HTTPException, status
+from logger.logger import log
+from utils.context import ctx_aiohttp_session
+
+
+async def auth_middleware(
+    req: aiohttp.ClientRequest, handler: aiohttp.ClientHandlerType
+) -> aiohttp.ClientResponse:
+    """MobyGames API authentication mechanism."""
+    req.url = req.url.update_query({"api_key": MOBYGAMES_API_KEY})
+    return await handler(req)
+
+
+class MobyGamesService:
+    """Service to interact with the MobyGames API.
+
+    Reference: https://www.mobygames.com/info/api/
+    """
+
+    def __init__(
+        self,
+        base_url: str | None = None,
+    ) -> None:
+        self.url = yarl.URL(base_url or "https://api.mobygames.com/v1")
+
+    async def _request(self, url: str, request_timeout: int = 120) -> dict:
+        aiohttp_session = ctx_aiohttp_session.get()
+        log.debug(
+            "API request: URL=%s, Timeout=%s",
+            url,
+            request_timeout,
+        )
+
+        try:
+            res = await aiohttp_session.get(
+                url,
+                middlewares=(auth_middleware,),
+                timeout=ClientTimeout(total=request_timeout),
+            )
+            res.raise_for_status()
+            return await res.json()
+        except aiohttp.ServerTimeoutError:
+            # Retry the request once if it times out
+            log.debug("Request to URL=%s timed out. Retrying...", url)
+        except aiohttp.ClientConnectionError as exc:
+            log.critical("Connection error: can't connect to MobyGames", exc_info=True)
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail="Can't connect to MobyGames, check your internet connection",
+            ) from exc
+        except aiohttp.ClientResponseError as exc:
+            if exc.status == http.HTTPStatus.UNAUTHORIZED:
+                # Sometimes MobyGames returns 401 even with a valid API key
+                log.error(exc)
+                return {}
+            elif exc.status == http.HTTPStatus.TOO_MANY_REQUESTS:
+                # Retry after 2 seconds if rate limit hit
+                await asyncio.sleep(2)
+            else:
+                # Log the error and return an empty dict if the request fails with a different code
+                log.error(exc)
+                return {}
+
+        # Retry the request once if it times out
+        try:
+            log.debug(
+                "API request: URL=%s, Timeout=%s",
+                url,
+                request_timeout,
+            )
+            res = await aiohttp_session.get(
+                url,
+                middlewares=(auth_middleware,),
+                timeout=ClientTimeout(total=request_timeout),
+            )
+            res.raise_for_status()
+            return await res.json()
+        except (aiohttp.ClientResponseError, aiohttp.ServerTimeoutError) as exc:
+            if (
+                isinstance(exc, aiohttp.ClientResponseError)
+                and exc.status == http.HTTPStatus.UNAUTHORIZED
+            ):
+                return {}
+
+            log.error(exc)
+            return {}
+
+    @overload
+    async def list_games(
+        self,
+        *,
+        game_id: int | None = ...,
+        platform_ids: Collection[int] | None = ...,
+        genre_ids: Collection[int] | None = ...,
+        group_ids: Collection[int] | None = ...,
+        title: str | None = ...,
+        output_format: Literal["id"],
+        limit: int | None = ...,
+        offset: int | None = ...,
+    ) -> list[int]: ...
+
+    @overload
+    async def list_games(
+        self,
+        *,
+        game_id: int | None = ...,
+        platform_ids: Collection[int] | None = ...,
+        genre_ids: Collection[int] | None = ...,
+        group_ids: Collection[int] | None = ...,
+        title: str | None = ...,
+        output_format: Literal["brief"],
+        limit: int | None = ...,
+        offset: int | None = ...,
+    ) -> list[MobyGameBrief]: ...
+
+    @overload
+    async def list_games(
+        self,
+        *,
+        game_id: int | None = ...,
+        platform_ids: Collection[int] | None = ...,
+        genre_ids: Collection[int] | None = ...,
+        group_ids: Collection[int] | None = ...,
+        title: str | None = ...,
+        output_format: Literal["normal"] = "normal",
+        limit: int | None = ...,
+        offset: int | None = ...,
+    ) -> list[MobyGame]: ...
+
+    async def list_games(
+        self,
+        *,
+        game_id: int | None = None,
+        platform_ids: Collection[int] | None = None,
+        genre_ids: Collection[int] | None = None,
+        group_ids: Collection[int] | None = None,
+        title: str | None = None,
+        output_format: MobyOutputFormat = "normal",
+        limit: int | None = None,
+        offset: int | None = None,
+    ) -> list[int] | list[MobyGameBrief] | list[MobyGame]:
+        """Provides a list of games matching the filters given in the query parameters, ordered by ID.
+
+        Reference: https://www.mobygames.com/info/api/#games
+        """
+        params: dict[str, list[str]] = {}
+        if game_id:
+            params["id"] = [str(game_id)]
+        if platform_ids:
+            params["platform"] = [str(id_) for id_ in platform_ids]
+        if genre_ids:
+            params["genre"] = [str(id_) for id_ in genre_ids]
+        if group_ids:
+            params["group"] = [str(id_) for id_ in group_ids]
+        if title:
+            params["title"] = [title]
+        if output_format:
+            params["format"] = [output_format]
+        if limit is not None:
+            params["limit"] = [str(limit)]
+        if offset is not None:
+            params["offset"] = [str(offset)]
+
+        url = self.url.joinpath("games").with_query(**params)
+        response = await self._request(str(url))
+        return response.get("games", [])

backend/adapters/services/mobygames_types.py

@@ -0,0 +1,60 @@
+from typing import Literal, TypedDict
+
+MobyOutputFormat = Literal["id", "brief", "normal"]
+
+
+class MobyGameAlternateTitle(TypedDict):
+    description: str
+    title: str
+
+
+class MobyGenre(TypedDict):
+    genre_category: str
+    genre_category_id: int
+    genre_id: int
+    genre_name: str
+
+
+class MobyPlatform(TypedDict):
+    first_release_date: str
+    platform_id: int
+    platform_name: str
+
+
+class MobyGameCover(TypedDict):
+    height: int
+    image: str
+    platforms: list[str]
+    thumbnail_image: str
+    width: int
+
+
+class MobyGameScreenshot(TypedDict):
+    caption: str
+    height: int
+    image: str
+    thumbnail_image: str
+    width: int
+
+
+# https://www.mobygames.com/info/api/#games
+class MobyGameBrief(TypedDict):
+    game_id: int
+    moby_url: str
+    title: str
+
+
+# https://www.mobygames.com/info/api/#games
+class MobyGame(TypedDict):
+    alternate_titles: list[MobyGameAlternateTitle]
+    description: str
+    game_id: int
+    genres: list[MobyGenre]
+    moby_score: float
+    moby_url: str
+    num_votes: int
+    official_url: str | None
+    platforms: list[MobyPlatform]
+    sample_cover: MobyGameCover
+    sample_screenshots: list[MobyGameScreenshot]
+    title: str

backend/handler/metadata/moby_handler.py

@@ -1,18 +1,11 @@
-import asyncio
-import http
-import json
 import re
 from typing import Final, NotRequired, TypedDict
 from urllib.parse import quote
 
-import httpx
-import pydash
-import yarl
+from adapters.services.mobygames import MobyGamesService
+from adapters.services.mobygames_types import MobyGame
 from config import MOBYGAMES_API_KEY
-from fastapi import HTTPException, status
-from logger.logger import log
 from unidecode import unidecode as uc
-from utils.context import ctx_httpx_client
 
 from .base_hander import (
     PS2_OPL_REGEX,
@@ -59,12 +52,14 @@ class MobyGamesRom(TypedDict):
     moby_metadata: NotRequired[MobyMetadata]
 
 
-def extract_metadata_from_moby_rom(rom: dict) -> MobyMetadata:
+def extract_metadata_from_moby_rom(rom: MobyGame) -> MobyMetadata:
     return MobyMetadata(
         {
             "moby_score": str(rom.get("moby_score", "")),
-            "genres": rom.get("genres.genre_name", []),
-            "alternate_titles": rom.get("alternate_titles.title", []),
+            "genres": [genre["genre_name"] for genre in rom.get("genres", [])],
+            "alternate_titles": [
+                alt["title"] for alt in rom.get("alternate_titles", [])
+            ],
             "platforms": [
                 {
                     "moby_id": p["platform_id"],
@@ -78,104 +73,33 @@ def extract_metadata_from_moby_rom(rom: dict) -> MobyMetadata:
 
 class MobyGamesHandler(MetadataHandler):
     def __init__(self) -> None:
-        self.BASE_URL = "https://api.mobygames.com/v1"
-        self.platform_endpoint = f"{self.BASE_URL}/platforms"
-        self.games_endpoint = f"{self.BASE_URL}/games"
+        self.moby_service = MobyGamesService()
 
-    async def _request(self, url: str, timeout: int = 120) -> dict:
-        httpx_client = ctx_httpx_client.get()
-        authorized_url = yarl.URL(url).update_query(api_key=MOBYGAMES_API_KEY)
-        masked_url = authorized_url.with_query(
-            self._mask_sensitive_values(dict(authorized_url.query))
-        )
-
-        log.debug(
-            "API request: URL=%s, Timeout=%s",
-            masked_url,
-            timeout,
-        )
-
-        try:
-            res = await httpx_client.get(str(authorized_url), timeout=timeout)
-            res.raise_for_status()
-            return res.json()
-        except httpx.NetworkError as exc:
-            log.critical("Connection error: can't connect to Mobygames", exc_info=True)
-            raise HTTPException(
-                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-                detail="Can't connect to Mobygames, check your internet connection",
-            ) from exc
-        except httpx.HTTPStatusError as exc:
-            if exc.response.status_code == http.HTTPStatus.UNAUTHORIZED:
-                # Sometimes Mobygames returns 401 even with a valid API key
-                log.error(exc)
-                return {}
-            elif exc.response.status_code == http.HTTPStatus.TOO_MANY_REQUESTS:
-                # Retry after 2 seconds if rate limit hit
-                await asyncio.sleep(2)
-            else:
-                # Log the error and return an empty dict if the request fails with a different code
-                log.error(exc)
-                return {}
-        except json.decoder.JSONDecodeError as exc:
-            # Log the error and return an empty list if the response is not valid JSON
-            log.error(exc)
-            return {}
-        except httpx.TimeoutException:
-            log.debug(
-                "Request to URL=%s timed out. Retrying with URL=%s", masked_url, url
-            )
-
-        # Retry the request once if it times out
-        try:
-            log.debug(
-                "API request: URL=%s, Timeout=%s",
-                url,
-                timeout,
-            )
-            res = await httpx_client.get(url, timeout=timeout)
-            res.raise_for_status()
-            return res.json()
-        except (
-            httpx.HTTPStatusError,
-            httpx.TimeoutException,
-            json.decoder.JSONDecodeError,
-        ) as exc:
-            if (
-                isinstance(exc, httpx.HTTPStatusError)
-                and exc.response.status_code == http.HTTPStatus.UNAUTHORIZED
-            ):
-                # Sometimes Mobygames returns 401 even with a valid API key
-                return {}
-
-            # Log the error and return an empty dict if the request fails with a different code
-            log.error(exc)
-            return {}
-
-    async def _search_rom(self, search_term: str, platform_moby_id: int) -> dict | None:
+    async def _search_rom(
+        self, search_term: str, platform_moby_id: int
+    ) -> MobyGame | None:
         if not platform_moby_id:
             return None
 
         search_term = uc(search_term)
-        url = yarl.URL(self.games_endpoint).with_query(
-            platform=[platform_moby_id],
+        roms = await self.moby_service.list_games(
+            platform_ids=[platform_moby_id],
             title=quote(search_term, safe="/ "),
         )
-        roms = (await self._request(str(url))).get("games", [])
+        if not roms:
+            return None
 
-        exact_matches = [
-            rom
-            for rom in roms
+        # Find an exact match.
+        search_term_casefold = search_term.casefold()
+        search_term_normalized = self._normalize_exact_match(search_term)
+        for rom in roms:
             if (
-                rom["title"].lower() == search_term.lower()
-                or (
-                    self._normalize_exact_match(rom["title"])
-                    == self._normalize_exact_match(search_term)
-                )
-            )
-        ]
+                rom["title"].casefold() == search_term_casefold
+                or self._normalize_exact_match(rom["title"]) == search_term_normalized
+            ):
+                return rom
 
-        return pydash.get(exact_matches or roms, "[0]", None)
+        return roms[0]
 
     def get_platform(self, slug: str) -> MobyGamesPlatform:
         platform = MOBYGAMES_PLATFORM_LIST.get(slug, None)
@@ -280,7 +204,7 @@ class MobyGamesHandler(MetadataHandler):
             "moby_id": res["game_id"],
             "name": res["title"],
             "summary": res.get("description", ""),
-            "url_cover": pydash.get(res, "sample_cover.image", ""),
+            "url_cover": res.get("sample_cover", {}).get("image", ""),
             "url_screenshots": [s["image"] for s in res.get("sample_screenshots", [])],
             "moby_metadata": extract_metadata_from_moby_rom(res),
         }
@@ -291,18 +215,16 @@ class MobyGamesHandler(MetadataHandler):
         if not MOBY_API_ENABLED:
             return MobyGamesRom(moby_id=None)
 
-        url = yarl.URL(self.games_endpoint).with_query(id=moby_id)
-        roms = (await self._request(str(url))).get("games", [])
-        res = pydash.get(roms, "[0]", None)
-
-        if not res:
+        roms = await self.moby_service.list_games(game_id=moby_id)
+        if not roms:
             return MobyGamesRom(moby_id=None)
 
+        res = roms[0]
         rom = {
             "moby_id": res["game_id"],
             "name": res["title"],
             "summary": res.get("description", None),
-            "url_cover": pydash.get(res, "sample_cover.image", None),
+            "url_cover": res.get("sample_cover", {}).get("image", None),
             "url_screenshots": [s["image"] for s in res.get("sample_screenshots", [])],
             "moby_metadata": extract_metadata_from_moby_rom(res),
         }
@@ -326,10 +248,10 @@ class MobyGamesHandler(MetadataHandler):
             return []
 
         search_term = uc(search_term)
-        url = yarl.URL(self.games_endpoint).with_query(
-            platform=[platform_moby_id], title=quote(search_term, safe="/ ")
+        matched_roms = await self.moby_service.list_games(
+            platform_ids=[platform_moby_id],
+            title=quote(search_term, safe="/ "),
         )
-        matched_roms = (await self._request(str(url))).get("games", [])
 
         return [
             MobyGamesRom(
@@ -339,7 +261,7 @@ class MobyGamesHandler(MetadataHandler):
                         "moby_id": rom["game_id"],
                         "name": rom["title"],
                         "summary": rom.get("description", ""),
-                        "url_cover": pydash.get(rom, "sample_cover.image", ""),
+                        "url_cover": rom.get("sample_cover", {}).get("image", ""),
                         "url_screenshots": [
                             s["image"] for s in rom.get("sample_screenshots", [])
                         ],