在本文中，我将简要介绍一些最佳实践，这些实践能够帮助项目保持条理，简化数据库管理，并避免在使用Alembic和SQLAlchemy时常见的错误。这些技巧曾多次帮我避免了麻烦。本文将涵盖以下内容：
- 命名规范
- 按日期排序迁移
- 表、列和迁移注释说明
- 在没有模型的情况下处理迁移中的数据
- 迁移测试（梯度测试）
- 运行迁移的服务程序
- 使用混入（Mixins）扩展模型
Dall-E 生成的图片
1. 命名规范SQLAlchemy 允许你设置这样的命名约定，这样的命名约定在生成迁移时会自动应用于所有表和约束。这可以节省手动命名索引、外键和其他约束的时间，从而使数据库结构更加可预测和一致化。
要在新项目中设置这个，在基类中添加一个约定，这样Alembic就会自动使用所需的命名格式。这里有一个常用的约定示例，适用于大多数情况：
从 sqlalchemy 和 sqlalchemy.orm 导入 MetaData 和 DeclarativeBase
convention = {
'all_column_names': lambda constraint, table: '_'.join(
[column.name for column in constraint.columns.values()]
), # 'all_column_names' 函数用于生成由列名组成的字符串
'ix': 'ix__%(table_name)s__%(all_column_names)s',
'uq': 'uq__%(table_name)s__%(all_column_names)s',
'ck': 'ck__%(table_name)s__%(constraint_name)s',
'fk': 'fk__%(table_name)s__%(all_column_names)s__%(referred_table_name)s',
'pk': 'pk__%(table_name)s',
}
# 定义一个基类 BaseModel，继承自 DeclarativeBase
class BaseModel(DeclarativeBase):
metadata = MetaData(naming_convention=convention)阿勒比迁移文件名通常以修订标签开头，这可能会让迁移的顺序看起来是随机的。有时按时间顺序排列它们会更有帮助。
Alembic 允许在 alembic.ini 文件中的 file_template 设置中自定义迁移文件名的模板。以下是两种方便的命名格式，用于保持迁移文件名的有序性，：
- 按照日期排序：
file_template = %%(年)d-%%(月)02d-%%(日)02d_%%(rev)s_%%(slug)s2. 基于 Unix 时间戳的方式：
_file_template = %%(epoch)d_%%(rev)s_%%(slug)s # _file_template = 时代_修订版_简短描述在文件名中使用日期或Unix时间戳可以保持迁移有序化，这样一来，导航更轻松。我更喜欢使用Unix时间戳，接下来会给出一个例子。
3 表和迁移的备注对于团队工作的成员来说，给属性添加注释是一个好习惯。在使用 SQLAlchemy 模型时，建议直接在列和表上添加注释，而不是依赖于文档字符串注释。这样，注释既可以在代码中，也可以在数据库中看到，这使得数据库管理员或分析师更容易理解表和字段的用途。
class Event(BaseModel):
__table_args__ = {'comment': '系统服务事件'}
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
primary_key=True,
comment='事件的主键ID',
)
service_id: Mapped[int] = mapped_column(
sa.Integer,
sa.ForeignKey(
f'{IntegrationServiceModel.__tablename__}.id',
ondelete='CASCADE',
),
nullable=False,
comment='指向拥有该事件的集成服务的外键ID',
)
name: Mapped[str] = mapped_column(
sa.String(256), nullable=False, comment='长度为256的字符串'
)添加注释到迁移中也可以帮助我们在文件系统中更容易地找到这些迁移。添加注释时可以使用 -m <注释>。注释会出现在文档字符串和文件名里。这样做使得查找所需的迁移就会变得容易得多。
1728372261_c0a05e0cd317_add_integration_service.py # 添加集成服务
1728372272_a1b4c9df789d_add_user.py # 添加用户
1728372283_f32d57aa1234_update_order_status.py # 更新订单状态
1728372294_9c8e7ab45e11_create_payment.py # 创建支付
1728372305_bef657cd9342_remove_old_column_from_users.py # 移除用户表中的旧列模型经常被用来进行数据操作，例如将数据从一张表转移到另一张表或修改列的值。然而，在迁移过程中使用ORM模型可能会带来问题，如果在创建迁移之后模型发生了变化。在这种情况下，基于旧模型的迁移在执行时可能会失败，因为数据库模式可能不再与当前模型匹配。
迁移应该独立于当前的模型状态，保持静态，以确保无论代码如何更改都能正确执行。以下有两种方法来避免使用模型来进行数据处理。
- 直接使用SQL进行数据处理:
或
- 直接使用SQL操作数据:
def 更新电子邮件():
# 更新用户的电子邮件设置
op.execute(
"UPDATE user_account SET email = CONCAT(username, '@example.com') WHERE email IS NULL;"
)
def 回退电子邮件设置():
# 回退用户的电子邮件设置
op.execute(
"UPDATE user_account SET email = NULL WHERE email LIKE '%@example.com';"
)- 直接在迁移中定义表： 如果你想使用SQLAlchemy来进行数据管理，可以直接在迁移中手动定义表。这确保了表结构在迁移执行时是固定的，并且不会受到模型变化的影响。
from sqlalchemy import table, column, String
def upgrade():
# 定义 user_account 表以处理用户数据
user_account = table(
'user_account',
column('id'),
column('username', String),
column('email', String)
)
# 获取数据库链接
conn = op.get_bind()
# 选择所有没有电子邮件的用户记录
users = conn.execute(
user_account.select().where(user_account.c.email == None)
)
# 更新每个用户的电子邮件地址
for user in users:
conn.execute(
user_account.update().where(
user_account.c.id == user.id
).values(
email=f"{user.username}@example.com"
)
)
def downgrade():
user_account = table(
'user_account',
column('id'),
column('email', String)
)
conn = op.get_bind()
# 移除在升级过程中添加的带有 '@example.com' 的电子邮件
conn.execute(
user_account.update().where(
user_account.c.email.like('%@example.com')
).values(email=None)
)楼梯测试（Stairway Test）涉及逐步测试 升级/降级 迁移，以确保整个迁移链工作正常。这确保了每个迁移都能从头开始成功创建新的数据库并降级而不会出现问题。将此测试加入 CI 对团队而言非常有价值，能节省时间并减少挫败感。
楼梯测试的步骤顺序
将测试集成到您的项目中非常简单快捷。您可以在这个代码库中找到代码示例。它还包含其他有价值的迁移测试，这些测试可能也会对你有帮助。
6. 移民服务一个专门的服务用于执行迁移。这只是执行迁移的一种方式。这种方法在本地开发环境或类似环境中非常适合。这里还有一个要点需要注意：条件[depends_on](https://docs.docker.com/compose/how-tos/startup-order/) 特性。我们将包含 Alembic 的应用镜像放在一个单独的容器中运行。我们还添加了对数据库的依赖条件，即迁移操作仅在数据库准备好 (service_healthy) 处理请求时才开始。此外，还可以为应用添加一个条件 depends_on (service_completed_successfully)，确保它仅在迁移顺利完成 (service_completed_successfully) 后才开始运行。
db:
image: postgres:15
...
健康检查:
test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER} -d ${POSTGRES_DB}"]
interval: 10s
start_period: 10s
app_migrations:
image: <app-image>
command: [
"python",
"-m",
"alembic",
"-c",
"<path>/alembic.ini",
"upgrade",
"head"
]
依赖于:
db:
条件: 服务健康
app:
...
依赖于:
app_migrations:
条件: 服务成功完成depends_on 条件确保数据库迁移仅在数据库完全准备好之后运行，并且应用程序在所有迁移完成后启动。
虽然这一点显而易见，但还是不要忽视。使用混入是一种避免代码重复的好方法。混入是包含常用字段和方法的类，可以在任何需要它们的地方集成到模型中。比如说，我们经常需要 created_at 和 updated_at 字段来追踪记录的创建和更新时间。使用基于 UUID 的 id 来标准化主键也很方便。所有这些都可以通过混入来实现。
import uuid
from sqlalchemy import Column, DateTime, func
from sqlalchemy.dialects.postgresql import UUID
class TimestampMixin:
created_at = Column(
DateTime,
server_default=func.now(),
nullable=False,
comment="记录创建时间"
)
updated_at = Column(
DateTime,
onupdate=func.now(),
nullable=True,
comment="最后更新时间"
)
class UUIDPrimaryKeyMixin:
id = Column(
UUID(as_uuid=True),
primary_key=True,
default=uuid.uuid4,
comment="记录的唯一标识"
)通过添加这些混合类（mixins），我们可以在任何需要 UUID id 和时间戳的模型中包含这些功能。
class User(UUIDPrimaryKeyMixin, TimestampMixin, BaseModel):
__tablename__ = 'user'
# 其他字段...处理迁移可能是有挑战的，但遵循这些简单的做法可以帮助项目保持有序，并更容易管理。命名规范、按日期排序、注释和测试曾帮我避免了混乱并预防了错误。希望这篇文章对你有帮助——欢迎在评论区分享你的迁移技巧！
共同學習，寫下你的評論
評論加載中...
作者其他優質文章
