akadmin
915242b330
- 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.
351 lines
9.4 KiB
Python
351 lines
9.4 KiB
Python
"""
|
|
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
|
|
|
|
|
|
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.
|
|
|
|
Args:
|
|
query: Search string
|
|
db: Database session
|
|
limit: Maximum results to return
|
|
offset: Number of results to skip
|
|
|
|
Returns:
|
|
Dictionary with results and total count
|
|
"""
|
|
search_term = f"%{query.lower()}%"
|
|
|
|
# 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)
|
|
)
|
|
|
|
result = await db.execute(stmt)
|
|
rows = result.all()
|
|
|
|
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)
|
|
|
|
# Get total count
|
|
count_stmt = select(func.count()).select_from(MtgCard)
|
|
count_result = await db.execute(count_stmt)
|
|
total = count_result.scalar()
|
|
|
|
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.
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
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 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]
|
|
|
|
|
|
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]
|
|
|
|
|
|
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
|
|
]
|
|
|
|
|
|
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,
|
|
}
|