- 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
89 lines
3.6 KiB
Python
89 lines
3.6 KiB
Python
"""Verify that Pydantic schemas match SQLAlchemy models."""
|
|
import sys
|
|
from sqlalchemy import inspect
|
|
|
|
# Import all schemas
|
|
from app.schemas import *
|
|
from app.models import *
|
|
|
|
def check_schema_model_match(schema_cls, model_cls, schema_name, model_name):
|
|
"""Check if a schema matches a model's fields."""
|
|
issues = []
|
|
|
|
# Get model columns
|
|
mapper = inspect(model_cls)
|
|
model_columns = {col.key for col in mapper.columns}
|
|
|
|
# Get schema fields
|
|
schema_fields = set(schema_cls.model_fields.keys())
|
|
|
|
# Check for extra fields in schema
|
|
extra = schema_fields - model_columns
|
|
if extra:
|
|
issues.append(f" Extra fields in {schema_name}: {extra}")
|
|
|
|
# Check for missing fields in schema (optional - some fields might be computed)
|
|
missing = model_columns - schema_fields
|
|
# Filter out common computed/relationship fields
|
|
computed_fields = {'created_at', 'updated_at', 'id'}
|
|
missing_actual = missing - computed_fields
|
|
if missing_actual:
|
|
issues.append(f" Missing fields in {schema_name}: {missing_actual}")
|
|
|
|
return issues
|
|
|
|
def main():
|
|
print("Schema-Model Verification")
|
|
print("=" * 60)
|
|
|
|
# Define schema-model pairs to check
|
|
pairs = [
|
|
(UserResponse, User, "UserResponse", "User"),
|
|
(DeckResponse, DecklistFile, "DeckResponse", "DecklistFile"),
|
|
(FolderResponse, DecklistFolder, "FolderResponse", "DecklistFolder"),
|
|
(GameResponse, None, "GameResponse", "Game (no model)"),
|
|
(RoomResponse, Room, "RoomResponse", "Room"),
|
|
(BanResponse, Ban, "BanResponse", "Ban"),
|
|
(CardMirrorResponse, MtgCardMirror, "CardMirrorResponse", "MtgCardMirror"),
|
|
(SessionResponse, UserSession, "SessionResponse", "UserSession"),
|
|
(DeckVersionResponse, DeckVersion, "DeckVersionResponse", "DeckVersion"),
|
|
(GameReplayResponse, GameReplay, "GameReplayResponse", "GameReplay"),
|
|
(GameOutcomeResponse, GameOutcome, "GameOutcomeResponse", "GameOutcome"),
|
|
(UserStatisticsResponse, UserStatistics, "UserStatisticsResponse", "UserStatistics"),
|
|
(CardCollectionResponse, UserCardCollection, "CardCollectionResponse", "UserCardCollection"),
|
|
(WishlistResponse, CardWishlist, "WishlistResponse", "CardWishlist"),
|
|
(GroupResponse, UserGroup, "GroupResponse", "UserGroup"),
|
|
(NetworkResponse, UserNetwork, "NetworkResponse", "UserNetwork"),
|
|
(UserPreferenceResponse, UserPreference, "UserPreferenceResponse", "UserPreference"),
|
|
(UserDeckResponse, UserDeck, "UserDeckResponse", "UserDeck"),
|
|
(DeckCardResponse, UserDeckCard, "DeckCardResponse", "UserDeckCard"),
|
|
(PrecedentResponse, DeckPrecedent, "PrecedentResponse", "DeckPrecedent"),
|
|
(SuggestionResponse, CardSuggestion, "SuggestionResponse", "CardSuggestion"),
|
|
]
|
|
|
|
total_issues = 0
|
|
for schema_cls, model_cls, schema_name, model_name in pairs:
|
|
if model_cls is None:
|
|
print(f"⏭️ {schema_name} - {model_name} (skipped - no direct model)")
|
|
continue
|
|
|
|
issues = check_schema_model_match(schema_cls, model_cls, schema_name, model_name)
|
|
if issues:
|
|
print(f"❌ {schema_name} vs {model_name}:")
|
|
for issue in issues:
|
|
print(issue)
|
|
total_issues += len(issues)
|
|
else:
|
|
print(f"✅ {schema_name} matches {model_name}")
|
|
|
|
print("\n" + "=" * 60)
|
|
if total_issues == 0:
|
|
print("✅ All schema-model pairs verified successfully!")
|
|
return 0
|
|
else:
|
|
print(f"⚠️ Found {total_issues} issues. Review needed.")
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|