mirror of
https://github.com/rommapp/romm.git
synced 2026-02-18 23:42:07 +01:00
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.
This commit is contained in:
Michael Manganiello
175
backend/adapters/services/steamgriddb.py
Normal file
175
backend/adapters/services/steamgriddb.py
Normal file
@@ -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", []))
|
||||
75
backend/adapters/services/steamgriddb_types.py
Normal file
75
backend/adapters/services/steamgriddb_types.py
Normal file
@@ -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]
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user