Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
Configure Alembic for async SQLAlchemy migrations with PostgreSQL
FastAPI Alembic Setup
Overview
This skill covers setting up Alembic for database migrations with async SQLAlchemy 2.0 and PostgreSQL.
Initialize Alembic
# Initialize alembic in the project root
uv run alembic init alembic
Create alembic.ini
Create/update alembic.ini in project root:
# Alembic Configuration[alembic]# Path to migration scriptsscript_location = alembic
# Template used to generate migration filesfile_template = %%(year)d%%(month).2d%%(day).2d_%%(hour).2d%%(minute).2d%%(second).2d_%%(slug)s
# Set to 'true' to run the environment during# the 'revision' command, regardless of autogenerate# revision_environment = false# Truncate table name in version filestruncate_slug_length = 40# Set to 'true' to allow .pyc and .pyo files without# having their source .py files alongside them# prepend_sys_path = .# Timezone for file generationtimezone = UTC
= root,sqlalchemy,alembic
= console
= generic
= WARNING
= console
=
= WARNING
=
= sqlalchemy.engine
= INFO
=
= alembic
= StreamHandler
= (sys.stderr,)
= NOTSET
= generic
= %(levelname)-s [%(name)s] %(message)s
= %H:%M:%S
# Logging configuration
[loggers]
keys
[handlers]
keys
[formatters]
keys
[logger_root]
level
handlers
qualname
[logger_sqlalchemy]
level
handlers
qualname
[logger_alembic]
level
handlers
qualname
[handler_console]
class
args
level
formatter
[formatter_generic]
format
5.5
datefmt
Create alembic/env.py
Replace alembic/env.py with:
import asyncio
from logging.config import fileConfig
from alembic import context
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
from app.config import settings
from app.core.models import Base
# Import all models here to ensure they're registered with Base.metadata# This is crucial for autogenerate to detect all tables# from app.items.models import Item# from app.users.models import User# Alembic Config object
config = context.config
# Set the database URL from settings
config.set_main_option("sqlalchemy.url", str(settings.database_url))
# Interpret the config file for Python loggingif config.config_file_name isnotNone:
fileConfig(config.config_file_name)
# Target metadata for autogenerate support
target_metadata = Base.metadata
defrun_migrations_offline() -> None:
"""
Run migrations in 'offline' mode.
This configures the context with just a URL and not an Engine,
allowing us to generate SQL scripts without a live database.
Usage: alembic upgrade head --sql
"""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
compare_type=True,
compare_server_default=True,
)
with context.begin_transaction():
context.run_migrations()
defdo_run_migrations(connection: Connection) -> None:
"""
Run migrations with the given connection.
"""
context.configure(
connection=connection,
target_metadata=target_metadata,
compare_type=True,
compare_server_default=True,
)
with context.begin_transaction():
context.run_migrations()
asyncdefrun_async_migrations() -> None:
"""
Run migrations in async mode.
Creates an async engine and runs migrations within a connection.
"""
connectable = async_engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
asyncwith connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
await connectable.dispose()
defrun_migrations_online() -> None:
"""
Run migrations in 'online' mode.
Uses asyncio to run migrations with a live database connection.
"""
asyncio.run(run_async_migrations())
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
Create alembic/script.py.mako
Create/update alembic/script.py.mako:
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision: str = ${repr(up_revision)}
down_revision: Union[str, None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
def upgrade() -> None:
"""Upgrade database schema."""
${upgrades if upgrades else "pass"}
def downgrade() -> None:
"""Downgrade database schema."""
${downgrades if downgrades else "pass"}
Create alembic/versions/.gitkeep
touch alembic/versions/.gitkeep
Important: Register Models
In alembic/env.py, you MUST import all models for autogenerate to work:
# Import all models to register them with Base.metadatafrom app.items.models import Item # noqa: F401from app.users.models import User # noqa: F401from app.orders.models import Order # noqa: F401
Alembic Commands
Create a New Migration
# Autogenerate migration from model changes
uv run alembic revision --autogenerate -m "add items table"# Create empty migration
uv run alembic revision -m "add custom index"
Run Migrations
# Upgrade to latest
uv run alembic upgrade head# Upgrade one step
uv run alembic upgrade +1
# Upgrade to specific revision
uv run alembic upgrade abc123
# Generate SQL without running (offline mode)
uv run alembic upgrade head --sql
Downgrade
# Downgrade one step
uv run alembic downgrade -1
# Downgrade to specific revision
uv run alembic downgrade abc123
# Downgrade to nothing (empty database)
uv run alembic downgrade base
Check Status
# Show current revision
uv run alembic current
# Show migration history
uv run alembic history# Show pending migrations
uv run alembic history --indicate-current
Migration Best Practices
1. Review Autogenerated Migrations
Always review autogenerated migrations before running:
# Check that it detected all changes# Add any missing indexes or constraints# Verify the downgrade path works
2. Add Data Migrations When Needed
defupgrade() -> None:
# Schema change
op.add_column("items", sa.Column("status", sa.String(50), nullable=True))
# Data migration
op.execute("UPDATE items SET status = 'active' WHERE status IS NULL")
# Make column non-nullable after data migration
op.alter_column("items", "status", nullable=False)
defdowngrade() -> None:
op.drop_column("items", "status")
3. Use Batch Operations for SQLite Compatibility
# If you need SQLite support (e.g., for tests)with op.batch_alter_table("items") as batch_op:
batch_op.add_column(sa.Column("new_col", sa.String(50)))
4. Handle Enum Types
from sqlalchemy.dialects import postgresql
defupgrade() -> None:
# Create enum type
status_enum = postgresql.ENUM("active", "inactive", "deleted", name="item_status")
status_enum.create(op.get_bind())
# Use enum in column
op.add_column(
"items",
sa.Column("status", status_enum, nullable=False, server_default="active"),
)
defdowngrade() -> None:
op.drop_column("items", "status")
# Drop enum type
status_enum = postgresql.ENUM("active", "inactive", "deleted", name="item_status")
status_enum.drop(op.get_bind())
