Files
mtgonline/backend/app/schemas/user_data_schemas.py
T
akadmin 1df04aea52 Phase 7: End-to-End API Testing - Database setup, migrations, and application fixes
- Switched from psycopg2 to asyncpg for async SQLAlchemy support
- Fixed router registration in main.py - removed duplicate prefixes
- Added user_data export to routers/__init__.py
- Refactored decks router to use DeckManager service layer
- Integrated FuzzyCardMatcher into card_router search endpoints
- Made WishlistCreate.card_id optional for proper schema validation
- Set PostgreSQL password and configured scram-sha-256 auth
- Updated alembic.ini to use local PostgreSQL instead of Docker hostname
- Created generic_schemas.py for reusable schema patterns
- Added test_routers.py and test_schema_validation.py test files
- All 6 Alembic migrations applied successfully (37 tables created)
- Application running on port 8000 with all services connected
2026-08-18 03:35:19 +00:00

435 lines
10 KiB
Python

"""
Pydantic schemas for user data features.
Covers sessions, decks, replays, cards, groups, networks, preferences, and activity logs.
"""
from pydantic import BaseModel, Field, ConfigDict
from typing import Optional, List, Dict, Any
from datetime import datetime
from enum import Enum
# ===== Enum Types =====
class DeckVersionStatus(str, Enum):
DRAFT = "DRAFT"
FINAL = "FINAL"
ARCHIVED = "ARCHIVED"
class GameReplayStatus(str, Enum):
IN_PROGRESS = "IN_PROGRESS"
COMPLETED = "COMPLETED"
FAILED = "FAILED"
CANCELLED = "CANCELLED"
class GameOutcomeType(str, Enum):
WIN = "WIN"
LOSS = "LOSS"
CONCESSION = "CONCESSION"
DISCONNECT = "DISCONNECT"
class GroupMemberRole(str, Enum):
OWNER = "OWNER"
ADMIN = "ADMIN"
MEMBER = "MEMBER"
class NetworkMemberRole(str, Enum):
OWNER = "OWNER"
ADMIN = "ADMIN"
MEMBER = "MEMBER"
class UserPreferenceTheme(str, Enum):
LIGHT = "light"
DARK = "dark"
SYSTEM = "system"
class ActivityType(str, Enum):
LOGIN = "LOGIN"
LOGOUT = "LOGOUT"
DECK_EDIT = "DECK_EDIT"
GAME_PLAYED = "GAME_PLAYED"
CARD_ACQUIRED = "CARD_ACQUIRED"
CARD_TRADED = "CARD_TRADED"
GROUP_CREATED = "GROUP_CREATED"
GROUP_JOINED = "GROUP_JOINED"
# ===== Session Schemas =====
class SessionResponse(BaseModel):
"""User session response."""
id: int
user_id: int
ip_address: Optional[str] = None
user_agent: Optional[str] = None
created_at: datetime
expires_at: datetime
is_active: bool
model_config = ConfigDict(from_attributes=True)
class SessionCleanupResponse(BaseModel):
"""Response after cleaning expired sessions."""
cleaned_count: int
message: str
# ===== Deck Version Schemas =====
class DeckVersionCreate(BaseModel):
"""Deck version creation request."""
content: str = Field(..., min_length=1)
status: DeckVersionStatus = DeckVersionStatus.DRAFT
comment: Optional[str] = None
class DeckVersionUpdate(BaseModel):
"""Deck version update request."""
content: Optional[str] = None
status: Optional[DeckVersionStatus] = None
comment: Optional[str] = None
class DeckVersionResponse(BaseModel):
"""Deck version response."""
id: int
deck_id: int
version_number: int
content: str
status: str
comment: Optional[str]
created_at: datetime
model_config = ConfigDict(from_attributes=True)
class DeckVersionListResponse(BaseModel):
"""List of deck versions."""
versions: List[DeckVersionResponse]
total: int
# ===== Game Replay Schemas =====
class GameReplayCreate(BaseModel):
"""Game replay creation request."""
game_uuid: str = Field(..., min_length=36, max_length=36)
room_id: Optional[int] = None
game_type: Optional[str] = None
format: Optional[str] = None
duration_seconds: Optional[int] = None
start_time: datetime
end_time: Optional[datetime] = None
status: GameReplayStatus = GameReplayStatus.IN_PROGRESS
replay_data: Optional[Dict[str, Any]] = None
class GameReplayUpdate(BaseModel):
"""Game replay update request."""
room_id: Optional[int] = None
game_type: Optional[str] = None
format: Optional[str] = None
duration_seconds: Optional[int] = None
end_time: Optional[datetime] = None
status: Optional[GameReplayStatus] = None
replay_data: Optional[Dict[str, Any]] = None
class GameReplayResponse(BaseModel):
"""Game replay response."""
id: int
game_uuid: str
room_id: Optional[int]
game_type: Optional[str]
format: Optional[str]
duration_seconds: Optional[int]
start_time: datetime
end_time: Optional[datetime]
status: str
replay_data: Optional[Dict[str, Any]]
created_at: datetime
updated_at: datetime
players: List[Dict[str, Any]] = []
model_config = ConfigDict(from_attributes=True)
class GameReplayListResponse(BaseModel):
"""List of game replays."""
replays: List[GameReplayResponse]
total: int
page: int
page_size: int
total_pages: int
# ===== Game Outcome Schemas =====
class GameOutcomeCreate(BaseModel):
"""Game outcome creation request."""
game_uuid: str
outcome: GameOutcomeType
opponent_id: Optional[int] = None
format: Optional[str] = None
rating_before: Optional[int] = None
rating_after: Optional[int] = None
rating_change: Optional[int] = None
class GameOutcomeResponse(BaseModel):
"""Game outcome response."""
id: int
user_id: int
game_uuid: str
outcome: str
opponent_id: Optional[int]
format: Optional[str]
rating_before: Optional[int]
rating_after: Optional[int]
rating_change: Optional[int]
created_at: datetime
model_config = ConfigDict(from_attributes=True)
class GameOutcomeListResponse(BaseModel):
"""List of game outcomes."""
outcomes: List[GameOutcomeResponse]
total: int
# ===== User Statistics Schemas =====
class UserStatisticsResponse(BaseModel):
"""User statistics summary response."""
user_id: int
total_games: int
total_wins: int
total_losses: int
total_concessions: int
win_rate: float
current_streak: int
best_streak: int
average_rating: float
last_game_date: Optional[datetime]
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
class StatisticsUpdateResponse(BaseModel):
"""Response after updating statistics."""
user_id: int
total_games: int
total_wins: int
total_losses: int
win_rate: float
current_streak: int
updated_at: datetime
# ===== Card Collection Schemas (moved to user_card_collection.py) =====
# These are kept here for backward compatibility but should be imported from user_card_collection.py
# ===== Wishlist Schemas (moved to user_card_collection.py) =====
# These are kept here for backward compatibility but should be imported from user_card_collection.py
# ===== Group Schemas =====
class GroupCreate(BaseModel):
"""User group creation request."""
name: str = Field(..., min_length=1, max_length=100)
description: Optional[str] = None
is_public: bool = True
max_members: int = Field(50, ge=2, le=500)
class GroupUpdate(BaseModel):
"""User group update request."""
name: Optional[str] = None
description: Optional[str] = None
is_public: Optional[bool] = None
max_members: Optional[int] = None
class GroupMemberCreate(BaseModel):
"""Group member addition request."""
user_id: int
role: GroupMemberRole = GroupMemberRole.MEMBER
class GroupMemberUpdate(BaseModel):
"""Group member role update request."""
role: GroupMemberRole
class GroupMemberRemove(BaseModel):
"""Group member removal request."""
user_id: int
class GroupResponse(BaseModel):
"""User group response."""
id: int
name: str
description: Optional[str]
owner_id: int
is_public: bool
max_members: int
created_at: datetime
updated_at: datetime
member_count: int = 0
is_member: bool = False
model_config = ConfigDict(from_attributes=True)
class GroupListResponse(BaseModel):
"""List of user groups."""
groups: List[GroupResponse]
total: int
class GroupChatMessageCreate(BaseModel):
"""Group chat message creation request."""
message: str = Field(..., min_length=1, max_length=2000)
class GroupChatMessageResponse(BaseModel):
"""Group chat message response."""
id: int
group_id: int
sender_id: int
sender_username: Optional[str] = None
message: str
created_at: datetime
model_config = ConfigDict(from_attributes=True)
class GroupChatMessageListResponse(BaseModel):
"""List of group chat messages."""
messages: List[GroupChatMessageResponse]
total: int
page: int
page_size: int
total_pages: int
# ===== Network Schemas =====
class NetworkCreate(BaseModel):
"""User network creation request."""
name: str = Field(..., min_length=1, max_length=100)
description: Optional[str] = None
is_public: bool = True
class NetworkUpdate(BaseModel):
"""User network update request."""
name: Optional[str] = None
description: Optional[str] = None
is_public: Optional[bool] = None
class NetworkMemberCreate(BaseModel):
"""Network member addition request."""
user_id: int
role: NetworkMemberRole = NetworkMemberRole.MEMBER
class NetworkResponse(BaseModel):
"""User network response."""
id: int
name: str
description: Optional[str]
creator_id: int
is_public: bool
created_at: datetime
member_count: int = 0
is_member: bool = False
model_config = ConfigDict(from_attributes=True)
class NetworkListResponse(BaseModel):
"""List of user networks."""
networks: List[NetworkResponse]
total: int
# ===== Preference Schemas =====
class UserPreferenceUpdate(BaseModel):
"""User preference update request."""
theme: Optional[UserPreferenceTheme] = None
notifications_enabled: Optional[bool] = None
email_notifications: Optional[bool] = None
auto_save_decks: Optional[bool] = None
default_format: Optional[str] = None
language: Optional[str] = None
class UserPreferenceResponse(BaseModel):
"""User preference response."""
user_id: int
theme: str
notifications_enabled: bool
email_notifications: bool
auto_save_decks: bool
default_format: str
language: str
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
# ===== Activity Log Schemas =====
class ActivityLogEntry(BaseModel):
"""Activity log entry."""
id: int
user_id: int
activity_type: str
activity_data: Optional[Dict[str, Any]]
ip_address: Optional[str]
created_at: datetime
model_config = ConfigDict(from_attributes=True)
class ActivityLogListResponse(BaseModel):
"""List of activity log entries."""
entries: List[ActivityLogEntry]
total: int
page: int
page_size: int
total_pages: int
# ===== Generic Response Schemas =====
class MessageResponse(BaseModel):
"""Generic message response."""
message: str
class CountResponse(BaseModel):
"""Generic count response."""
count: int
class ErrorDetail(BaseModel):
"""Error detail."""
error: str
detail: str