Alembic migration patterns for SQLAlchemy 2.0 async. Use when creating database migrations, managing schema versions, handling zero-downtime deployments, or implementing reversible database changes.
View on GitHubyonatangross/skillforge-claude-plugin
ork
January 25, 2026
Select agents to install to:
npx add-skill https://github.com/yonatangross/skillforge-claude-plugin/blob/main/plugins/ork/skills/alembic-migrations/SKILL.md -a claude-code --skill alembic-migrationsInstallation paths:
.claude/skills/alembic-migrations/# Alembic Migration Patterns (2026)
Database migration management with Alembic for SQLAlchemy 2.0 async applications.
## Overview
- Creating or modifying database tables and columns
- Auto-generating migrations from SQLAlchemy models
- Implementing zero-downtime schema changes
- Rolling back or managing migration history
- Adding indexes on large production tables
- Setting up Alembic with async PostgreSQL (asyncpg)
## Quick Reference
### Initialize Alembic (Async Template)
```bash
# Initialize with async template for asyncpg
alembic init -t async migrations
# Creates:
# - alembic.ini
# - migrations/env.py (async-ready)
# - migrations/script.py.mako
# - migrations/versions/
```
### Async env.py Configuration
```python
# migrations/env.py
import asyncio
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 models' Base for autogenerate
from app.models.base import Base
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode - generates SQL."""
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="sqlal