Phase 3: Schema Layer - Pydantic v2 migration, deduplication, and missing schemas

- Migrated all schemas to Pydantic v2 syntax (model_config, ConfigDict)
- Fixed mutable default in ProtoMessageBase using Field(default_factory=datetime.now)
- Consolidated CardCollection and Wishlist schemas in user_card_collection.py
- Created game_schemas.py with GameCreate, GameResponse, GameJoinRequest, etc.
- Created mtg_card_schemas.py with MtgCardResponse, MtgCardSearchRequest, etc.
- Added CardImportBatchCreate, CardImportBatchResponse, UserCardImportCreate/Response schemas
- Fixed duplicate UserCardImportRecord class between card_import_batch.py and user_card_import_record.py
- Updated __init__.py with comprehensive schema exports
- Created verify_schemas.py for schema-model matching verification
This commit is contained in:
2026-08-16 05:22:17 +00:00
parent 5f324cf8a9
commit bea91db64d
20 changed files with 2733 additions and 342 deletions
+53 -15
View File
@@ -1,5 +1,13 @@
"""SQLAlchemy ORM model for card import batches."""
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Text, JSON, Boolean
"""
SQLAlchemy ORM models for card import tracking.
This module provides models for tracking card import batches
and individual card import records.
"""
from sqlalchemy import (
Column, Integer, String, BigInteger, Boolean, DateTime, Text,
ForeignKey, UniqueConstraint
)
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.core.database import Base
@@ -7,30 +15,60 @@ from app.core.database import Base
class CardImportBatch(Base):
"""
Card import batch record.
Card import batch tracking.
Tracks a single file import with its status and match results.
Tracks a batch of card imports from a file, including
matching results and error information.
"""
__tablename__ = "card_import_batches"
id = Column(Integer, primary_key=True, autoincrement=True)
user_id = Column(Integer, ForeignKey("mtgonline_users.id", ondelete="CASCADE"), nullable=False, index=True)
id = Column(BigInteger, primary_key=True, autoincrement=True)
user_id = Column(Integer, ForeignKey("mtgonline_users.id", ondelete="CASCADE"), nullable=False)
filename = Column(String(255), nullable=False)
file_type = Column(String(10), nullable=False) # xlsx, csv, json, ods
file_size = Column(Integer, nullable=False) # File size in bytes
status = Column(String(20), nullable=False, default="pending", index=True) # pending, processing, completed, failed
file_type = Column(String(50), nullable=True) # 'mtgjson', 'csv', etc.
file_size = Column(BigInteger, nullable=True) # Size in bytes
status = Column(String(20), default="PENDING") # PENDING, PROCESSING, COMPLETED, FAILED
total_cards = Column(Integer, default=0)
matched_cards = Column(Integer, default=0)
unmatched_cards = Column(Integer, default=0)
match_results = Column(JSON, nullable=True) # Store match results for later retrieval
match_results = Column(Text, nullable=True) # JSON of match details
error_message = Column(Text, nullable=True)
created_at = Column(DateTime, server_default=func.now())
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
# Relationships
user = relationship("User", backref="import_batches")
user = relationship("User", backref="card_import_batches")
records = relationship(
"UserCardImportRecord",
back_populates="batch",
cascade="all, delete-orphan"
)
def __repr__(self) -> str:
return f"<CardImportBatch id={self.id} user={self.user_id} status={self.status}>"
return f"<CardImportBatch {self.filename} (status={self.status})>"
class UserCardImportRecord(Base):
"""
Individual card import record within a batch.
Tracks whether each card in an import batch was confirmed
and when the confirmation happened.
"""
__tablename__ = "user_card_imports_confirmed"
id = Column(BigInteger, primary_key=True, autoincrement=True)
user_id = Column(Integer, ForeignKey("mtgonline_users.id", ondelete="CASCADE"), nullable=False)
batch_id = Column(BigInteger, ForeignKey("card_import_batches.id", ondelete="CASCADE"), nullable=False)
is_confirmed = Column(Boolean, default=False)
confirmed_at = Column(DateTime, nullable=True)
# Relationships
batch = relationship("CardImportBatch", back_populates="records")
__table_args__ = (
UniqueConstraint('user_id', 'batch_id', name='uq_user_card_import_batch'),
)
def __repr__(self) -> str:
return f"<UserCardImportRecord user={self.user_id} batch={self.batch_id} confirmed={self.is_confirmed}>"