feat: MTG database integration with Redis caching
- Added MTG card ORM models (mtg_cards, mtg_sets tables) - Created card_database service with search, get_by_name, get_by_set - Added Redis client with caching layer (3600s TTL default) - Created card router with caching on all endpoints: - Search cards (5min cache) - Get card by name (10min cache) - Get cards by set (15min cache) - Get card types/rarities (30min cache) - Get sets (1hr cache) - Get statistics (1hr cache) - Updated settings.py: - Added JWT_SECRET_KEY field - Added DB_CONFIG and REDIS_CONFIG dictionaries - Updated security.py to use JWT_SECRET_KEY with fallback - Updated auth.py to use timezone-aware datetimes - Updated refresh_mtg.py to use settings instead of os.environ - Updated mtg_monitor.py to use settings for connections - Added services package with __init__.py All 20 tests passing.
This commit is contained in:
@@ -0,0 +1,22 @@
|
||||
"""Services package."""
|
||||
from app.services.card_database import (
|
||||
search_cards,
|
||||
get_card_by_name,
|
||||
get_cards_by_set,
|
||||
get_card_types,
|
||||
get_card_rarities,
|
||||
get_sets,
|
||||
get_set_by_code,
|
||||
get_card_statistics,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"search_cards",
|
||||
"get_card_by_name",
|
||||
"get_cards_by_set",
|
||||
"get_card_types",
|
||||
"get_card_rarities",
|
||||
"get_sets",
|
||||
"get_set_by_code",
|
||||
"get_card_statistics",
|
||||
]
|
||||
@@ -1,225 +1,350 @@
|
||||
"""Card database service for importing and querying card data."""
|
||||
import httpx
|
||||
from typing import List, Dict, Optional
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
"""
|
||||
MTG Card Database Service.
|
||||
|
||||
Queries the MTG PostgreSQL database for card data.
|
||||
"""
|
||||
from typing import List, Dict, Any, Optional
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy import select, and_, or_, func
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from app.models.mtg_models import MtgCard, MtgSet
|
||||
from app.core.database import mtg_get_db
|
||||
|
||||
|
||||
class CardType(str, Enum):
|
||||
"""Card types from Magic: The Gathering."""
|
||||
CREATURES = "Creature"
|
||||
INSTANT = "Instant"
|
||||
SORCERY = "Sorcery"
|
||||
ENCHANTMENT = "Enchantment"
|
||||
ARTIFACT = "Artifact"
|
||||
PLANE = "Plane"
|
||||
PLANESWALKER = "Planeswalker"
|
||||
LAND = "Land"
|
||||
BATTLE = "Battle"
|
||||
|
||||
|
||||
class CardColor(str, Enum):
|
||||
"""Card colors."""
|
||||
WHITE = "W"
|
||||
BLUE = "U"
|
||||
BLACK = "B"
|
||||
RED = "R"
|
||||
GREEN = "G"
|
||||
COLORLESS = "C"
|
||||
MULTICOLOR = "M"
|
||||
SHARD = "S"
|
||||
WIDGET = "X"
|
||||
|
||||
|
||||
class CardRarity(str, Enum):
|
||||
"""Card rarities."""
|
||||
COMMON = "common"
|
||||
UNCOMMON = "uncommon"
|
||||
RARE = "rare"
|
||||
MYTHIC = "mythic"
|
||||
SPECIAL = "special"
|
||||
|
||||
|
||||
@dataclass
|
||||
class CardData:
|
||||
"""Card information from card database."""
|
||||
id: int
|
||||
name: str
|
||||
types: List[CardType]
|
||||
colors: List[CardColor]
|
||||
rarity: CardRarity
|
||||
set_code: str
|
||||
collector_number: str
|
||||
flavor_text: Optional[str] = None
|
||||
rules_text: Optional[str] = None
|
||||
power: Optional[str] = None
|
||||
toughness: Optional[str] = None
|
||||
artist: Optional[str] = None
|
||||
image_url: Optional[str] = None
|
||||
provider_id: Optional[str] = None
|
||||
async def search_cards(
|
||||
query: str,
|
||||
db: AsyncSession,
|
||||
limit: int = 100,
|
||||
offset: int = 0,
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Search cards by name, type, or mana cost.
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"{self.name} ({self.set_code}-{self.collector_number})"
|
||||
|
||||
|
||||
class CardDatabase:
|
||||
"""Card database service for importing and querying card data."""
|
||||
Args:
|
||||
query: Search string
|
||||
db: Database session
|
||||
limit: Maximum results to return
|
||||
offset: Number of results to skip
|
||||
|
||||
def __init__(self):
|
||||
self.cards: Dict[int, CardData] = {}
|
||||
self._next_id = 1
|
||||
Returns:
|
||||
Dictionary with results and total count
|
||||
"""
|
||||
search_term = f"%{query.lower()}%"
|
||||
|
||||
async def import_from_mtjson(self, url: str = "https://mtjson.xyz/api/5.0.0/") -> List[CardData]:
|
||||
"""Import card data from MTJSON API."""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(url)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
imported_cards = []
|
||||
for card_data in data:
|
||||
card = self._parse_mtjson_card(card_data)
|
||||
self.cards[self._next_id] = card
|
||||
imported_cards.append(card)
|
||||
self._next_id += 1
|
||||
|
||||
return imported_cards
|
||||
|
||||
def _parse_mtjson_card(self, data: dict) -> CardData:
|
||||
"""Parse MTJSON card data into CardData."""
|
||||
card_id = self._next_id
|
||||
|
||||
# Extract types
|
||||
types = []
|
||||
if "types" in data:
|
||||
for type_str in data["types"]:
|
||||
try:
|
||||
types.append(CardType(type_str))
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Extract colors
|
||||
colors = []
|
||||
if "colors" in data:
|
||||
for color_str in data["colors"]:
|
||||
try:
|
||||
colors.append(CardColor(color_str))
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Extract rarity
|
||||
rarity = CardRarity.COMMON
|
||||
if "rarity" in data:
|
||||
try:
|
||||
rarity = CardRarity(data["rarity"].lower())
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Extract set and collector number
|
||||
set_code = ""
|
||||
collector_number = ""
|
||||
if "set" in data:
|
||||
set_code = data["set"]
|
||||
if "collectorNumber" in data:
|
||||
collector_number = data["collectorNumber"]
|
||||
|
||||
# Extract image URL
|
||||
image_url = None
|
||||
if "imageUris" in data and "normal" in data["imageUris"]:
|
||||
image_url = data["imageUris"]["normal"]
|
||||
|
||||
return CardData(
|
||||
id=card_id,
|
||||
name=data.get("name", ""),
|
||||
types=types,
|
||||
colors=colors,
|
||||
rarity=rarity,
|
||||
set_code=set_code,
|
||||
collector_number=collector_number,
|
||||
flavor_text=data.get("flavorText"),
|
||||
rules_text=data.get("rulesText"),
|
||||
power=data.get("power"),
|
||||
toughness=data.get("toughness"),
|
||||
artist=data.get("artist"),
|
||||
image_url=image_url,
|
||||
provider_id=data.get("multiverseId"),
|
||||
# Search across multiple fields
|
||||
stmt = (
|
||||
select(MtgCard, MtgSet)
|
||||
.join(MtgSet, MtgCard.set_id == MtgSet.id, isouter=True)
|
||||
.where(
|
||||
or_(
|
||||
MtgCard.name.ilike(search_term),
|
||||
MtgCard.type_line.ilike(search_term),
|
||||
MtgCard.mana_cost.ilike(search_term),
|
||||
)
|
||||
)
|
||||
.offset(offset)
|
||||
.limit(limit)
|
||||
)
|
||||
|
||||
def get_card_by_id(self, card_id: int) -> Optional[CardData]:
|
||||
"""Get card by ID."""
|
||||
return self.cards.get(card_id)
|
||||
result = await db.execute(stmt)
|
||||
rows = result.all()
|
||||
|
||||
def get_card_by_name(self, name: str) -> List[CardData]:
|
||||
"""Get cards by name (case-insensitive)."""
|
||||
name_lower = name.lower()
|
||||
return [card for card in self.cards.values() if card.name.lower() == name_lower]
|
||||
cards = []
|
||||
for card, mtg_set in rows:
|
||||
card_data = {
|
||||
"id": card.id,
|
||||
"name": card.name,
|
||||
"mana_cost": card.mana_cost,
|
||||
"type_line": card.type_line,
|
||||
"oracle_text": card.oracle_text,
|
||||
"power": card.power,
|
||||
"toughness": card.toughness,
|
||||
"rarity": card.rarity,
|
||||
"layout": card.layout,
|
||||
"artist": card.artist,
|
||||
"flavor_text": card.flavor_text,
|
||||
"set_code": mtg_set.code if mtg_set else None,
|
||||
"set_name": mtg_set.name if mtg_set else None,
|
||||
"release_date": mtg_set.release_date.isoformat() if mtg_set and mtg_set.release_date else None,
|
||||
"identifiers": card.identifiers,
|
||||
"images": card.images,
|
||||
}
|
||||
cards.append(card_data)
|
||||
|
||||
def search_cards(
|
||||
self,
|
||||
query: str = "",
|
||||
card_type: Optional[CardType] = None,
|
||||
color: Optional[CardColor] = None,
|
||||
rarity: Optional[CardRarity] = None,
|
||||
set_code: Optional[str] = None,
|
||||
limit: int = 50,
|
||||
) -> List[CardData]:
|
||||
"""Search cards with filters."""
|
||||
results = list(self.cards.values())
|
||||
|
||||
# Filter by query
|
||||
if query:
|
||||
query_lower = query.lower()
|
||||
results = [card for card in results if query_lower in card.name.lower()]
|
||||
|
||||
# Filter by type
|
||||
if card_type:
|
||||
results = [card for card in results if card_type in card.types]
|
||||
|
||||
# Filter by color
|
||||
if color:
|
||||
results = [card for card in results if color in card.colors]
|
||||
|
||||
# Filter by rarity
|
||||
if rarity:
|
||||
results = [card for card in results if card.rarity == rarity]
|
||||
|
||||
# Filter by set
|
||||
if set_code:
|
||||
results = [card for card in results if card.set_code == set_code.upper()]
|
||||
|
||||
return results[:limit]
|
||||
# Get total count
|
||||
count_stmt = select(func.count()).select_from(MtgCard)
|
||||
count_result = await db.execute(count_stmt)
|
||||
total = count_result.scalar()
|
||||
|
||||
def get_random_card(self) -> Optional[CardData]:
|
||||
"""Get a random card from the database."""
|
||||
import random
|
||||
if not self.cards:
|
||||
return None
|
||||
return random.choice(list(self.cards.values()))
|
||||
return {
|
||||
"results": cards,
|
||||
"total": total,
|
||||
"limit": limit,
|
||||
"offset": offset,
|
||||
}
|
||||
|
||||
|
||||
async def get_card_by_name(
|
||||
name: str,
|
||||
db: AsyncSession,
|
||||
set_code: Optional[str] = None,
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get a specific card by name.
|
||||
|
||||
def get_card_count(self) -> int:
|
||||
"""Get total number of cards in database."""
|
||||
return len(self.cards)
|
||||
Args:
|
||||
name: Card name
|
||||
db: Database session
|
||||
set_code: Optional set code to filter by
|
||||
|
||||
Returns:
|
||||
Card data or None
|
||||
"""
|
||||
stmt = (
|
||||
select(MtgCard, MtgSet)
|
||||
.join(MtgSet, MtgCard.set_id == MtgSet.id, isouter=True)
|
||||
.where(MtgCard.name.ilike(name))
|
||||
)
|
||||
|
||||
if set_code:
|
||||
stmt = stmt.where(MtgSet.code == set_code)
|
||||
|
||||
stmt = stmt.limit(1)
|
||||
|
||||
result = await db.execute(stmt)
|
||||
row = result.fetchone()
|
||||
|
||||
if not row:
|
||||
return None
|
||||
|
||||
card, mtg_set = row
|
||||
|
||||
return {
|
||||
"id": card.id,
|
||||
"name": card.name,
|
||||
"mana_cost": card.mana_cost,
|
||||
"type_line": card.type_line,
|
||||
"oracle_text": card.oracle_text,
|
||||
"power": card.power,
|
||||
"toughness": card.toughness,
|
||||
"rarity": card.rarity,
|
||||
"layout": card.layout,
|
||||
"artist": card.artist,
|
||||
"flavor_text": card.flavor_text,
|
||||
"set_code": mtg_set.code if mtg_set else None,
|
||||
"set_name": mtg_set.name if mtg_set else None,
|
||||
"release_date": mtg_set.release_date.isoformat() if mtg_set and mtg_set.release_date else None,
|
||||
"identifiers": card.identifiers,
|
||||
"images": card.images,
|
||||
}
|
||||
|
||||
|
||||
# Singleton instance
|
||||
card_database = CardDatabase()
|
||||
async def get_cards_by_set(
|
||||
set_code: str,
|
||||
db: AsyncSession,
|
||||
limit: int = 1000,
|
||||
offset: int = 0,
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Get all cards in a specific set.
|
||||
|
||||
Args:
|
||||
set_code: Set code
|
||||
db: Database session
|
||||
limit: Maximum results to return
|
||||
offset: Number of results to skip
|
||||
|
||||
Returns:
|
||||
Dictionary with results and total count
|
||||
"""
|
||||
# First get the set
|
||||
set_stmt = select(MtgSet).where(MtgSet.code == set_code)
|
||||
set_result = await db.execute(set_stmt)
|
||||
mtg_set = set_result.scalar_one_or_none()
|
||||
|
||||
if not mtg_set:
|
||||
return {"results": [], "total": 0, "limit": limit, "offset": offset}
|
||||
|
||||
# Get cards in the set
|
||||
stmt = (
|
||||
select(MtgCard, MtgSet)
|
||||
.join(MtgSet, MtgCard.set_id == MtgSet.id, isouter=True)
|
||||
.where(MtgCard.set_id == mtg_set.id)
|
||||
.offset(offset)
|
||||
.limit(limit)
|
||||
)
|
||||
|
||||
result = await db.execute(stmt)
|
||||
rows = result.all()
|
||||
|
||||
cards = []
|
||||
for card, _ in rows:
|
||||
card_data = {
|
||||
"id": card.id,
|
||||
"name": card.name,
|
||||
"mana_cost": card.mana_cost,
|
||||
"type_line": card.type_line,
|
||||
"oracle_text": card.oracle_text,
|
||||
"power": card.power,
|
||||
"toughness": card.toughness,
|
||||
"rarity": card.rarity,
|
||||
"layout": card.layout,
|
||||
"artist": card.artist,
|
||||
"flavor_text": card.flavor_text,
|
||||
"set_code": mtg_set.code,
|
||||
"set_name": mtg_set.name,
|
||||
"release_date": mtg_set.release_date.isoformat() if mtg_set.release_date else None,
|
||||
"identifiers": card.identifiers,
|
||||
"images": card.images,
|
||||
}
|
||||
cards.append(card_data)
|
||||
|
||||
# Get total count
|
||||
count_stmt = select(func.count()).where(MtgCard.set_id == mtg_set.id)
|
||||
count_result = await db.execute(count_stmt)
|
||||
total = count_result.scalar()
|
||||
|
||||
return {
|
||||
"results": cards,
|
||||
"total": total,
|
||||
"limit": limit,
|
||||
"offset": offset,
|
||||
}
|
||||
|
||||
|
||||
async def import_cards() -> List[CardData]:
|
||||
"""Import cards from MTJSON."""
|
||||
return await card_database.import_from_mtjson()
|
||||
async def get_card_types(db: AsyncSession) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get all unique card types.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
List of card types
|
||||
"""
|
||||
stmt = select(MtgCard.type_line).distinct().order_by(MtgCard.type_line)
|
||||
result = await db.execute(stmt)
|
||||
rows = result.fetchall()
|
||||
|
||||
return [{"type": row[0]} for row in rows]
|
||||
|
||||
|
||||
def search_cards(**kwargs) -> List[CardData]:
|
||||
"""Search cards with filters."""
|
||||
return card_database.search_cards(**kwargs)
|
||||
async def get_card_rarities(db: AsyncSession) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get all unique card rarities.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
List of rarities
|
||||
"""
|
||||
stmt = select(MtgCard.rarity).distinct().order_by(MtgCard.rarity)
|
||||
result = await db.execute(stmt)
|
||||
rows = result.fetchall()
|
||||
|
||||
return [{"rarity": row[0]} for row in rows]
|
||||
|
||||
|
||||
def get_card_by_name(name: str) -> List[CardData]:
|
||||
"""Get cards by name."""
|
||||
return card_database.get_card_by_name(name)
|
||||
async def get_sets(db: AsyncSession) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get all sets.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
List of sets
|
||||
"""
|
||||
stmt = select(MtgSet).order_by(MtgSet.release_date.desc())
|
||||
result = await db.execute(stmt)
|
||||
rows = result.fetchall()
|
||||
|
||||
return [
|
||||
{
|
||||
"id": s.id,
|
||||
"code": s.code,
|
||||
"name": s.name,
|
||||
"release_date": s.release_date.isoformat() if s.release_date else None,
|
||||
"total_size": s.total_size,
|
||||
"base_set_size": s.base_set_size,
|
||||
}
|
||||
for s in rows
|
||||
]
|
||||
|
||||
|
||||
def get_card_by_id(card_id: int) -> Optional[CardData]:
|
||||
"""Get card by ID."""
|
||||
return card_database.get_card_by_id(card_id)
|
||||
async def get_set_by_code(code: str, db: AsyncSession) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get a specific set by code.
|
||||
|
||||
Args:
|
||||
code: Set code
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
Set data or None
|
||||
"""
|
||||
stmt = select(MtgSet).where(MtgSet.code == code)
|
||||
result = await db.execute(stmt)
|
||||
mtg_set = result.scalar_one_or_none()
|
||||
|
||||
if not mtg_set:
|
||||
return None
|
||||
|
||||
return {
|
||||
"id": mtg_set.id,
|
||||
"code": mtg_set.code,
|
||||
"name": mtg_set.name,
|
||||
"type": mtg_set.type,
|
||||
"release_date": mtg_set.release_date.isoformat() if mtg_set.release_date else None,
|
||||
"base_set_size": mtg_set.base_set_size,
|
||||
"total_size": mtg_set.total_size,
|
||||
"is_foil_only": mtg_set.is_foil_only,
|
||||
"is_non_foil_only": mtg_set.is_non_foil_only,
|
||||
"digital": mtg_set.digital,
|
||||
"icon_svg_url": mtg_set.icon_svg_url,
|
||||
"parent_code": mtg_set.parent_code,
|
||||
"mtgo_code": mtg_set.mtgo_code,
|
||||
}
|
||||
|
||||
|
||||
async def get_card_statistics(db: AsyncSession) -> Dict[str, Any]:
|
||||
"""
|
||||
Get overall card database statistics.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
Dictionary with statistics
|
||||
"""
|
||||
# Total cards
|
||||
card_count_stmt = select(func.count()).select_from(MtgCard)
|
||||
card_count = (await db.execute(card_count_stmt)).scalar()
|
||||
|
||||
# Total sets
|
||||
set_count_stmt = select(func.count()).select_from(MtgSet)
|
||||
set_count = (await db.execute(set_count_stmt)).scalar()
|
||||
|
||||
# Cards by rarity
|
||||
rarity_stmt = select(MtgCard.rarity, func.count()).group_by(MtgCard.rarity)
|
||||
rarity_result = await db.execute(rarity_stmt)
|
||||
rarities = {row[0]: row[1] for row in rarity_result}
|
||||
|
||||
# Cards by type
|
||||
type_stmt = select(MtgCard.type_line, func.count()).group_by(MtgCard.type_line)
|
||||
type_result = await db.execute(type_stmt)
|
||||
types = {row[0]: row[1] for row in type_result}
|
||||
|
||||
# Average mana cost (approximate)
|
||||
avg_mana_stmt = select(func.count()).where(MtgCard.mana_cost.isnot(None))
|
||||
avg_mana_count = (await db.execute(avg_mana_stmt)).scalar()
|
||||
|
||||
return {
|
||||
"total_cards": card_count,
|
||||
"total_sets": set_count,
|
||||
"cards_by_rarity": rarities,
|
||||
"cards_by_type": types,
|
||||
"cards_with_mana_cost": avg_mana_count,
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user