# =============================
# src/datastore.py
# =============================
"""
Pokemon data store implementation with PokeAPI integration and local caching.
Handles fetching, caching, and structuring Pokemon data from the PokeAPI.
"""
from __future__ import annotations
import asyncio
from typing import Any, Dict, List, Optional
import httpx
from types import SimpleNamespace
from .pokemon_types import Ability, Move, PokemonRecord, StatBlock
from .utils import JsonCache, CACHE_PATH, logger
class PokemonDataStore:
"""
Main data store class for Pokemon data management.
Provides caching layer over PokeAPI to reduce API calls and improve performance.
"""
def __init__(self) -> None:
"""
Initialize the datastore with a JSON cache instance.
The cache persists Pokemon data between server restarts.
"""
self.cache = JsonCache(CACHE_PATH)
async def list_pokemon(self, offset: int = 0, limit: int = 20) -> List[Dict[str, Any]]:
"""
Get a paginated list of Pokemon with basic information including types.
This method fetches a list from PokeAPI, then enriches each entry with
type information by calling get_pokemon() for detailed data.
Args:
offset: Starting index for pagination (default 0)
limit: Maximum number of Pokemon to return (default 20)
Returns:
List[Dict[str, Any]]: List of Pokemon data with id, name, and types
"""
# Fetch paginated list from PokeAPI
async with httpx.AsyncClient(timeout=30.0) as client:
r = await client.get(f"https://pokeapi.co/api/v2/pokemon?offset={offset}&limit={limit}")
r.raise_for_status()
data = r.json()
results = []
# Enrich each entry with detailed information (especially types)
for entry in data.get("results", []):
name = entry["name"]
mon = await self.get_pokemon(name) # This will use cache if available
results.append({
"id": mon.id,
"name": mon.display_name.lower(),
"types": mon.types
})
return results
async def get_pokemon(self, name_or_id: str | int) -> PokemonRecord:
"""
Get detailed Pokemon information by name or ID.
First checks the local cache, then fetches from PokeAPI if not cached.
Processes raw API data into structured PokemonRecord format.
Args:
name_or_id: Pokemon name (string) or Pokedex ID (integer)
Returns:
PokemonRecord: Complete Pokemon data including stats, moves, abilities, etc.
"""
key = str(name_or_id).lower()
# Check cache first
if key in self.cache.data["pokemon"]:
return self._from_cache_record(self.cache.data["pokemon"][key])
# Fetch from PokeAPI if not cached
async with httpx.AsyncClient(timeout=30.0) as client:
r = await client.get(f"https://pokeapi.co/api/v2/pokemon/{key}")
r.raise_for_status()
p = r.json()
# Extract basic information
display = p["name"]
pid: int = p["id"]
types = [p_t["type"]["name"].capitalize() for p_t in p["types"]]
# Process base stats into StatBlock
stats_map = {s["stat"]["name"]: s["base_stat"] for s in p["stats"]}
base = StatBlock(
hp=stats_map.get("hp", 1),
atk=stats_map.get("attack", 1),
def_=stats_map.get("defense", 1),
spa=stats_map.get("special-attack", 1),
spd=stats_map.get("special-defense", 1),
spe=stats_map.get("speed", 1),
)
# Process abilities
abilities = [Ability(name=a["ability"]["name"].capitalize()) for a in p["abilities"]]
# Fetch move details (limit to first 40 to reduce API calls)
move_names = [m["move"]["name"] for m in p["moves"]][:40]
moves: List[Move] = await self._fetch_moves(move_names)
# Process evolution chain
evo_from: Optional[str] = None
evo_to: List[str] = []
try:
async with httpx.AsyncClient(timeout=30.0) as client:
# Get species data for evolution chain URL
species = (await client.get(p["species"]["url"]))
species.raise_for_status()
s = species.json()
# Get what this Pokemon evolves from
evo_from = s.get("evolves_from_species", {}).get("name")
if evo_from:
evo_from = evo_from.capitalize()
# Get evolution chain data
evo_url = s.get("evolution_chain", {}).get("url")
if evo_url:
chain = (await client.get(evo_url)).json()
self._collect_evolutions(chain.get("chain"), display, evo_to)
except Exception:
# Ignore evolution chain errors - not critical data
pass
# Structure data for caching
record = {
"id": pid,
"name": display.lower(),
"display_name": display.capitalize(),
"types": types,
"base_stats": base.as_dict(),
"abilities": [{"name": ab.name, "effect": ab.effect} for ab in abilities],
"moves": [m.__dict__ for m in moves],
"evolution_from": evo_from,
"evolution_to": evo_to,
}
# Cache under both name and ID for flexible lookup
self.cache.data["pokemon"][str(pid)] = record
self.cache.data["pokemon"][display.lower()] = record
self.cache.save()
return self._from_cache_record(record)
async def _fetch_moves(self, names: List[str]) -> List[Move]:
"""
Fetch detailed move information from PokeAPI.
Makes concurrent requests for multiple moves to improve performance.
Filters out failed requests and returns only successful move data.
Args:
names: List of move names to fetch from PokeAPI
Returns:
List[Move]: List of successfully fetched Move objects
"""
out: List[Move] = []
async with httpx.AsyncClient(timeout=30.0) as client:
async def fetch_one(n: str) -> Optional[Move]:
"""
Fetch a single move's data from PokeAPI.
Args:
n: Move name to fetch
Returns:
Optional[Move]: Move object if successful, None if failed
"""
try:
r = await client.get(f"https://pokeapi.co/api/v2/move/{n}")
if r.status_code != 200:
return None
m = r.json()
# Determine move category from damage class
dmg = m.get("damage_class", {}).get("name")
cat: str
if dmg == "physical":
cat = "physical"
elif dmg == "special":
cat = "special"
else:
cat = "status"
return Move(
name=m["name"].capitalize(),
type=m.get("type", {}).get("name", "").capitalize(),
category=cat, # type: ignore
power=m.get("power"),
accuracy=m.get("accuracy"),
pp=m.get("pp"),
effect=self._move_effect(m),
)
except Exception:
return None
# Fetch all moves concurrently
results = await asyncio.gather(*(fetch_one(n) for n in names))
# Filter out None results
for mv in results:
if mv:
out.append(mv)
return out
def _move_effect(self, m: Dict[str, Any]) -> Optional[str]:
"""
Extract move effect description from PokeAPI move data.
Searches for English effect text, preferring short_effect over full effect.
Args:
m: Raw move data from PokeAPI
Returns:
Optional[str]: Move effect description in English, or None if not found
"""
entries = m.get("effect_entries", [])
# Find English language entry
en = next((e for e in entries if e.get("language", {}).get("name") == "en"), None)
if en:
# Prefer short effect, fallback to full effect
return en.get("short_effect") or en.get("effect")
return None
def _collect_evolutions(self, node: Dict[str, Any], current: str, acc: List[str]) -> None:
"""
Recursively collect evolution targets from evolution chain data.
Traverses the evolution tree and collects all Pokemon that the current
Pokemon can evolve into (directly or indirectly).
Args:
node: Current node in the evolution chain tree
current: Name of the current Pokemon we're finding evolutions for
acc: Accumulator list to collect evolution names
"""
try:
# Process all evolution branches from this node
for evo in node.get("evolves_to", []) or []:
name = evo.get("species", {}).get("name", "").capitalize()
# Avoid self-references
if name.lower() != current.lower():
acc.append(name)
# Recursively process further evolutions
self._collect_evolutions(evo, current, acc)
except Exception:
# Ignore errors in evolution chain processing
pass
def _from_cache_record(self, rec: Dict[str, Any]) -> PokemonRecord:
"""
Convert cached dictionary data back into a PokemonRecord object.
Reconstructs the full object hierarchy from flattened cache data.
Args:
rec: Dictionary containing cached Pokemon data
Returns:
PokemonRecord: Fully reconstructed Pokemon object
"""
base = rec["base_stats"]
return PokemonRecord(
id=rec["id"],
name=rec["name"],
display_name=rec["display_name"],
types=list(rec["types"]),
base_stats=StatBlock(
hp=base["hp"],
atk=base["atk"],
def_=base["def"],
spa=base["spa"],
spd=base["spd"],
spe=base["spe"]
),
abilities=[Ability(**a) for a in rec["abilities"]],
moves=[Move(**m) for m in rec["moves"]],
evolution_from=rec.get("evolution_from"),
evolution_to=rec.get("evolution_to"),
)
