From bf9fb12f917ae0f22b331d8a2b738a3f5ebedf5e Mon Sep 17 00:00:00 2001 From: Michael Manganiello Date: Fri, 20 Jun 2025 12:32:58 -0300 Subject: [PATCH] misc: Create SteamGridDB service adapter Add a new service adapter for the SteamGridDB 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/steamgriddb.py | 175 ++++++++++++++++++ .../adapters/services/steamgriddb_types.py | 75 ++++++++ backend/handler/metadata/sgdb_handler.py | 118 +++--------- 3 files changed, 272 insertions(+), 96 deletions(-) create mode 100644 backend/adapters/services/steamgriddb.py create mode 100644 backend/adapters/services/steamgriddb_types.py diff --git a/backend/adapters/services/steamgriddb.py b/backend/adapters/services/steamgriddb.py new file mode 100644 index 000000000..d4486b86a --- /dev/null +++ b/backend/adapters/services/steamgriddb.py @@ -0,0 +1,175 @@ +import http +import itertools +import json +from collections.abc import AsyncIterator, Collection +from typing import Literal, cast + +import aiohttp +import yarl +from adapters.services.steamgriddb_types import ( + SGDBDimension, + SGDBGame, + SGDBGrid, + SGDBGridList, + SGDBMime, + SGDBStyle, + SGDBTag, + SGDBType, +) +from aiohttp.client import ClientTimeout +from config import STEAMGRIDDB_API_KEY +from exceptions.endpoint_exceptions import SGDBInvalidAPIKeyException +from logger.logger import log +from utils.context import ctx_aiohttp_session + + +async def auth_middleware( + req: aiohttp.ClientRequest, handler: aiohttp.ClientHandlerType +) -> aiohttp.ClientResponse: + """SteamGridDB API authentication mechanism.""" + req.headers["Authorization"] = f"Bearer {STEAMGRIDDB_API_KEY}" + return await handler(req) + + +class SteamGridDBService: + """Service to interact with the SteamGridDB API. + + Reference: https://www.steamgriddb.com/api/v2 + """ + + def __init__( + self, + base_url: str | None = None, + ) -> None: + self.url = yarl.URL(base_url or "https://steamgriddb.com/api/v2") + + 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.ClientResponseError as exc: + if exc.status == http.HTTPStatus.UNAUTHORIZED: + raise SGDBInvalidAPIKeyException from exc + # 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.error( + "Failed to decode JSON response from SteamGridDB: %s", + str(exc), + ) + return {} + + async def get_grids_for_game( + self, + game_id: int, + *, + styles: Collection[SGDBStyle] | None = None, + dimensions: Collection[SGDBDimension] | None = None, + mimes: Collection[SGDBMime] | None = None, + types: Collection[SGDBType] | None = None, + any_of_tags: Collection[SGDBTag] | None = None, + is_nsfw: bool | Literal["any"] | None = None, + is_humor: bool | Literal["any"] | None = None, + is_epilepsy: bool | Literal["any"] | None = None, + limit: int | None = None, + page_number: int | None = None, + ) -> SGDBGridList: + """Retrieve grids by game ID. + + Reference: https://www.steamgriddb.com/api/v2#tag/GRIDS/operation/getGridsByGameId + """ + params: dict[str, list[str]] = {} + if styles: + params["styles"] = [",".join(styles)] + if dimensions: + params["dimensions"] = [",".join(dimensions)] + if mimes: + params["mimes"] = [",".join(mimes)] + if types: + params["types"] = [",".join(types)] + if any_of_tags: + params["oneoftag"] = [",".join(any_of_tags)] + if is_nsfw is not None: + params["nsfw"] = [str(is_nsfw).lower()] + if is_humor is not None: + params["humor"] = [str(is_humor).lower()] + if is_epilepsy is not None: + params["epilepsy"] = [str(is_epilepsy).lower()] + if limit is not None: + params["limit"] = [str(limit)] + if page_number is not None: + params["page"] = [str(page_number)] + + url = self.url.joinpath("grids/game", str(game_id)).with_query(**params) + response = await self._request(str(url)) + if not response: + return SGDBGridList( + page=0, + total=0, + limit=limit or 50, + data=[], + ) + return cast(SGDBGridList, response) + + async def iter_grids_for_game( + self, + game_id: int, + *, + styles: Collection[SGDBStyle] | None = None, + dimensions: Collection[SGDBDimension] | None = None, + mimes: Collection[SGDBMime] | None = None, + types: Collection[SGDBType] | None = None, + any_of_tags: Collection[SGDBTag] | None = None, + is_nsfw: bool | Literal["any"] | None = None, + is_humor: bool | Literal["any"] | None = None, + is_epilepsy: bool | Literal["any"] | None = None, + ) -> AsyncIterator[SGDBGrid]: + """Iterate through grids by game ID. + + Reference: https://www.steamgriddb.com/api/v2#tag/GRIDS/operation/getGridsByGameId + """ + page_size = 50 # Maximum page size for this endpoint. + offset = 0 + + for page_number in itertools.count(start=0): + response = await self.get_grids_for_game( + game_id, + styles=styles, + dimensions=dimensions, + mimes=mimes, + types=types, + any_of_tags=any_of_tags, + is_nsfw=is_nsfw, + is_humor=is_humor, + is_epilepsy=is_epilepsy, + limit=page_size, + page_number=page_number, + ) + results = response["data"] + for result in results: + yield result + + offset += len(results) + if len(results) < page_size or offset >= response["total"]: + break + + async def search_games(self, term: str) -> list[SGDBGame]: + """Search for games by name. + + Reference: https://www.steamgriddb.com/api/v2#tag/SEARCH/operation/searchGrids + """ + url = self.url.joinpath("search/autocomplete", term) + response = await self._request(str(url)) + return cast(list[SGDBGame], response.get("data", [])) diff --git a/backend/adapters/services/steamgriddb_types.py b/backend/adapters/services/steamgriddb_types.py new file mode 100644 index 000000000..bade565f4 --- /dev/null +++ b/backend/adapters/services/steamgriddb_types.py @@ -0,0 +1,75 @@ +import enum +from typing import Mapping, TypedDict + + +@enum.unique +class SGDBStyle(enum.StrEnum): + ALTERNATE = "alternate" + BLURRED = "blurred" + WHITE_LOGO = "white_logo" + MATERIAL = "material" + NO_LOGO = "no_logo" + + +@enum.unique +class SGDBDimension(enum.StrEnum): + STEAM_HORIZONTAL = "460x215" + STEAM_HORIZONTAL_2X = "920x430" + STEAM_VERTICAL = "600x900" + GOG_GALAXY_TILE = "342x482" + GOG_GALAXY_COVER = "660x930" + SQUARE_512 = "512x512" + SQUARE_1024 = "1024x1024" + + +@enum.unique +class SGDBMime(enum.StrEnum): + PNG = "image/png" + JPEG = "image/jpeg" + WEBP = "image/webp" + + +@enum.unique +class SGDBType(enum.StrEnum): + STATIC = "static" + ANIMATED = "animated" + + +@enum.unique +class SGDBTag(enum.StrEnum): + HUMOR = "humor" + NSFW = "nsfw" + EPILEPSY = "epilepsy" + + +class PaginatedResponse[T: Mapping](TypedDict): + page: int + total: int + limit: int + data: list[T] + + +class SGDBGridAuthor(TypedDict): + name: str + steam64: str + avatar: str + + +class SGDBGrid(TypedDict): + id: int + score: int + style: SGDBStyle + url: str + thumb: str + tags: list[str] + author: SGDBGridAuthor + + +class SGDBGame(TypedDict): + id: int + name: str + types: list[str] + verified: bool + + +SGDBGridList = PaginatedResponse[SGDBGrid] diff --git a/backend/handler/metadata/sgdb_handler.py b/backend/handler/metadata/sgdb_handler.py index 1ea113c3e..77c0baf3f 100644 --- a/backend/handler/metadata/sgdb_handler.py +++ b/backend/handler/metadata/sgdb_handler.py @@ -1,79 +1,30 @@ import asyncio -import itertools -import json from typing import Any, Final +from adapters.services.steamgriddb import SteamGridDBService +from adapters.services.steamgriddb_types import SGDBDimension, SGDBType from config import STEAMGRIDDB_API_KEY -from exceptions.endpoint_exceptions import SGDBInvalidAPIKeyException from logger.logger import log -from utils.context import ctx_httpx_client from .base_hander import MetadataHandler # Used to display the Mobygames API status in the frontend STEAMGRIDDB_API_ENABLED: Final = bool(STEAMGRIDDB_API_KEY) -# SteamGridDB dimensions -STEAMVERTICAL: Final = "600x900" -GALAXY342: Final = "342x482" -GALAXY660: Final = "660x930" -SQUARE512: Final = "512x512" -SQUARE1024: Final = "1024x1024" - -# SteamGridDB types -STATIC: Final = "static" -ANIMATED: Final = "animated" - -SGDB_API_COVER_LIMIT: Final = 50 - class SGDBBaseHandler(MetadataHandler): def __init__(self) -> None: - self.BASE_URL = "https://www.steamgriddb.com/api/v2" - self.search_endpoint = f"{self.BASE_URL}/search/autocomplete" - self.grid_endpoint = f"{self.BASE_URL}/grids/game" - self.headers = { - "Authorization": f"Bearer {STEAMGRIDDB_API_KEY}", - "Accept": "*/*", - } - self.masked_headers = self._mask_sensitive_values(self.headers) - self.timeout = 120 + self.sgdb_service = SteamGridDBService() async def get_details(self, search_term: str) -> list[dict[str, Any]]: - httpx_client = ctx_httpx_client.get() - url = f"{self.search_endpoint}/{search_term}" - - log.debug( - "API request: Method=GET, URL=%s, Headers=%s, Timeout=%s", - url, - self.masked_headers, - self.timeout, - ) - - res = await httpx_client.get( - url, - headers=self.headers, - timeout=self.timeout, - ) - - try: - if res.status_code == 401: - raise SGDBInvalidAPIKeyException - search_response = res.json() - except json.decoder.JSONDecodeError as exc: - log.error( - "Failed to decode JSON response from SteamGridDB: %s", - str(exc), - ) - return [] - - if len(search_response["data"]) == 0: + games = await self.sgdb_service.search_games(term=search_term) + if not games: log.warning(f"Could not find '{search_term}' on SteamGridDB") return [] tasks = [ self._get_game_covers(game_id=game["id"], game_name=game["name"]) - for game in search_response["data"] + for game in games ] results = await asyncio.gather(*tasks) @@ -82,48 +33,23 @@ class SGDBBaseHandler(MetadataHandler): async def _get_game_covers( self, game_id: int, game_name: str ) -> dict[str, Any] | None: - httpx_client = ctx_httpx_client.get() - game_covers = [] - - for page in itertools.count(start=0): - url = f"{self.grid_endpoint}/{game_id}" - params = { - "dimensions": f"{STEAMVERTICAL},{GALAXY342},{GALAXY660},{SQUARE512},{SQUARE1024}", - "types": f"{STATIC},{ANIMATED}", - "limit": SGDB_API_COVER_LIMIT, - "page": page, - } - - log.debug( - "API request: Method=GET, URL=%s, Headers=%s, Params=%s, Timeout=%s", - url, - self.masked_headers, - params, - self.timeout, + game_covers = [ + cover + async for cover in self.sgdb_service.iter_grids_for_game( + game_id=game_id, + dimensions=( + SGDBDimension.STEAM_VERTICAL, + SGDBDimension.GOG_GALAXY_TILE, + SGDBDimension.GOG_GALAXY_COVER, + SGDBDimension.SQUARE_512, + SGDBDimension.SQUARE_1024, + ), + types=( + SGDBType.STATIC, + SGDBType.ANIMATED, + ), ) - - res = await httpx_client.get( - url, - headers=self.headers, - timeout=self.timeout, - params=params, - ) - - try: - covers_response = res.json() - except json.decoder.JSONDecodeError as exc: - log.error( - "Failed to decode JSON response from SteamGridDB: %s", - str(exc), - ) - return None - - page_covers = covers_response["data"] - game_covers.extend(page_covers) - - if len(page_covers) < SGDB_API_COVER_LIMIT: - break - + ] if not game_covers: return None