alembic

完成配置后，用的 alembic 数据迁移工具

安装

pip install alembic

初始化

alembi init <文件名> 可以任意起名字。 运行后，会有 alembic.init文件 与对应的文件夹

alembic init alembic

修改配置 alembic.init

[alembic]
# path to migration scripts
script_location = migrations

prepend_sys_path = .                            

# database driver and location
# 需要配置自己的数据库连接方式
sqlalchemy.url = sqlite:///local-database.db 

# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console
qualname =

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

验证配置

运行以下命令测试 alembic 是否能正确连接到数据库：

alembic current

如果配置正确，你应该看到类似以下输出：

INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.

常见问题排查

• 错误：无法连接到数据库

  • 确保数据库服务已启动。
  • 检查 sqlalchemy.url 是否正确，包括用户名、密码、主机名和端口。
  • 确保安装了正确的数据库驱动（如 psycopg2、pymysql 等）。

• 错误：缺少目标元数据

  • 如果你使用 SQLAlchemy 模型，请确保在 env.py 中正确设置了 target_metadata。例如：

  from sqlmodel import SQLModel
  target_metadata = SQLModel.metadata

自动生成迁移脚本（基于模型差异）

生成的迁移脚本在 alembic/versions 目录下。

alembic revision --autogenerate -m "描述信息"

运行迁移脚本

完成迁移脚本的创建或修改后，可以通过以下命令应用升级到最新版本：

alembic upgrade head

其他命令

回滚到指定版本

alembic downgrade <revision_id>

回滚到初始状态

alembic downgrade base

查看当前版本

alembic current

列出所有版本

alembic history

示例

在fastapi中使用alembic

修改 alembic/script.py.mako, 导入sqlmdoel

"""${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
import sqlmodel # [!!code ++]
${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 schema."""
    ${upgrades if upgrades else "pass"}


def downgrade() -> None:
    """Downgrade schema."""
    ${downgrades if downgrades else "pass"}

修改 alembic/env.py, 我这是获取自己的配置文件中的数据库连接地址

from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context

from sqlmodel import SQLModel
from app.models import *
from app.settings import settings

config = context.config
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.

setting_orm = settings.ORM
orm_conns = setting_orm.get('connections')
def mysql_db_url():
    c = orm_conns.get('mysql')
    return f"mysql+pymysql://{c.get('user')}:{c.get('password')}@{c.get('host')}:{c.get('port')}/{c.get('database')}" if c else ''

def sqlite_db_url():
    c = orm_conns.get('sqlite')
    return f"sqlite:///{c.file_path}" if c else ''

def postgresql_db_url():
    c = orm_conns.get('postgresql')
    return f"postgresql+psycopg2://{c.get('user')}:{c.get('password')}@{c.get('host')}:{c.get('port')}/{c.get('database')}？schema={c.get('？schema')}" if c else ''


# 连接类型
default_connection = setting_orm.get('default_connection')
# 获取数据库连接地址
DATABASE_URL = ''
if 'mysql' == default_connection:
    DATABASE_URL = mysql_db_url()
elif 'sqlite' == default_connection:
    DATABASE_URL = sqlite_db_url()
elif 'postgresql' == default_connection:
    DATABASE_URL = postgresql_db_url()
# 脚本修改链接地址而不是使用 alembic.ini 中的 sqlalchemy.url， 这样就可以集中配置链接地址
if DATABASE_URL:
    config.set_main_option("sqlalchemy.url", DATABASE_URL)

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = SQLModel.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    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 run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata
        )

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


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