From 4209ee4481d16b7bcc068c4fe22330c786f0dc96 Mon Sep 17 00:00:00 2001 From: Michael Manganiello Date: Fri, 27 Jun 2025 01:15:49 -0300 Subject: [PATCH] misc: Create MobyGames service adapter Add a new service adapter for the MobyGames API, to separate concerns with RomM's handler for metadata. This adapter is agnostic to the handler and only provides methods to interact with the API, and correctly return typed responses. --- backend/adapters/services/mobygames.py | 176 +++++++++++++++++++ backend/adapters/services/mobygames_types.py | 60 +++++++ backend/handler/metadata/moby_handler.py | 144 ++++----------- 3 files changed, 269 insertions(+), 111 deletions(-) create mode 100644 backend/adapters/services/mobygames.py create mode 100644 backend/adapters/services/mobygames_types.py diff --git a/backend/adapters/services/mobygames.py b/backend/adapters/services/mobygames.py new file mode 100644 index 000000000..28ca8bfb7 --- /dev/null +++ b/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", []) diff --git a/backend/adapters/services/mobygames_types.py b/backend/adapters/services/mobygames_types.py new file mode 100644 index 000000000..b61f7d4c0 --- /dev/null +++ b/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 diff --git a/backend/handler/metadata/moby_handler.py b/backend/handler/metadata/moby_handler.py index 03b49ccf3..1a9f49300 100644 --- a/backend/handler/metadata/moby_handler.py +++ b/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", []) ],