[FEATURE] Implement Alembic for Database Migrations

[nanotaboada]

Problem

The FastAPI project currently lacks a robust database migration management system. Without proper migration tooling, schema changes are difficult to track, version control, and deploy consistently across environments. This creates risks during database evolution and makes collaboration challenging when multiple developers work on schema changes.

Additional Challenge: The project uses async SQLAlchemy with aiosqlite, which requires special consideration when setting up Alembic to ensure compatibility with the async database engine and session management.

Project Architecture Note: The project has an unconventional but functional structure:

• models/player_model.py → Pydantic models for API validation
• schemas/player_schema.py → SQLAlchemy ORM models (actual database tables)

This is opposite to the typical FastAPI convention where "schemas" are Pydantic and "models" are SQLAlchemy, but Alembic will work with the existing structure.

Proposed Solution

Implement Alembic as the database migration tool for the FastAPI project. Alembic will provide:

• Automatic migration generation from SQLAlchemy model changes
• Version-controlled schema changes that can be tracked in Git
• Bidirectional migrations (upgrade/downgrade capabilities)
• Environment-specific configurations for development, staging, and production
• Seamless SQLAlchemy integration leveraging existing ORM models

Why Alembic over Prisma Client?

After evaluation, Alembic is the recommended choice for the following reasons:

Alembic Advantages:

• Native SQLAlchemy integration (built by the same author)
• Pure Python implementation, no additional toolchain required
• Mature ecosystem with extensive community support
• Fine-grained control over migration scripts
• Handles complex migration scenarios (data migrations, custom SQL)
• Works with any SQLAlchemy-supported database

Prisma Client Limitations:

• Requires Node.js runtime alongside Python
• Uses its own schema definition language (Prisma Schema), not SQLAlchemy models
• Would require maintaining two sources of truth for database schema
• Less mature Python support (Prisma primarily TypeScript-focused)
• Additional complexity managing cross-language tooling

Suggested Approach

1. Project Structure

project_root/
├── alembic/
│   ├── versions/          # Migration scripts (to be created)
│   ├── env.py            # Migration environment config (to be created)
│   ├── script.py.mako    # Migration template (to be created)
│   └── README
├── alembic.ini           # Alembic configuration (to be created)
├── databases/
│   ├── __init__.py
│   └── player_database.py  # Async SQLAlchemy setup ✓
├── models/
│   ├── __init__.py
│   └── player_model.py    # Pydantic models for API ✓
├── schemas/
│   ├── __init__.py
│   └── player_schema.py   # SQLAlchemy ORM models ✓
├── storage/
│   └── players-sqlite3.db
├── main.py
└── requirements.txt

Note: The project uses an unconventional naming where "schemas" contains SQLAlchemy ORM and "models" contains Pydantic. This is functional and Alembic will work with it.

2. Installation and Setup

• Add alembic to requirements.txt
• Initialize Alembic in project root: alembic init alembic
• Configure alembic.ini with database connection string
• Update alembic/env.py to import SQLAlchemy models from schemas/player_schema.py

3. Configuration

• Set up environment variable support for database URLs (using existing STORAGE_PATH env var)
• Configure target_metadata in env.py to reference SQLAlchemy Base.metadata
• Important: Configure Alembic to work with async SQLAlchemy engine using run_async mode
• Implement offline and online migration modes
• Add logging configuration for migration tracking

4. Migration Workflow

• Auto-generate migrations: alembic revision --autogenerate -m "description"
• Review generated scripts: Manually verify migration logic
• Apply migrations: alembic upgrade head
• Rollback if needed: alembic downgrade -1

5. Key Files to Modify/Create

alembic/env.py:

import asyncio
import os
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context

# Import your Base and ORM models
from databases.player_database import Base
from schemas.player_schema import Player  # Import SQLAlchemy ORM models

# Get database URL from environment
storage_path = os.getenv("STORAGE_PATH", "./storage/players-sqlite3.db")
database_url = f"sqlite+aiosqlite:///{storage_path}"

# Alembic Config object
config = context.config
config.set_main_option("sqlalchemy.url", database_url)

# Set up logging
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# Set target metadata for autogenerate
target_metadata = Base.metadata


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode."""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    """Run migrations in 'online' mode with async engine."""
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""
    asyncio.run(run_async_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

alembic.ini:

• Configure sqlalchemy.url to use STORAGE_PATH environment variable
• Set up logging levels and output formats
• Note: The URL will be dynamically set in env.py from the environment variable

schemas/player_schema.py:

• No changes needed - already properly defined with SQLAlchemy columns
• Current implementation includes camelCase column names (e.g., name="firstName")
• Alembic will detect these column definitions for migration generation

main.py:

• Consider adding startup migration check (optional):

@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    """Lifespan event handler for FastAPI."""
    logger.info("Application starting up...")
    
    # Optional: Check if migrations are up to date
    # Note: This is informational only, doesn't auto-migrate
    # from alembic.config import Config
    # from alembic import command
    # alembic_cfg = Config("alembic.ini")
    # command.check(alembic_cfg)
    
    logger.info("Lifespan event handler execution complete.")
    yield
    logger.info("Application shutting down...")

databases/player_database.py:

• No changes needed - already properly configured with Base export
• Ensure Base is imported before running migrations
• The async engine configuration is compatible with Alembic's offline mode

6. Integration with FastAPI

• Add migration check on application startup in main.py (optional)
• Document migration workflow in project README
• Update compose.yaml to run migrations in containerized environments
• Set up CI/CD to run migrations automatically or with manual approval
• Ensure storage/ directory is properly configured in .gitignore for local databases

Acceptance Criteria

• Alembic is installed and initialized in the project
• alembic.ini is configured with proper database connection settings
• alembic/env.py is configured to work with async SQLAlchemy and import ORM models from schemas/player_schema.py
• Environment variable STORAGE_PATH is properly integrated with Alembic configuration
• Initial migration is created and successfully applied to development database
• Async compatibility is verified - migrations run successfully with aiosqlite driver
• CamelCase column names in player_schema.py are properly handled in migrations
• Documentation is added explaining:
  • Project's unconventional naming (schemas=ORM, models=Pydantic)
  • How to create new migrations
  • How to apply migrations
  • How to rollback migrations
  • Best practices for reviewing auto-generated migrations
  • How async SQLAlchemy works with Alembic
• Migration workflow is tested in local development environment
• Docker configuration in compose.yaml is updated to run migrations on container startup
• .gitignore is updated to exclude environment-specific Alembic files if needed
• Team is trained on basic Alembic commands and workflow

References

• Alembic Documentation
• Alembic Tutorial
• FastAPI with Alembic Guide
• SQLAlchemy Documentation

---

Follow-up Issues

After implementation, consider creating issues for:

• Setting up automated migration testing in CI/CD pipeline
• Implementing migration rollback strategy for production
• Creating migration templates for common patterns (adding indexes, constraints, etc.)

[SachinPrasanth777]

Why alembic and not prisma-client-py?

[nanotaboada]

Hi @SachinPrasanth777! That's a great question :)
Would you mind elaborating pros/cons of prisma-client-py?