Files
akadmin 900f51af3a Complete architecture simplification: 54+ files → 37 files
- Removed 11 duplicate router files (auth.py, users.py, decks.py, card_router.py, interactions.py, user_data.py, rooms.py, games/router.py, refresh.py, card_import.py, ws.py)
- Removed duplicate core files (logging_config.py, settings.py)
- Removed unused services (card_mirror_service.py, mtgjson_loader.py, mtgjson_uploader.py)
- Removed unused schemas (proto_messages.py, protocol_constants.py)
- Created run.py startup script for simple execution
- Created app/core/config.py for unified configuration management
- Created app/core/logging.py for logging configuration
- Consolidated routers into 6 focused files (auth_router.py, users_router.py, decks_router.py, cards_router.py, user_data_router.py, games_router.py)
- Updated all imports to use consolidated model paths
- Added backward compatibility aliases in database.py
- Created SIMPLIFICATION_SUMMARY.md documentation

Architecture improvements:
- Single unified database engine with auto-detection (SQLite/PostgreSQL)
- Consolidated models into 2 files (models.py, mtg_models.py)
- Clean router structure with one file per domain
- Simplified configuration with sensible defaults
- SQLite by default for development, PostgreSQL for production

Testing:
- All imports verified working
- App loads successfully
- Ready for development and testing
2026-08-26 15:15:36 +00:00

2238 lines
67 KiB
Python

"""
User Data and Card Import Router.
Consolidated router combining user personal data management and card import
functionality. Provides endpoints for:
**User Data** (prefix: /api/v1/user-data):
- Session management (get, cleanup, logout)
- Deck versions (CRUD)
- Game replays and replay players (CRUD)
- Game outcomes (create, list)
- User statistics (get, update)
- Card collection (CRUD)
- Wishlist (CRUD)
- User groups with members and chat messages (CRUD)
- User networks with members (CRUD)
- User preferences (get, update)
- Activity log (list)
**Card Imports** (prefix: /api/v1/card-import):
- Upload import files (XLSX, CSV, JSON, ODS)
- Import batch status and results
- Confirm imports to card collection
- List/delete user's imported cards
- List user's decks (backward compatibility)
Both sub-routers share the UserCardCollection table — user_data.py handles
full CRUD while card_import.py manages batch imports.
"""
import json
from pathlib import Path
from typing import Any, Dict, List, Optional
from datetime import datetime, timezone
import tempfile
from fastapi import APIRouter, Depends, HTTPException, Query, status, UploadFile, File
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, update, delete
from sqlalchemy.orm import selectinload
from app.core.database import get_db
from app.core.security import get_current_user
from app.models.models import (
User, MtgonlineCard,
UserSession, DeckVersion, GameReplay, ReplayPlayer,
GameOutcome, UserStatistics, UserCardCollection, CardWishlist,
UserGroup, GroupMember, GroupChatMessage, UserNetwork,
NetworkMember, UserPreference, UserActivityLog,
CardImportBatch, UserCardImportRecord,
UserDeck, UserDeckCard,
)
from app.services.file_parser import FileParser
from app.services.import_batch_processor import ImportBatchProcessor
from app.services.fuzzy_card_matcher import FuzzyCardMatcher
from app.schemas.user_card_collection import (
CardCollectionCreate,
CardCollectionUpdate,
CardCollectionResponse,
CardCollectionListResponse,
WishlistCreate,
WishlistUpdate,
WishlistResponse,
WishlistListResponse,
)
from app.schemas.generic_schemas import MessageResponse, CountResponse, ErrorDetail
from app.schemas.user_data_schemas import (
SessionCleanupResponse,
DeckVersionCreate, DeckVersionUpdate, DeckVersionResponse, DeckVersionListResponse,
GameReplayCreate, GameReplayUpdate, GameReplayResponse, GameReplayListResponse,
GameOutcomeCreate, GameOutcomeResponse, GameOutcomeListResponse,
UserStatisticsResponse, StatisticsUpdateResponse,
GroupCreate, GroupUpdate, GroupResponse, GroupListResponse,
GroupMemberCreate, GroupMemberUpdate, GroupMemberRemove,
GroupChatMessageCreate, GroupChatMessageResponse, GroupChatMessageListResponse,
NetworkCreate, NetworkUpdate, NetworkResponse, NetworkListResponse,
NetworkMemberCreate,
UserPreferenceUpdate, UserPreferenceResponse,
ActivityLogEntry, ActivityLogListResponse,
)
from app.schemas.card_import_schemas import (
CardImportRequest,
CardImportResponse,
CardImportStatusResponse,
CardMatchResult,
CardImportSummary,
)
from app.schemas.user_deck_schemas import DeckCardResponse, DeckCardListResponse
# =============================================================================
# User Data Router (prefix: /api/v1/user-data)
# =============================================================================
user_data_router = APIRouter()
# ===== Session Management =====
@user_data_router.get("/sessions/me", response_model=List[SessionCleanupResponse])
async def get_current_sessions(
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> List[SessionCleanupResponse]:
"""Get active sessions for current user."""
user_id: int = int(current_user["user_id"])
stmt = (
select(UserSession)
.where(
UserSession.user_id == user_id,
UserSession.is_active == True, # noqa: E712
UserSession.expires_at > datetime.now(timezone.utc)
)
)
result = await db.execute(stmt)
sessions = result.scalars().all()
return [
SessionCleanupResponse(
cleaned_count=len(sessions),
message=f"Found {len(sessions)} active sessions"
)
]
@user_data_router.delete("/sessions/cleanup", response_model=MessageResponse)
async def cleanup_expired_sessions(
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> MessageResponse:
"""Cleanup expired sessions for current user."""
user_id: int = int(current_user["user_id"])
now = datetime.now(timezone.utc)
stmt = (
update(UserSession)
.where(
UserSession.user_id == user_id,
UserSession.is_active == True, # noqa: E712
UserSession.expires_at < now
)
.values(is_active=False)
)
await db.execute(stmt)
await db.flush()
# Count cleaned sessions
count_stmt = select(func.count()).select_from(UserSession).where(
UserSession.user_id == user_id,
UserSession.is_active == False, # noqa: E712
UserSession.expires_at < now
)
count_result = await db.execute(count_stmt)
cleaned_count = count_result.scalar()
return MessageResponse(message=f"Cleaned {cleaned_count} expired sessions")
@user_data_router.post("/sessions/logout", response_model=MessageResponse)
async def logout_current_session(
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> MessageResponse:
"""Logout current user session."""
user_id: int = int(current_user["user_id"])
stmt = (
update(UserSession)
.where(
UserSession.user_id == user_id,
UserSession.is_active == True # noqa: E712
)
.values(is_active=False)
)
await db.execute(stmt)
await db.flush()
return MessageResponse(message="Logged out successfully")
# ===== Deck Versions =====
@user_data_router.post(
"/decks/{deck_id}/versions",
response_model=DeckVersionResponse,
status_code=status.HTTP_201_CREATED
)
async def create_deck_version(
deck_id: int,
request: DeckVersionCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> DeckVersionResponse:
"""Create a new version of a deck."""
user_id: int = int(current_user["user_id"])
# Verify user exists
deck_check = select(User).where(User.id == user_id)
deck_result = await db.execute(deck_check)
user = deck_result.scalar_one_or_none()
if not user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
# Get current version number
version_stmt = (
select(func.max(DeckVersion.version_number))
.where(DeckVersion.deck_id == deck_id)
)
version_result = await db.execute(version_stmt)
max_version = version_result.scalar() or 0
new_version = DeckVersion(
deck_id=deck_id,
version_number=max_version + 1,
content=request.content,
status=request.status.value,
comment=request.comment,
created_at=datetime.now(timezone.utc)
)
db.add(new_version)
await db.flush()
return DeckVersionResponse.model_validate(new_version)
@user_data_router.get(
"/decks/{deck_id}/versions",
response_model=DeckVersionListResponse
)
async def get_deck_versions(
deck_id: int,
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=100),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> DeckVersionListResponse:
"""Get all versions of a deck."""
offset = (page - 1) * page_size
count_stmt = select(func.count()).select_from(DeckVersion).where(DeckVersion.deck_id == deck_id)
total_result = await db.execute(count_stmt)
total = total_result.scalar()
stmt = (
select(DeckVersion)
.where(DeckVersion.deck_id == deck_id)
.order_by(DeckVersion.version_number.desc())
.offset(offset)
.limit(page_size)
)
result = await db.execute(stmt)
versions = result.scalars().all()
return DeckVersionListResponse(
versions=[DeckVersionResponse.model_validate(v) for v in versions],
total=total,
page=page,
page_size=page_size,
total_pages=(total + page_size - 1) // page_size
)
@user_data_router.patch(
"/decks/{deck_id}/versions/{version_id}",
response_model=DeckVersionResponse
)
async def update_deck_version(
deck_id: int,
version_id: int,
request: DeckVersionUpdate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> DeckVersionResponse:
"""Update a deck version."""
stmt = select(DeckVersion).where(DeckVersion.id == version_id)
result = await db.execute(stmt)
version = result.scalar_one_or_none()
if not version:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Version not found")
update_data = request.model_dump(exclude_unset=True)
if "status" in update_data:
update_data["status"] = update_data["status"].value
stmt = (
update(DeckVersion)
.where(DeckVersion.id == version_id)
.values(**update_data)
)
await db.execute(stmt)
await db.flush()
stmt = select(DeckVersion).where(DeckVersion.id == version_id)
result = await db.execute(stmt)
updated_version = result.scalar_one_or_none()
return DeckVersionResponse.model_validate(updated_version)
@user_data_router.delete(
"/decks/{deck_id}/versions/{version_id}",
response_model=MessageResponse
)
async def delete_deck_version(
deck_id: int,
version_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> MessageResponse:
"""Delete a deck version."""
stmt = select(DeckVersion).where(DeckVersion.id == version_id)
result = await db.execute(stmt)
version = result.scalar_one_or_none()
if not version:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Version not found")
stmt = delete(DeckVersion).where(DeckVersion.id == version_id)
await db.execute(stmt)
await db.flush()
return MessageResponse(message="Version deleted successfully")
# ===== Game Replays =====
@user_data_router.post(
"/replays",
response_model=GameReplayResponse,
status_code=status.HTTP_201_CREATED
)
async def create_game_replay(
request: GameReplayCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GameReplayResponse:
"""Create a new game replay."""
# Check if replay already exists
stmt = select(GameReplay).where(GameReplay.game_uuid == request.game_uuid)
result = await db.execute(stmt)
existing = result.scalar_one_or_none()
if existing:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Replay already exists")
replay = GameReplay(
game_uuid=request.game_uuid,
room_id=request.room_id,
game_type=request.game_type,
format=request.format,
duration_seconds=request.duration_seconds,
start_time=request.start_time,
end_time=request.end_time,
status=request.status.value,
replay_data=request.replay_data,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc)
)
db.add(replay)
await db.flush()
return GameReplayResponse.model_validate(replay)
@user_data_router.get("/replays", response_model=GameReplayListResponse)
async def get_game_replays(
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=100),
user_id: Optional[int] = None,
status_filter: Optional[str] = None,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GameReplayListResponse:
"""Get game replays with optional filtering."""
offset = (page - 1) * page_size
# Build WHERE clause
conditions = []
if user_id:
conditions.append(ReplayPlayer.user_id == user_id)
if status_filter:
conditions.append(GameReplay.status == status_filter)
# Get total count
count_stmt = select(func.count()).select_from(GameReplay)
if conditions:
count_stmt = count_stmt.join(ReplayPlayer).where(*conditions)
total_result = await db.execute(count_stmt)
total = total_result.scalar()
# Get replays
stmt = select(GameReplay)
if conditions:
stmt = stmt.join(ReplayPlayer).where(*conditions)
stmt = stmt.order_by(GameReplay.start_time.desc()).offset(offset).limit(page_size)
result = await db.execute(stmt)
replays = result.scalars().all()
return GameReplayListResponse(
replays=[GameReplayResponse.model_validate(r) for r in replays],
total=total,
page=page,
page_size=page_size,
total_pages=(total + page_size - 1) // page_size
)
@user_data_router.get(
"/replays/{replay_id}",
response_model=GameReplayResponse
)
async def get_game_replay(
replay_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GameReplayResponse:
"""Get a specific game replay."""
stmt = select(GameReplay).where(GameReplay.id == replay_id)
result = await db.execute(stmt)
replay = result.scalar_one_or_none()
if not replay:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Replay not found")
return GameReplayResponse.model_validate(replay)
@user_data_router.patch(
"/replays/{replay_id}",
response_model=GameReplayResponse
)
async def update_game_replay(
replay_id: int,
request: GameReplayUpdate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GameReplayResponse:
"""Update a game replay."""
stmt = select(GameReplay).where(GameReplay.id == replay_id)
result = await db.execute(stmt)
replay = result.scalar_one_or_none()
if not replay:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Replay not found")
update_data = request.model_dump(exclude_unset=True)
if "status" in update_data:
update_data["status"] = update_data["status"].value
update_data["updated_at"] = datetime.now(timezone.utc)
stmt = (
update(GameReplay)
.where(GameReplay.id == replay_id)
.values(**update_data)
)
await db.execute(stmt)
await db.flush()
stmt = select(GameReplay).where(GameReplay.id == replay_id)
result = await db.execute(stmt)
updated_replay = result.scalar_one_or_none()
return GameReplayResponse.model_validate(updated_replay)
@user_data_router.delete(
"/replays/{replay_id}",
response_model=MessageResponse
)
async def delete_game_replay(
replay_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> MessageResponse:
"""Delete a game replay."""
stmt = select(GameReplay).where(GameReplay.id == replay_id)
result = await db.execute(stmt)
replay = result.scalar_one_or_none()
if not replay:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Replay not found")
stmt = delete(GameReplay).where(GameReplay.id == replay_id)
await db.execute(stmt)
await db.flush()
return MessageResponse(message="Replay deleted successfully")
# ===== Replay Players =====
@user_data_router.post(
"/replays/{replay_id}/players",
status_code=status.HTTP_201_CREATED
)
async def add_replay_player(
replay_id: int,
user_id: int,
deck_id: Optional[int] = None,
position: Optional[int] = None,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> Dict[str, Any]:
"""Add a player to a game replay."""
# Verify replay exists
stmt = select(GameReplay).where(GameReplay.id == replay_id)
result = await db.execute(stmt)
replay = result.scalar_one_or_none()
if not replay:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Replay not found")
# Check if player already exists
player_check = (
select(ReplayPlayer)
.where(ReplayPlayer.replay_id == replay_id, ReplayPlayer.user_id == user_id)
)
player_result = await db.execute(player_check)
if player_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Player already in replay")
player = ReplayPlayer(
replay_id=replay_id,
user_id=user_id,
deck_id=deck_id,
position=position,
created_at=datetime.now(timezone.utc)
)
db.add(player)
await db.flush()
return {"message": "Player added to replay", "player_id": player.id}
@user_data_router.get(
"/replays/{replay_id}/players",
response_model=List[dict]
)
async def get_replay_players(
replay_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> List[dict]:
"""Get all players in a game replay."""
stmt = select(ReplayPlayer).where(ReplayPlayer.replay_id == replay_id)
result = await db.execute(stmt)
players = result.scalars().all()
return [
{
"id": p.id,
"user_id": p.user_id,
"deck_id": p.deck_id,
"position": p.position,
"won": p.won,
"lost": p.lost,
"concession": p.concession,
"turn_one": p.turn_one,
}
for p in players
]
# ===== Game Outcomes =====
@user_data_router.post(
"/outcomes",
response_model=GameOutcomeResponse,
status_code=status.HTTP_201_CREATED
)
async def create_game_outcome(
request: GameOutcomeCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GameOutcomeResponse:
"""Create a game outcome record."""
outcome = GameOutcome(
user_id=int(current_user["user_id"]),
game_uuid=request.game_uuid,
outcome=request.outcome.value,
opponent_id=request.opponent_id,
format=request.format,
rating_before=request.rating_before,
rating_after=request.rating_after,
rating_change=request.rating_change,
created_at=datetime.now(timezone.utc)
)
db.add(outcome)
await db.flush()
return GameOutcomeResponse.model_validate(outcome)
@user_data_router.get("/outcomes", response_model=GameOutcomeListResponse)
async def get_game_outcomes(
user_id: Optional[int] = None,
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=100),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GameOutcomeListResponse:
"""Get game outcomes with optional filtering."""
offset = (page - 1) * page_size
stmt = select(GameOutcome)
if user_id:
stmt = stmt.where(GameOutcome.user_id == user_id)
stmt = stmt.order_by(GameOutcome.created_at.desc()).offset(offset).limit(page_size)
result = await db.execute(stmt)
outcomes = result.scalars().all()
count_stmt = select(func.count()).select_from(GameOutcome)
if user_id:
count_stmt = count_stmt.where(GameOutcome.user_id == user_id)
total_result = await db.execute(count_stmt)
total = total_result.scalar()
return GameOutcomeListResponse(
outcomes=[GameOutcomeResponse.model_validate(o) for o in outcomes],
total=total,
page=page,
page_size=page_size,
total_pages=(total + page_size - 1) // page_size
)
# ===== User Statistics =====
@user_data_router.get(
"/statistics/{user_id}",
response_model=UserStatisticsResponse
)
async def get_user_statistics(
user_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> UserStatisticsResponse:
"""Get user game statistics."""
stmt = select(UserStatistics).where(UserStatistics.user_id == user_id)
result = await db.execute(stmt)
stats = result.scalar_one_or_none()
if not stats:
# Create default statistics
stats = UserStatistics(
user_id=user_id,
total_games=0,
total_wins=0,
total_losses=0,
total_concessions=0,
win_rate=0.0,
current_streak=0,
best_streak=0,
average_rating=0.0,
updated_at=datetime.now(timezone.utc)
)
db.add(stats)
await db.flush()
return UserStatisticsResponse.model_validate(stats)
@user_data_router.post(
"/statistics/update",
response_model=StatisticsUpdateResponse
)
async def update_user_statistics(
user_id: int,
outcome: str, # 'WIN', 'LOSS', 'CONCESSION'
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> StatisticsUpdateResponse:
"""Update user statistics after a game."""
stmt = select(UserStatistics).where(UserStatistics.user_id == user_id)
result = await db.execute(stmt)
stats = result.scalar_one_or_none()
if not stats:
stats = UserStatistics(
user_id=user_id,
total_games=0,
total_wins=0,
total_losses=0,
total_concessions=0,
win_rate=0.0,
current_streak=0,
best_streak=0,
average_rating=0.0,
updated_at=datetime.now(timezone.utc)
)
db.add(stats)
# Update statistics
stats.total_games += 1
stats.last_game_date = datetime.now(timezone.utc)
if outcome == "WIN":
stats.total_wins += 1
stats.current_streak = max(0, stats.current_streak) + 1
if stats.current_streak > stats.best_streak:
stats.best_streak = stats.current_streak
elif outcome in ("LOSS", "CONCESSION"):
stats.total_losses += 1
stats.total_concessions += 1 if outcome == "CONCESSION" else 0
stats.current_streak = 0
# Calculate win rate
if stats.total_games > 0:
stats.win_rate = (stats.total_wins / stats.total_games) * 100
stats.updated_at = datetime.now(timezone.utc)
await db.flush()
return StatisticsUpdateResponse(
user_id=stats.user_id,
total_games=stats.total_games,
total_wins=stats.total_wins,
total_losses=stats.total_losses,
win_rate=stats.win_rate,
current_streak=stats.current_streak,
updated_at=stats.updated_at
)
# ===== Card Collection =====
@user_data_router.post(
"/collection",
response_model=CardCollectionResponse,
status_code=status.HTTP_201_CREATED
)
async def add_card_to_collection(
request: CardCollectionCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> CardCollectionResponse:
"""Add a card to user collection."""
user_id: int = int(current_user["user_id"])
# Check if card already exists in collection
check = (
select(UserCardCollection)
.where(
UserCardCollection.user_id == user_id,
UserCardCollection.card_id == request.card_id,
UserCardCollection.is_foil == request.is_foil,
UserCardCollection.is_alt_art == request.is_alt_art,
)
)
check_result = await db.execute(check)
existing = check_result.scalar_one_or_none()
if existing:
# Update quantity
stmt = (
update(UserCardCollection)
.where(UserCardCollection.id == existing.id)
.values(quantity=existing.quantity + request.quantity)
)
await db.execute(stmt)
await db.flush()
stmt = select(UserCardCollection).where(UserCardCollection.id == existing.id)
result = await db.execute(stmt)
updated = result.scalar_one_or_none()
return CardCollectionResponse.model_validate(updated)
card = UserCardCollection(
user_id=user_id,
card_id=request.card_id,
quantity=request.quantity,
condition=request.condition,
language=request.language,
is_foil=request.is_foil,
is_alt_art=request.is_alt_art,
acquired_date=request.acquired_date or datetime.now(timezone.utc),
acquisition_method=request.acquisition_method,
notes=request.notes,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc)
)
db.add(card)
await db.flush()
return CardCollectionResponse.model_validate(card)
@user_data_router.get("/collection", response_model=CardCollectionListResponse)
async def get_user_collection(
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=100),
is_foil: Optional[bool] = None,
is_alt_art: Optional[bool] = None,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> CardCollectionListResponse:
"""Get user's card collection."""
user_id: int = int(current_user["user_id"])
offset = (page - 1) * page_size
conditions = [UserCardCollection.user_id == user_id]
if is_foil is not None:
conditions.append(UserCardCollection.is_foil == is_foil)
if is_alt_art is not None:
conditions.append(UserCardCollection.is_alt_art == is_alt_art)
count_stmt = select(func.count()).select_from(UserCardCollection)
for cond in conditions:
count_stmt = count_stmt.where(cond)
total_result = await db.execute(count_stmt)
total = total_result.scalar()
stmt = select(UserCardCollection)
for cond in conditions:
stmt = stmt.where(cond)
stmt = stmt.order_by(UserCardCollection.acquired_date.desc()).offset(offset).limit(page_size)
result = await db.execute(stmt)
cards = result.scalars().all()
return CardCollectionListResponse(
cards=[CardCollectionResponse.model_validate(c) for c in cards],
total=total,
page=page,
page_size=page_size,
total_pages=(total + page_size - 1) // page_size
)
@user_data_router.patch(
"/collection/{card_id}",
response_model=CardCollectionResponse
)
async def update_card_in_collection(
card_id: int,
request: CardCollectionUpdate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> CardCollectionResponse:
"""Update a card in user collection."""
user_id: int = int(current_user["user_id"])
stmt = select(UserCardCollection).where(
UserCardCollection.id == card_id,
UserCardCollection.user_id == user_id
)
result = await db.execute(stmt)
card = result.scalar_one_or_none()
if not card:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Card not found in collection")
update_data = request.model_dump(exclude_unset=True)
update_data["updated_at"] = datetime.now(timezone.utc)
stmt = (
update(UserCardCollection)
.where(UserCardCollection.id == card_id)
.values(**update_data)
)
await db.execute(stmt)
await db.flush()
stmt = select(UserCardCollection).where(UserCardCollection.id == card_id)
result = await db.execute(stmt)
updated = result.scalar_one_or_none()
return CardCollectionResponse.model_validate(updated)
@user_data_router.delete(
"/collection/{card_id}",
response_model=MessageResponse
)
async def remove_card_from_collection(
card_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> MessageResponse:
"""Remove a card from user collection."""
user_id: int = int(current_user["user_id"])
stmt = select(UserCardCollection).where(
UserCardCollection.id == card_id,
UserCardCollection.user_id == user_id
)
result = await db.execute(stmt)
card = result.scalar_one_or_none()
if not card:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Card not found in collection")
stmt = delete(UserCardCollection).where(UserCardCollection.id == card_id)
await db.execute(stmt)
await db.flush()
return MessageResponse(message="Card removed from collection")
# ===== Wishlist =====
@user_data_router.post(
"/wishlist",
response_model=WishlistResponse,
status_code=status.HTTP_201_CREATED
)
async def add_to_wishlist(
request: WishlistCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> WishlistResponse:
"""Add a card to wishlist."""
user_id: int = int(current_user["user_id"])
# Check if already in wishlist
check = (
select(CardWishlist)
.where(CardWishlist.user_id == user_id, CardWishlist.card_id == request.card_id)
)
check_result = await db.execute(check)
if check_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Card already in wishlist")
item = CardWishlist(
user_id=user_id,
card_id=request.card_id,
max_price=request.max_price,
notes=request.notes,
created_at=datetime.now(timezone.utc)
)
db.add(item)
await db.flush()
return WishlistResponse.model_validate(item)
@user_data_router.get("/wishlist", response_model=WishlistListResponse)
async def get_wishlist(
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=100),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> WishlistListResponse:
"""Get user's wishlist."""
user_id: int = int(current_user["user_id"])
offset = (page - 1) * page_size
count_stmt = select(func.count()).select_from(CardWishlist).where(CardWishlist.user_id == user_id)
total_result = await db.execute(count_stmt)
total = total_result.scalar()
stmt = (
select(CardWishlist)
.where(CardWishlist.user_id == user_id)
.order_by(CardWishlist.created_at.desc())
.offset(offset)
.limit(page_size)
)
result = await db.execute(stmt)
items = result.scalars().all()
return WishlistListResponse(
items=[WishlistResponse.model_validate(i) for i in items],
total=total,
page=page,
page_size=page_size,
total_pages=(total + page_size - 1) // page_size
)
@user_data_router.patch(
"/wishlist/{item_id}",
response_model=WishlistResponse
)
async def update_wishlist_item(
item_id: int,
request: WishlistUpdate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> WishlistResponse:
"""Update a wishlist item."""
user_id: int = int(current_user["user_id"])
stmt = select(CardWishlist).where(
CardWishlist.id == item_id,
CardWishlist.user_id == user_id
)
result = await db.execute(stmt)
item = result.scalar_one_or_none()
if not item:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Wishlist item not found")
update_data = request.model_dump(exclude_unset=True)
stmt = (
update(CardWishlist)
.where(CardWishlist.id == item_id)
.values(**update_data)
)
await db.execute(stmt)
await db.flush()
stmt = select(CardWishlist).where(CardWishlist.id == item_id)
result = await db.execute(stmt)
updated = result.scalar_one_or_none()
return WishlistResponse.model_validate(updated)
@user_data_router.delete(
"/wishlist/{item_id}",
response_model=MessageResponse
)
async def remove_from_wishlist(
item_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> MessageResponse:
"""Remove a card from wishlist."""
user_id: int = int(current_user["user_id"])
stmt = select(CardWishlist).where(
CardWishlist.id == item_id,
CardWishlist.user_id == user_id
)
result = await db.execute(stmt)
item = result.scalar_one_or_none()
if not item:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Wishlist item not found")
stmt = delete(CardWishlist).where(CardWishlist.id == item_id)
await db.execute(stmt)
await db.flush()
return MessageResponse(message="Removed from wishlist")
# ===== Groups =====
@user_data_router.post(
"/groups",
response_model=GroupResponse,
status_code=status.HTTP_201_CREATED
)
async def create_group(
request: GroupCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GroupResponse:
"""Create a new user group."""
user_id: int = int(current_user["user_id"])
group = UserGroup(
name=request.name,
description=request.description,
owner_id=user_id,
is_public=request.is_public,
max_members=request.max_members,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc)
)
db.add(group)
await db.flush()
# Add creator as owner
member = GroupMember(
group_id=group.id,
user_id=user_id,
role="OWNER",
joined_at=datetime.now(timezone.utc)
)
db.add(member)
await db.flush()
return GroupResponse.model_validate(group)
@user_data_router.get("/groups", response_model=GroupListResponse)
async def get_user_groups(
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=100),
is_public: Optional[bool] = None,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GroupListResponse:
"""Get user's groups."""
user_id: int = int(current_user["user_id"])
offset = (page - 1) * page_size
# Get groups where user is member or owner
member_groups = select(GroupMember.group_id).where(GroupMember.user_id == user_id)
count_stmt = select(func.count()).select_from(UserGroup).where(
UserGroup.id.in_(member_groups)
)
if is_public is not None:
count_stmt = count_stmt.where(UserGroup.is_public == is_public)
total_result = await db.execute(count_stmt)
total = total_result.scalar()
stmt = (
select(UserGroup)
.where(UserGroup.id.in_(member_groups))
.order_by(UserGroup.created_at.desc())
.offset(offset)
.limit(page_size)
)
result = await db.execute(stmt)
groups = result.scalars().all()
# Get member counts
group_ids = [g.id for g in groups]
member_counts: Dict[int, int] = {}
if group_ids:
count_subquery = (
select(GroupMember.group_id, func.count().label('count'))
.where(GroupMember.group_id.in_(group_ids))
.group_by(GroupMember.group_id)
.subquery()
)
count_stmt = select(count_subquery.c.group_id, count_subquery.c.count).where(
count_subquery.c.group_id.in_(group_ids)
)
count_result = await db.execute(count_stmt)
member_counts = {row[0]: row[1] for row in count_result.fetchall()}
return GroupListResponse(
groups=[
GroupResponse(
**GroupResponse.model_validate(g).model_dump(),
member_count=member_counts.get(g.id, 0)
)
for g in groups
],
total=total,
page=page,
page_size=page_size,
total_pages=(total + page_size - 1) // page_size
)
@user_data_router.get(
"/groups/{group_id}",
response_model=GroupResponse
)
async def get_group(
group_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GroupResponse:
"""Get a specific group."""
stmt = select(UserGroup).where(UserGroup.id == group_id)
result = await db.execute(stmt)
group = result.scalar_one_or_none()
if not group:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
return GroupResponse.model_validate(group)
@user_data_router.patch(
"/groups/{group_id}",
response_model=GroupResponse
)
async def update_group(
group_id: int,
request: GroupUpdate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GroupResponse:
"""Update a group."""
user_id: int = int(current_user["user_id"])
# Verify user is owner
check = (
select(GroupMember)
.where(
GroupMember.group_id == group_id,
GroupMember.user_id == user_id,
GroupMember.role == "OWNER"
)
)
check_result = await db.execute(check)
if not check_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Only group owner can update group")
stmt = select(UserGroup).where(UserGroup.id == group_id)
result = await db.execute(stmt)
group = result.scalar_one_or_none()
if not group:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
update_data = request.model_dump(exclude_unset=True)
update_data["updated_at"] = datetime.now(timezone.utc)
stmt = (
update(UserGroup)
.where(UserGroup.id == group_id)
.values(**update_data)
)
await db.execute(stmt)
await db.flush()
stmt = select(UserGroup).where(UserGroup.id == group_id)
result = await db.execute(stmt)
updated = result.scalar_one_or_none()
return GroupResponse.model_validate(updated)
@user_data_router.delete(
"/groups/{group_id}",
response_model=MessageResponse
)
async def delete_group(
group_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> MessageResponse:
"""Delete a group."""
user_id: int = int(current_user["user_id"])
# Verify user is owner
check = (
select(GroupMember)
.where(
GroupMember.group_id == group_id,
GroupMember.user_id == user_id,
GroupMember.role == "OWNER"
)
)
check_result = await db.execute(check)
if not check_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Only group owner can delete group")
stmt = delete(UserGroup).where(UserGroup.id == group_id)
await db.execute(stmt)
await db.flush()
return MessageResponse(message="Group deleted successfully")
# ===== Group Members =====
@user_data_router.post(
"/groups/{group_id}/members",
status_code=status.HTTP_201_CREATED
)
async def add_group_member(
group_id: int,
request: GroupMemberCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> Dict[str, Any]:
"""Add a member to a group."""
user_id: int = int(current_user["user_id"])
# Verify user is admin or owner
check = (
select(GroupMember)
.where(
GroupMember.group_id == group_id,
GroupMember.user_id == user_id,
GroupMember.role.in_(["OWNER", "ADMIN"])
)
)
check_result = await db.execute(check)
if not check_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin privileges required")
# Check if already a member
member_check = (
select(GroupMember)
.where(GroupMember.group_id == group_id, GroupMember.user_id == request.user_id)
)
member_result = await db.execute(member_check)
if member_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Already a member")
member = GroupMember(
group_id=group_id,
user_id=request.user_id,
role=request.role.value,
joined_at=datetime.now(timezone.utc)
)
db.add(member)
await db.flush()
return {"message": "Member added", "member_id": member.id}
@user_data_router.patch(
"/groups/{group_id}/members/{member_id}",
response_model=GroupMemberUpdate
)
async def update_group_member(
group_id: int,
member_id: int,
request: GroupMemberUpdate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GroupMemberUpdate:
"""Update a group member's role."""
user_id: int = int(current_user["user_id"])
# Verify user is owner
check = (
select(GroupMember)
.where(
GroupMember.group_id == group_id,
GroupMember.user_id == user_id,
GroupMember.role == "OWNER"
)
)
check_result = await db.execute(check)
if not check_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Only group owner can update roles")
stmt = select(GroupMember).where(GroupMember.id == member_id)
result = await db.execute(stmt)
member = result.scalar_one_or_none()
if not member:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Member not found")
stmt = (
update(GroupMember)
.where(GroupMember.id == member_id)
.values(role=request.role.value)
)
await db.execute(stmt)
await db.flush()
return {"message": "Member role updated"}
@user_data_router.delete(
"/groups/{group_id}/members/{member_id}",
response_model=MessageResponse
)
async def remove_group_member(
group_id: int,
member_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> MessageResponse:
"""Remove a member from a group."""
user_id: int = int(current_user["user_id"])
# Verify user is admin or owner
check = (
select(GroupMember)
.where(
GroupMember.group_id == group_id,
GroupMember.user_id == user_id,
GroupMember.role.in_(["OWNER", "ADMIN"])
)
)
check_result = await db.execute(check)
if not check_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin privileges required")
stmt = select(GroupMember).where(GroupMember.id == member_id)
result = await db.execute(stmt)
member = result.scalar_one_or_none()
if not member:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Member not found")
stmt = delete(GroupMember).where(GroupMember.id == member_id)
await db.execute(stmt)
await db.flush()
return MessageResponse(message="Member removed from group")
# ===== Group Chat Messages =====
@user_data_router.post(
"/groups/{group_id}/messages",
response_model=GroupChatMessageResponse,
status_code=status.HTTP_201_CREATED
)
async def send_group_message(
group_id: int,
request: GroupChatMessageCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GroupChatMessageResponse:
"""Send a message in a group chat."""
user_id: int = int(current_user["user_id"])
# Verify user is a member
member_check = (
select(GroupMember)
.where(GroupMember.group_id == group_id, GroupMember.user_id == user_id)
)
member_result = await db.execute(member_check)
if not member_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not a group member")
# Verify group exists
group_check = select(UserGroup).where(UserGroup.id == group_id)
group_result = await db.execute(group_check)
if not group_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
message = GroupChatMessage(
group_id=group_id,
sender_id=user_id,
message=request.message,
created_at=datetime.now(timezone.utc)
)
db.add(message)
await db.flush()
# Get sender username
user_check = select(User).where(User.id == user_id)
user_result = await db.execute(user_check)
sender = user_result.scalar_one_or_none()
return GroupChatMessageResponse(
id=message.id,
group_id=message.group_id,
sender_id=message.sender_id,
sender_username=sender.username if sender else None,
message=message.message,
created_at=message.created_at
)
@user_data_router.get(
"/groups/{group_id}/messages",
response_model=GroupChatMessageListResponse
)
async def get_group_messages(
group_id: int,
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=100),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> GroupChatMessageListResponse:
"""Get messages in a group chat."""
offset = (page - 1) * page_size
# Verify user is a member
member_check = (
select(GroupMember)
.where(GroupMember.group_id == group_id, GroupMember.user_id == int(current_user["user_id"]))
)
member_result = await db.execute(member_check)
if not member_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not a group member")
count_stmt = select(func.count()).select_from(GroupChatMessage).where(GroupChatMessage.group_id == group_id)
total_result = await db.execute(count_stmt)
total = total_result.scalar()
stmt = (
select(GroupChatMessage)
.where(GroupChatMessage.group_id == group_id)
.order_by(GroupChatMessage.created_at.desc())
.offset(offset)
.limit(page_size)
)
result = await db.execute(stmt)
messages = result.scalars().all()
# Get sender usernames
sender_ids = [m.sender_id for m in messages]
sender_names: Dict[int, str] = {}
if sender_ids:
user_stmt = select(User.id, User.username).where(User.id.in_(sender_ids))
user_result = await db.execute(user_stmt)
sender_names = {row[0]: row[1] for row in user_result.fetchall()}
return GroupChatMessageListResponse(
messages=[
GroupChatMessageResponse(
id=m.id,
group_id=m.group_id,
sender_id=m.sender_id,
sender_username=sender_names.get(m.sender_id),
message=m.message,
created_at=m.created_at
)
for m in messages
],
total=total,
page=page,
page_size=page_size,
total_pages=(total + page_size - 1) // page_size
)
# ===== Networks =====
@user_data_router.post(
"/networks",
response_model=NetworkResponse,
status_code=status.HTTP_201_CREATED
)
async def create_network(
request: NetworkCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> NetworkResponse:
"""Create a new user network."""
user_id: int = int(current_user["user_id"])
network = UserNetwork(
name=request.name,
description=request.description,
creator_id=user_id,
is_public=request.is_public,
created_at=datetime.now(timezone.utc)
)
db.add(network)
await db.flush()
# Add creator as owner
member = NetworkMember(
network_id=network.id,
user_id=user_id,
role="OWNER",
joined_at=datetime.now(timezone.utc)
)
db.add(member)
await db.flush()
return NetworkResponse.model_validate(network)
@user_data_router.get("/networks", response_model=NetworkListResponse)
async def get_user_networks(
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=100),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> NetworkListResponse:
"""Get user's networks."""
user_id: int = int(current_user["user_id"])
offset = (page - 1) * page_size
member_networks = select(NetworkMember.network_id).where(NetworkMember.user_id == user_id)
count_stmt = select(func.count()).select_from(UserNetwork).where(
UserNetwork.id.in_(member_networks)
)
total_result = await db.execute(count_stmt)
total = total_result.scalar()
stmt = (
select(UserNetwork)
.where(UserNetwork.id.in_(member_networks))
.order_by(UserNetwork.created_at.desc())
.offset(offset)
.limit(page_size)
)
result = await db.execute(stmt)
networks = result.scalars().all()
# Get member counts
network_ids = [n.id for n in networks]
member_counts: Dict[int, int] = {}
if network_ids:
count_subquery = (
select(NetworkMember.network_id, func.count().label('count'))
.where(NetworkMember.network_id.in_(network_ids))
.group_by(NetworkMember.network_id)
.subquery()
)
count_stmt = select(count_subquery.c.network_id, count_subquery.c.count).where(
count_subquery.c.network_id.in_(network_ids)
)
count_result = await db.execute(count_stmt)
member_counts = {row[0]: row[1] for row in count_result.fetchall()}
return NetworkListResponse(
networks=[
NetworkResponse(
**NetworkResponse.model_validate(n).model_dump(),
member_count=member_counts.get(n.id, 0)
)
for n in networks
],
total=total,
page=page,
page_size=page_size,
total_pages=(total + page_size - 1) // page_size
)
@user_data_router.get(
"/networks/{network_id}",
response_model=NetworkResponse
)
async def get_network(
network_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> NetworkResponse:
"""Get a specific network."""
stmt = select(UserNetwork).where(UserNetwork.id == network_id)
result = await db.execute(stmt)
network = result.scalar_one_or_none()
if not network:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Network not found")
return NetworkResponse.model_validate(network)
@user_data_router.patch(
"/networks/{network_id}",
response_model=NetworkResponse
)
async def update_network(
network_id: int,
request: NetworkUpdate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> NetworkResponse:
"""Update a network."""
user_id: int = int(current_user["user_id"])
# Verify user is creator
check = (
select(UserNetwork)
.where(UserNetwork.id == network_id, UserNetwork.creator_id == user_id)
)
check_result = await db.execute(check)
if not check_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Only creator can update network")
stmt = select(UserNetwork).where(UserNetwork.id == network_id)
result = await db.execute(stmt)
network = result.scalar_one_or_none()
if not network:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Network not found")
update_data = request.model_dump(exclude_unset=True)
stmt = (
update(UserNetwork)
.where(UserNetwork.id == network_id)
.values(**update_data)
)
await db.execute(stmt)
await db.flush()
stmt = select(UserNetwork).where(UserNetwork.id == network_id)
result = await db.execute(stmt)
updated = result.scalar_one_or_none()
return NetworkResponse.model_validate(updated)
@user_data_router.delete(
"/networks/{network_id}",
response_model=MessageResponse
)
async def delete_network(
network_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> MessageResponse:
"""Delete a network."""
user_id: int = int(current_user["user_id"])
# Verify user is creator
check = (
select(UserNetwork)
.where(UserNetwork.id == network_id, UserNetwork.creator_id == user_id)
)
check_result = await db.execute(check)
if not check_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Only creator can delete network")
stmt = delete(UserNetwork).where(UserNetwork.id == network_id)
await db.execute(stmt)
await db.flush()
return MessageResponse(message="Network deleted successfully")
# ===== Network Members =====
@user_data_router.post(
"/networks/{network_id}/members",
status_code=status.HTTP_201_CREATED
)
async def add_network_member(
network_id: int,
request: NetworkMemberCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> Dict[str, Any]:
"""Add a member to a network."""
user_id: int = int(current_user["user_id"])
# Verify user is admin or owner
check = (
select(NetworkMember)
.where(
NetworkMember.network_id == network_id,
NetworkMember.user_id == user_id,
NetworkMember.role.in_(["OWNER", "ADMIN"])
)
)
check_result = await db.execute(check)
if not check_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin privileges required")
# Check if already a member
member_check = (
select(NetworkMember)
.where(NetworkMember.network_id == network_id, NetworkMember.user_id == request.user_id)
)
member_result = await db.execute(member_check)
if member_result.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Already a member")
member = NetworkMember(
network_id=network_id,
user_id=request.user_id,
role=request.role.value,
joined_at=datetime.now(timezone.utc)
)
db.add(member)
await db.flush()
return {"message": "Member added", "member_id": member.id}
# ===== User Preferences =====
@user_data_router.get(
"/preferences",
response_model=UserPreferenceResponse
)
async def get_user_preferences(
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> UserPreferenceResponse:
"""Get user preferences."""
user_id: int = int(current_user["user_id"])
stmt = select(UserPreference).where(UserPreference.user_id == user_id)
result = await db.execute(stmt)
prefs = result.scalar_one_or_none()
if not prefs:
# Create default preferences
prefs = UserPreference(
user_id=user_id,
theme="light",
notifications_enabled=True,
email_notifications=True,
auto_save_decks=True,
default_format="standard",
language="EN",
updated_at=datetime.now(timezone.utc)
)
db.add(prefs)
await db.flush()
return UserPreferenceResponse.model_validate(prefs)
@user_data_router.patch(
"/preferences",
response_model=UserPreferenceResponse
)
async def update_user_preferences(
request: UserPreferenceUpdate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> UserPreferenceResponse:
"""Update user preferences."""
user_id: int = int(current_user["user_id"])
stmt = select(UserPreference).where(UserPreference.user_id == user_id)
result = await db.execute(stmt)
prefs = result.scalar_one_or_none()
if not prefs:
# Create default preferences first
prefs = UserPreference(
user_id=user_id,
updated_at=datetime.now(timezone.utc)
)
db.add(prefs)
await db.flush()
update_data = request.model_dump(exclude_unset=True)
update_data["updated_at"] = datetime.now(timezone.utc)
stmt = (
update(UserPreference)
.where(UserPreference.user_id == user_id)
.values(**update_data)
)
await db.execute(stmt)
await db.flush()
stmt = select(UserPreference).where(UserPreference.user_id == user_id)
result = await db.execute(stmt)
updated = result.scalar_one_or_none()
return UserPreferenceResponse.model_validate(updated)
# ===== Activity Log =====
@user_data_router.get("/activity", response_model=ActivityLogListResponse)
async def get_activity_log(
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=100),
activity_type: Optional[str] = None,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> ActivityLogListResponse:
"""Get user activity log."""
user_id: int = int(current_user["user_id"])
offset = (page - 1) * page_size
conditions = [UserActivityLog.user_id == user_id]
if activity_type:
conditions.append(UserActivityLog.activity_type == activity_type)
count_stmt = select(func.count()).select_from(UserActivityLog)
for cond in conditions:
count_stmt = count_stmt.where(cond)
total_result = await db.execute(count_stmt)
total = total_result.scalar()
stmt = select(UserActivityLog)
for cond in conditions:
stmt = stmt.where(cond)
stmt = stmt.order_by(UserActivityLog.created_at.desc()).offset(offset).limit(page_size)
result = await db.execute(stmt)
entries = result.scalars().all()
return ActivityLogListResponse(
entries=[ActivityLogEntry.model_validate(e) for e in entries],
total=total,
page=page,
page_size=page_size,
total_pages=(total + page_size - 1) // page_size
)
# =============================================================================
# Card Import Router (prefix: /api/v1/card-import)
# =============================================================================
card_import_router = APIRouter()
@card_import_router.post(
"/import",
response_model=CardImportResponse,
status_code=status.HTTP_201_CREATED
)
async def upload_card_import(
file: UploadFile = File(...),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> CardImportResponse:
"""
Upload a card import file (XLSX, CSV, JSON, ODS).
Parses the file, performs fuzzy matching, and creates an import batch.
"""
user_id: int = int(current_user["user_id"])
# Validate file type
if not file.filename:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="File must have a filename"
)
file_type = file.filename.split(".")[-1].lower()
supported_types = ["xlsx", "csv", "json", "ods"]
if file_type not in supported_types:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Unsupported file type: {file_type}. Supported types: {supported_types}"
)
# Read file content
content = await file.read()
file_size = len(content)
with tempfile.NamedTemporaryFile(suffix=f".{file_type}", delete=False) as tmp_file:
tmp_file.write(content)
tmp_file_path = Path(tmp_file.name)
try:
card_names = await FileParser.parse_file(tmp_file_path)
except Exception as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Failed to parse file: {str(e)}"
)
if not card_names:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="File contains no card names"
)
# Create import batch
batch = await ImportBatchProcessor.create_batch(
db=db,
user_id=user_id,
filename=file.filename,
file_type=file_type,
file_size=file_size,
card_names=card_names,
)
# Process batch
result = await ImportBatchProcessor.process_batch(
db=db,
batch=batch,
card_names=card_names,
)
return CardImportResponse(
message=f"Import batch created successfully",
batch_id=batch.id,
card_count=len(card_names),
status=result["status"],
imported_at=datetime.utcnow(),
)
@card_import_router.get(
"/import/{import_id}/status",
response_model=CardImportStatusResponse
)
async def get_import_status(
import_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> CardImportStatusResponse:
"""Get the status of an import batch."""
user_id: int = int(current_user["user_id"])
batch = await ImportBatchProcessor.get_batch_status(db, import_id)
if not batch:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Import batch {import_id} not found"
)
if batch.user_id != user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied"
)
return CardImportStatusResponse(
has_import=True,
batch_id=batch.id,
status=batch.status,
card_count=batch.total_cards,
matched_count=batch.matched_cards,
unmatched_count=batch.unmatched_cards,
last_imported=batch.updated_at,
error_message=batch.error_message,
)
@card_import_router.get(
"/import/{import_id}/results",
response_model=CardImportSummary
)
async def get_import_results(
import_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> CardImportSummary:
"""Get the match results for an import batch."""
user_id: int = int(current_user["user_id"])
batch = await ImportBatchProcessor.get_batch_status(db, import_id)
if not batch:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Import batch {import_id} not found"
)
if batch.user_id != user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied"
)
if batch.status != "completed":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Import batch {import_id} has not completed yet (status: {batch.status})"
)
match_results = batch.match_results or []
matched_cards = []
unmatched_cards = []
for original_name, card_id, matched_name, confidence, match_type in match_results:
if matched_name:
matched_cards.append(CardMatchResult(
card_id=card_id,
card_name=original_name,
matched_name=matched_name,
match_type=match_type,
confidence=confidence,
))
else:
unmatched_cards.append(original_name)
return CardImportSummary(
total_cards=len(match_results),
matched_cards=matched_cards,
unmatched_cards=unmatched_cards,
import_id=batch.id,
)
@card_import_router.post(
"/import/{import_id}/confirm",
response_model=MessageResponse
)
async def confirm_import(
import_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> MessageResponse:
"""Confirm an import batch and save to user's card collection."""
user_id: int = int(current_user["user_id"])
batch = await ImportBatchProcessor.get_batch_status(db, import_id)
if not batch:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Import batch {import_id} not found"
)
if batch.user_id != user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied"
)
if batch.status != "completed":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Import batch {import_id} has not completed yet (status: {batch.status})"
)
# Confirm batch
try:
await ImportBatchProcessor.confirm_batch(db, import_id, user_id)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(e)
)
# Save matched cards to user's collection
match_results = batch.match_results or []
saved_count = 0
for original_name, card_id, matched_name, confidence, match_type in match_results:
if card_id and match_type in ["exact", "high_confidence"]:
# Check if already in collection
stmt = select(UserCardCollection).where(
UserCardCollection.user_id == user_id,
UserCardCollection.card_id == card_id,
)
result = await db.execute(stmt)
existing = result.scalar_one_or_none()
if existing:
existing.quantity += 1
else:
new_collection = UserCardCollection(
user_id=user_id,
card_id=card_id,
quantity=1,
)
db.add(new_collection)
saved_count += 1
await db.flush()
return MessageResponse(
message=f"Import confirmed. {saved_count} cards added to your collection."
)
@card_import_router.get(
"/user/cards",
response_model=List[Dict[str, Any]]
)
async def get_user_cards(
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> List[Dict[str, Any]]:
"""List user's imported cards."""
user_id: int = int(current_user["user_id"])
stmt = select(UserCardCollection).where(UserCardCollection.user_id == user_id).order_by(UserCardCollection.created_at.desc())
result = await db.execute(stmt)
collections = result.scalars().all()
# Fetch card details
card_ids = [c.card_id for c in collections]
card_details: Dict[int, MtgonlineCard] = {}
if card_ids:
card_stmt = select(MtgonlineCard).where(MtgonlineCard.id.in_(card_ids))
card_result = await db.execute(card_stmt)
for card in card_result.scalars().all():
card_details[card.id] = card
result_list = []
for collection in collections:
card = card_details.get(collection.card_id)
result_list.append({
"collection_id": collection.id,
"card_id": collection.card_id,
"card_name": card.name if card else f"Card#{collection.card_id}",
"card_type_line": card.type_line if card else "",
"quantity": collection.quantity,
"condition": collection.condition,
"language": collection.language,
"is_foil": collection.is_foil,
"is_alt_art": collection.is_alt_art,
"acquired_date": collection.acquired_date,
})
return result_list
@card_import_router.delete(
"/user/cards/{card_import_id}",
response_model=MessageResponse
)
async def delete_user_card(
card_import_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> MessageResponse:
"""Remove a card from user's collection."""
user_id: int = int(current_user["user_id"])
stmt = select(UserCardCollection).where(
UserCardCollection.id == card_import_id,
UserCardCollection.user_id == user_id,
)
result = await db.execute(stmt)
collection = result.scalar_one_or_none()
if not collection:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Card not found in your collection"
)
await db.delete(collection)
await db.flush()
return MessageResponse(message="Card removed from your collection")
@card_import_router.get(
"/user/decks",
response_model=List[Dict[str, Any]]
)
async def get_user_decks(
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> List[Dict[str, Any]]:
"""List user's decks.
Kept for backward compatibility. This duplicates functionality in decks.py.
"""
user_id: int = int(current_user["user_id"])
stmt = select(UserDeck).where(UserDeck.user_id == user_id).order_by(UserDeck.updated_at.desc())
result = await db.execute(stmt)
decks = result.scalars().all()
result_list = []
for deck in decks:
result_list.append({
"deck_id": deck.id,
"name": deck.name,
"status": deck.status,
"format": deck.format,
"folder_id": deck.folder_id,
"notes": deck.notes,
"is_precedent": deck.is_precedent,
"created_at": deck.created_at,
"updated_at": deck.updated_at,
})
return result_list