Alembic 是 SQLAlchemy 官方推荐的数据库迁移工具，它旨在帮助开发者对数据库 Schema 进行版本控制，并与 SQLAlchemy ORM 定义的模型（Model）保持同步。你可以将其理解为一个针对数据库模式的"Git"，通过生成和应用迁移脚本来跟踪和管理数据库结构的变化。

安装 Alembic

首先，在您的项目虚拟环境中安装 Alembic：

pip install alembic

初始化项目

在您的项目根目录下执行初始化命令，这将创建一个 Alembic 工作目录及配置文件：

# 确保您在项目根目录
cd your_project_root
# 执行初始化
alembic init alembic_env

执行后，项目目录中会新增一个名为 alembic_env 的文件夹（您可以自定义此名称，这里使用 alembic_env 作为示例）和一个 alembic.ini 配置文件。

配置数据库连接

alembic.ini 文件是 Alembic 的核心配置文件。您需要修改其中的 sqlalchemy.url 配置项，将其设置为您的 SQLAlchemy 数据库连接字符串。

# alembic.ini
# ... 其他配置 ...
sqlalchemy.url = mysql+pymysql://user:password@host:port/database_name

注意： 数据库驱动的选择可能影响兼容性。例如，某些异步驱动（如 aiomysql）可能不被 Alembic 的同步操作直接支持，建议使用对应的同步驱动（如 pymysql）。

定义数据模型

在您的项目中创建 SQLAlchemy 模型。例如，我们可以在 app_models/base.py 中定义基础声明式模型，并在 app_models/customer_model.py 中定义一个客户模型：

app_models/base.py:

from sqlalchemy.orm import declarative_base

# 声明基础类，所有模型都将继承自此
AppBase = declarative_base()

app_models/customer_model.py:

from sqlalchemy import Column, Integer, String, Text
from .base import AppBase # 从同一包中导入AppBase

class Customer(AppBase):
    __tablename__ = 'customers' # 数据库中的表名
    customer_id = Column(Integer, primary_key=True, autoincrement=True)
    full_name = Column(String(100), nullable=False)
    email_address = Column(String(150), unique=True)
    address = Column(Text, nullable=True)

关联模型与 Alembic

为了让 Alembic 能够"发现"并跟踪您的 SQLAlchemy 模型，您需要修改 alembic_env/env.py 文件。找到 target_metadata 变量，将其从 None 修改为您的模型基础类的 .metadata 属性。

# alembic_env/env.py
# ... 导入部分 ...

# 添加您的模型MetaData对象，以便'autogenerate'支持
# 假设您的模型基础类定义在 'app_models.base' 模块中
from app_models.base import AppBase
target_metadata = AppBase.metadata

# ... 其他 env.py 内容 ...

这一步至关重要，它建立了 Alembic 和您数据模型之间的连接，使得 Alembic 能够自动检测模型类的变化。

执行首次数据库迁移

在模型定义和 Alembic 配置完成后，我们可以执行首次迁移来创建数据库表。

1. 生成迁移脚本

运行以下命令生成一个新的迁移文件：

alembic revision --autogenerate -m 'create initial customer table'

您会看到类似如下的输出，表明 Alembic 检测到了新添加的 customers 表：

INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'customers'
  Generating /path/to/your_project/alembic_env/versions/xxxxxxxxxxxx_create_initial_customer_table.py ...  done

这会在 alembic_env/versions/ 目录下生成一个 Python 脚本文件，其中包含了创建 customers 表的 SQL 语句。

2. 应用迁移到数据库

使用 upgrade head 命令将生成的迁移脚本应用到数据库：

alembic upgrade head

输出将显示迁移正在执行：

INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> xxxxxxxxxxxx, create initial customer table

命令成功执行后，您的数据库中将创建两张表：一张是 customers 表（对应您的模型），另一张是 alembic_version 表，Alembic 用它来记录当前的数据库模式版本。

处理多个模型文件

当项目规模扩大时，您通常会将不同模块的模型定义在不同的文件中。例如，除了 Customer 模型，可能还有 Order 和 Product 模型。

app_models/order_model.py:

from sqlalchemy import Column, Integer, DECIMAL, DateTime, String
from datetime import datetime
from .base import AppBase

class ProductOrder(AppBase):
    __tablename__ = 'product_orders'
    order_id = Column(Integer, primary_key=True, autoincrement=True)
    customer_ref_id = Column(Integer, nullable=False) # 假设关联Customer的ID
    order_timestamp = Column(DateTime, default=datetime.utcnow)
    total_amount = Column(DECIMAL(10, 2), nullable=False)
    status_code = Column(String(20), default='pending')

app_models/product_model.py:

from sqlalchemy import Column, Integer, String, Float, Text
from .base import AppBase

class StoreProduct(AppBase):
    __tablename__ = 'store_products'
    product_code = Column(Integer, primary_key=True, autoincrement=True)
    product_name = Column(String(100), nullable=False, unique=True)
    unit_price = Column(Float, nullable=False)
    description = Column(Text, nullable=True)
    current_stock = Column(Integer, default=0)

为了让 Alembic 能够检测到所有这些模型，您需要在 app_models/base.py 文件中导入这些新的模型文件，确保它们的元数据被 AppBase.metadata 收集：

app_models/base.py:

from sqlalchemy.orm import declarative_base

AppBase = declarative_base()

# 导入所有模型文件，确保AppBase.metadata能够发现它们
from . import customer_model
from . import order_model
from . import product_model

再次执行迁移

完成上述修改后，再次执行迁移步骤：

# 1. 生成新的迁移脚本，Alembic 会检测到新增的 product_orders 和 store_products 表
alembic revision --autogenerate -m 'add order and product tables'

# 2. 应用迁移脚本到数据库
alembic upgrade head

Alembic 将自动生成创建这些新表的 SQL 语句，并将其应用到数据库中。