-
Notifications
You must be signed in to change notification settings - Fork 18
Description
Problem
The application currently maintains duplicate Player model definitions across two separate layers, creating significant maintenance overhead:
Current Architecture:
schemas/player_schema.py→ SQLAlchemy ORM class (11 attributes)models/player_model.py→ Pydantic validation model (11 attributes)
Pain Points:
- Duplication: Every field is defined twice with slightly different syntax
- Synchronization Risk: Schema changes require updating both files, risking inconsistencies
- Maintenance Burden: Adding/modifying fields requires coordinated changes across multiple files
- Translation Overhead: Converting between SQLAlchemy and Pydantic objects adds boilerplate
- Testing Complexity: Both model types need separate test fixtures and validation
Example of current duplication:
# schemas/player_schema.py
first_name = Column(String, name="firstName", nullable=False)
# models/player_model.py
first_name: str # Must manually keep in syncThis violates the DRY (Don't Repeat Yourself) principle and increases the likelihood of bugs during schema evolution.
Proposed Solution
Consolidate both layers into a single SQLModel-based class that serves dual purposes:
- Acts as the SQLAlchemy ORM model for database operations
- Acts as the Pydantic model for request/response validation and OpenAPI documentation
Benefits:
- Single Source of Truth: One class definition for all player data
- Automatic Validation: Inherits Pydantic's validation at the ORM level
- Type Safety: Full type hints work for both DB and API layers
- Simplified Codebase: Eliminates conversion logic between models
- FastAPI Native: SQLModel is created by the same author as FastAPI (Sebastián Ramírez)
- Migration Path: Gradual migration possible without breaking existing code
SQLModel combines the best of both worlds:
class Player(SQLModel, table=True):
# One definition, two purposes ✨
first_name: str = Field(alias="firstName")Suggested Approach
Phase 1: Setup and Preparation
1.1 Install SQLModel
pip install sqlmodelAdd to requirements.txt:
sqlmodel>=0.0.14
1.2 Update Database Configuration
Modify databases/player_database.py to support SQLModel:
"""
Database setup and session management for async SQLModel with SQLite.
"""
import logging
import os
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlmodel import SQLModel # Replace declarative_base
storage_path = os.getenv("STORAGE_PATH", "./storage/players-sqlite3.db")
DATABASE_URL = f"sqlite+aiosqlite:///{storage_path}"
logger = logging.getLogger("uvicorn")
logging.getLogger("sqlalchemy.engine.Engine").handlers = logger.handlers
async_engine = create_async_engine(
DATABASE_URL, connect_args={"check_same_thread": False}, echo=True
)
async_sessionmaker = sessionmaker(
bind=async_engine, class_=AsyncSession, autocommit=False, autoflush=False
)
# SQLModel uses SQLModel.metadata instead of Base
# Keep Base as alias for backward compatibility during migration
Base = SQLModel
async def generate_async_session() -> AsyncGenerator[AsyncSession, None]:
"""Dependency function to yield an async SQLAlchemy ORM session."""
async with async_sessionmaker() as async_session:
yield async_sessionPhase 2: Create SQLModel Player Class
2.1 Create New Unified Model
Create models/player_sqlmodel.py (temporary file during migration):
"""
Unified SQLModel class for Player - combines ORM and validation.
This model serves both as:
- SQLAlchemy ORM table for database operations
- Pydantic model for API validation and serialization
"""
from typing import Optional
from sqlmodel import SQLModel, Field
class Player(SQLModel, table=True):
"""
Unified Player model using SQLModel.
Combines SQLAlchemy ORM functionality with Pydantic validation.
Uses camelCase aliases for API compatibility.
"""
__tablename__ = "players"
id: Optional[int] = Field(default=None, primary_key=True)
first_name: str = Field(sa_column_kwargs={"name": "firstName"}, alias="firstName")
middle_name: Optional[str] = Field(default=None, sa_column_kwargs={"name": "middleName"}, alias="middleName")
last_name: str = Field(sa_column_kwargs={"name": "lastName"}, alias="lastName")
date_of_birth: Optional[str] = Field(default=None, sa_column_kwargs={"name": "dateOfBirth"}, alias="dateOfBirth")
squad_number: int = Field(sa_column_kwargs={"name": "squadNumber", "unique": True}, alias="squadNumber")
position: str = Field()
abbr_position: Optional[str] = Field(default=None, sa_column_kwargs={"name": "abbrPosition"}, alias="abbrPosition")
team: Optional[str] = Field(default=None)
league: Optional[str] = Field(default=None)
starting11: Optional[bool] = Field(default=None)
class Config:
"""Pydantic configuration for the model."""
populate_by_name = True # Allow both snake_case and camelCase
# Optional: Create specific schemas for different use cases
class PlayerCreate(SQLModel):
"""Schema for creating a new player (excludes id)."""
first_name: str = Field(alias="firstName")
middle_name: Optional[str] = Field(default=None, alias="middleName")
last_name: str = Field(alias="lastName")
date_of_birth: Optional[str] = Field(default=None, alias="dateOfBirth")
squad_number: int = Field(alias="squadNumber")
position: str
abbr_position: Optional[str] = Field(default=None, alias="abbrPosition")
team: Optional[str] = Field(default=None)
league: Optional[str] = Field(default=None)
starting11: Optional[bool] = Field(default=None)
class PlayerRead(SQLModel):
"""Schema for reading player data (includes all fields)."""
id: int
first_name: str = Field(alias="firstName")
middle_name: Optional[str] = Field(default=None, alias="middleName")
last_name: str = Field(alias="lastName")
date_of_birth: Optional[str] = Field(default=None, alias="dateOfBirth")
squad_number: int = Field(alias="squadNumber")
position: str
abbr_position: Optional[str] = Field(default=None, alias="abbrPosition")
team: Optional[str] = Field(default=None)
league: Optional[str] = Field(default=None)
starting11: Optional[bool] = Field(default=None)Note on sa_column_kwargs: This is used to pass SQLAlchemy-specific column arguments (like database column name) while keeping the Pydantic alias separate.
Phase 3: Update Services and Routes
3.1 Update Service Layer
Modify services/player_service.py to use the new SQLModel:
from models.player_sqlmodel import Player, PlayerCreate, PlayerRead
from sqlmodel import select
# Example: Updated query method
async def get_player_by_id(session: AsyncSession, player_id: int) -> Optional[Player]:
"""Retrieve a player by ID using SQLModel."""
result = await session.execute(select(Player).where(Player.id == player_id))
return result.scalar_one_or_none()3.2 Update Route Handlers
Modify routes/player_route.py:
from models.player_sqlmodel import Player, PlayerCreate, PlayerRead
@api_router.post("/players", response_model=PlayerRead, status_code=201)
async def create_player(
player_data: PlayerCreate, # Input validation
session: AsyncSession = Depends(generate_async_session)
):
"""Create a new player using unified SQLModel."""
player = Player(**player_data.dict())
session.add(player)
await session.commit()
await session.refresh(player)
return player # Automatically serialized to PlayerRead
@api_router.get("/players/{player_id}", response_model=PlayerRead)
async def get_player(
player_id: int,
session: AsyncSession = Depends(generate_async_session)
):
"""Retrieve a player by ID."""
player = await player_service.get_player_by_id(session, player_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
return player # No conversion needed!Phase 4: Migration and Cleanup
4.1 Database Migration Compatibility
Ensure Alembic works with SQLModel:
# alembic/env.py
from sqlmodel import SQLModel
from models.player_sqlmodel import Player
target_metadata = SQLModel.metadata4.2 Update Tests
Modify tests/player_stub.py and tests/test_main.py:
from models.player_sqlmodel import Player
# Simplified test fixture
def create_test_player() -> Player:
return Player(
id=1,
first_name="John",
last_name="Doe",
squad_number=10,
position="Forward"
)4.3 Remove Legacy Code
Once all services and routes are migrated:
- Delete
schemas/player_schema.py - Delete
models/player_model.py(or rename if keeping for reference) - Rename
models/player_sqlmodel.py→models/player_model.py - Update all imports across the codebase
4.4 Update Documentation
Update README.md to reflect the new architecture:
## Project Architecture
- `databases/` - Database connection and session management
- `models/` - SQLModel classes (unified ORM + validation)
- `routes/` - FastAPI route handlers
- `services/` - Business logic layerAcceptance Criteria
- SQLModel is installed and added to
requirements.txt -
databases/player_database.pyis updated to work with SQLModel - A unified
PlayerSQLModel class is created with all 11 attributes - CamelCase field naming is preserved via
aliasandsa_column_kwargs - All CRUD operations in
services/player_service.pyuse the SQLModel class - All route handlers in
routes/player_route.pyare updated to use SQLModel - Legacy
schemas/player_schema.pyis removed - Legacy Pydantic
PlayerModelinmodels/player_model.pyis removed - FastAPI OpenAPI docs correctly show camelCase field names
- All existing integration tests pass without modifications
- New test fixtures use the SQLModel class
- Database migrations (Alembic) work with SQLModel metadata
-
player_stub.pyis updated to use SQLModel - Documentation is updated to reflect the new architecture
- No runtime errors occur when creating, reading, updating, or deleting players
- API responses maintain the same JSON structure (backward compatibility)
References
- [SQLModel Documentation](https://sqlmodel.tiangolo.com/)
- [SQLModel Tutorial](https://sqlmodel.tiangolo.com/tutorial/)
- [FastAPI with SQLModel](https://sqlmodel.tiangolo.com/tutorial/fastapi/)
- [SQLModel GitHub Repository](https://github.com/tiangolo/sqlmodel)
- [Alembic with SQLModel](https://sqlmodel.tiangolo.com/tutorial/create-db-and-table/#alembic-migrations)
- Related: Database Migration Tool Implementation Issue
- [Pydantic Field Aliases](https://docs.pydantic.dev/latest/concepts/fields/#field-aliases)
Migration Strategy Notes
Gradual Migration Approach
To minimize risk, consider this phased rollout:
- Phase 1: Create SQLModel class alongside existing models
- Phase 2: Update one service/route at a time
- Phase 3: Run tests after each change
- Phase 4: Remove legacy code only after everything works
Backward Compatibility
The API contract remains unchanged:
- Same endpoints
- Same request/response formats
- Same camelCase field names in JSON
- Existing clients continue working without changes
Rollback Plan
If issues arise:
- Keep legacy files temporarily (don't delete immediately)
- Git branch allows easy revert
- Feature flag could control which model system is active