import httpx
import asyncio
from typing import Dict, List, Optional, Any
from config import settings
import logging
logger = logging.getLogger(__name__)
class RAWGAPIError(Exception):
"""RAWG API specific error"""
pass
class RAWGClient:
"""Async client for RAWG API"""
def __init__(self):
self.base_url = settings.rawg_base_url
self.api_key = settings.rawg_api_key
self._client: Optional[httpx.AsyncClient] = None
async def get_client(self) -> httpx.AsyncClient:
"""Get or create HTTP client"""
if self._client is None:
self._client = httpx.AsyncClient(timeout=30.0)
return self._client
async def close(self):
"""Close the HTTP client"""
if self._client:
await self._client.aclose()
self._client = None
async def _make_request(
self, endpoint: str, params: Dict[str, Any] = None
) -> Dict[str, Any]:
"""Make an authenticated request to RAWG API"""
if not self.api_key:
raise RAWGAPIError(
"RAWG API key not configured. Please set RAWG_API_KEY in your .env file."
)
if params is None:
params = {}
params["key"] = self.api_key
url = f"{self.base_url}/{endpoint}"
client = await self.get_client()
try:
response = await client.get(url, params=params)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
logger.error(
f"RAWG API error: {e.response.status_code} - {e.response.text}"
)
raise RAWGAPIError(f"API request failed: {e.response.status_code}")
except httpx.RequestError as e:
logger.error(f"Request error: {e}")
raise RAWGAPIError(f"Request failed: {str(e)}")
async def search_games(self, query: str, page_size: int = 10) -> Dict[str, Any]:
"""Search for games by title"""
params = {
"search": query,
"page_size": min(page_size, 40), # RAWG API limit
"search_precise": True,
}
return await self._make_request("games", params)
async def get_game_details(self, game_id: int) -> Dict[str, Any]:
"""Get detailed information about a specific game"""
return await self._make_request(f"games/{game_id}")
async def list_genres(self) -> Dict[str, Any]:
"""Get list of all game genres"""
return await self._make_request("genres")
async def get_games_by_genre(
self, genre_id: int, page_size: int = 10
) -> Dict[str, Any]:
"""Get games filtered by genre"""
params = {
"genres": genre_id,
"page_size": min(page_size, 40),
"ordering": "-rating",
}
return await self._make_request("games", params)
async def get_popular_games(self, page_size: int = 10) -> Dict[str, Any]:
"""Get popular games"""
params = {"page_size": min(page_size, 40), "ordering": "-rating"}
return await self._make_request("games", params)
# Global client instance
_rawg_client = RAWGClient()
async def get_rawg_client() -> RAWGClient:
"""Get the global RAWG client instance"""
return _rawg_client
async def cleanup_rawg_client():
"""Cleanup the global RAWG client"""
await _rawg_client.close()
