自动生成#
注意
本节讨论了 Alembic 的内部 API，涉及 alembic revision 命令的自动生成功能。本节仅对希望扩展 Alembic 功能的开发人员有用。有关自动生成功能的一般文档，请参阅 自动生成迁移。
自动生成系统具有广泛的公共 API，包括以下领域
对
MetaData对象和数据库进行“差异”的能力，并接收数据结构作为返回。此结构可以作为基本更改列表或MigrateOperation结构提供。更改
alembic revision命令生成修订脚本的方式的能力，包括支持一次生成多个修订脚本。向自动生成添加新操作指令的能力，包括自定义模式/模型比较函数和修订脚本呈现。
获取差异#
自动生成提供的最简单的 API 是“模式比较”API；这些是简单的函数，它们将在 MetaData 对象和数据库后端之间运行所有已注册的“比较”函数，以生成一个显示它们如何不同的结构。提供的两个函数是 compare_metadata()，它更像是生成差异元组的“传统”函数，以及 produce_migrations()，它生成一个由 操作指令 中详述的操作指令组成的结构。
- alembic.autogenerate.compare_metadata(context: MigrationContext, metadata: MetaData) Any#
将数据库模式与
MetaData实例中给出的模式进行比较。数据库连接以
MigrationContext对象的上下文形式呈现，该对象提供数据库连接以及用于数据类型和服务器默认值的可选比较函数 - 有关这些函数的详细信息，请参阅EnvironmentContext.configure()中的“autogenerate”参数。返回格式是一个“diff”指令列表，每个指令表示各个差异
from alembic.migration import MigrationContext from alembic.autogenerate import compare_metadata from sqlalchemy import ( create_engine, MetaData, Column, Integer, String, Table, text, ) import pprint engine = create_engine("sqlite://") with engine.begin() as conn: conn.execute( text( ''' create table foo ( id integer not null primary key, old_data varchar, x integer ) ''' ) ) conn.execute(text("create table bar (data varchar)")) metadata = MetaData() Table( "foo", metadata, Column("id", Integer, primary_key=True), Column("data", Integer), Column("x", Integer, nullable=False), ) Table("bat", metadata, Column("info", String)) mc = MigrationContext.configure(engine.connect()) diff = compare_metadata(mc, metadata) pprint.pprint(diff, indent=2, width=20)
输出
[ ( "add_table", Table( "bat", MetaData(), Column("info", String(), table=<bat>), schema=None, ), ), ( "remove_table", Table( "bar", MetaData(), Column("data", VARCHAR(), table=<bar>), schema=None, ), ), ( "add_column", None, "foo", Column("data", Integer(), table=<foo>), ), [ ( "modify_nullable", None, "foo", "x", { "existing_comment": None, "existing_server_default": False, "existing_type": INTEGER(), }, True, False, ) ], ( "remove_column", None, "foo", Column("old_data", VARCHAR(), table=<foo>), ), ]
- 参数:
context¶ – 一个
MigrationContext实例。
另请参阅
produce_migrations()- 根据元数据比较生成MigrationScript结构。
- alembic.autogenerate.produce_migrations(context: MigrationContext, metadata: MetaData) MigrationScript#
根据架构比较生成
MigrationScript结构。此函数基本上执行
compare_metadata()的操作，但随后运行差异结果列表以生成完整的MigrationScript对象。有关此操作示例，请参阅 自定义修订版生成 中的示例。另请参阅
compare_metadata()- 从比较架构中返回更基本的“差异”数据。
自定义修订版生成#
alembic revisioncommand.revision() 以编程方式使用）在运行后本质上会生成一个单一迁移脚本。是否指定 --autogenerateupgrade()downgrade()
在任何一种情况下，系统都会以 MigrateOperation 结构的形式创建要执行操作的完整计划，然后使用该计划生成脚本。
例如，假设我们运行了 alembic revision --autogenerate'eced083f5df'
"""create the organization table."""
# revision identifiers, used by Alembic.
revision = 'eced083f5df'
down_revision = 'beafc7d709f'
from alembic import op
import sqlalchemy as sa
def upgrade():
op.create_table(
'organization',
sa.Column('id', sa.Integer(), primary_key=True),
sa.Column('name', sa.String(50), nullable=False)
)
op.add_column(
'user',
sa.Column('organization_id', sa.Integer())
)
op.create_foreign_key(
'org_fk', 'user', 'organization', ['organization_id'], ['id']
)
def downgrade():
op.drop_constraint('org_fk', 'user')
op.drop_column('user', 'organization_id')
op.drop_table('organization')
上述脚本由 MigrateOperation 结构生成，如下所示
from alembic.operations import ops
import sqlalchemy as sa
migration_script = ops.MigrationScript(
'eced083f5df',
ops.UpgradeOps(
ops=[
ops.CreateTableOp(
'organization',
[
sa.Column('id', sa.Integer(), primary_key=True),
sa.Column('name', sa.String(50), nullable=False)
]
),
ops.ModifyTableOps(
'user',
ops=[
ops.AddColumnOp(
'user',
sa.Column('organization_id', sa.Integer())
),
ops.CreateForeignKeyOp(
'org_fk', 'user', 'organization',
['organization_id'], ['id']
)
]
)
]
),
ops.DowngradeOps(
ops=[
ops.ModifyTableOps(
'user',
ops=[
ops.DropConstraintOp('org_fk', 'user'),
ops.DropColumnOp('user', 'organization_id')
]
),
ops.DropTableOp('organization')
]
),
message='create the organization table.'
)
当我们处理 MigrationScript 结构时，我们可以使用 render_python_code() 帮助函数将升级/降级部分呈现为字符串，以便进行调试
from alembic.autogenerate import render_python_code
print(render_python_code(migration_script.upgrade_ops))
呈现
### commands auto generated by Alembic - please adjust! ###
op.create_table('organization',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(length=50), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.add_column('user', sa.Column('organization_id', sa.Integer(), nullable=True))
op.create_foreign_key('org_fk', 'user', 'organization', ['organization_id'], ['id'])
### end Alembic commands ###
鉴于上述结构用于生成新的修订文件，并且我们希望在创建这些文件时能够对其进行更改，因此我们需要一个系统在使用 command.revision() 命令时访问此结构。 EnvironmentContext.configure.process_revision_directives 参数为我们提供了一种更改此结构的方法。这是一个函数，将由 Alembic 生成的上述结构作为参数传递给它，从而使我们有机会对其进行更改。例如，如果我们希望将所有“升级”操作放入某个分支，并且希望我们的脚本根本不包含任何“降级”操作，那么我们可以构建一个扩展，如下所示，在 env.py 脚本中进行说明
def process_revision_directives(context, revision, directives):
script = directives[0]
# set specific branch
script.head = "mybranch@head"
# erase downgrade operations
script.downgrade_ops.ops[:] = []
# ...
def run_migrations_online():
# ...
with engine.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata,
process_revision_directives=process_revision_directives)
with context.begin_transaction():
context.run_migrations()
在上面，directives 参数是一个 Python 列表。我们可以在此列表中就地更改给定的结构，或用一个包含零个或多个 MigrationScript 指令的新结构替换它。然后，command.revision() 命令将生成与此列表中内容相对应的脚本。
另请参阅
使用 EnvironmentContext.configure.process_revision_directives 的更多示例
- alembic.autogenerate.render_python_code(up_or_down_op: UpgradeOps | DowngradeOps, sqlalchemy_module_prefix: str = 'sa.', alembic_module_prefix: str = 'op.', render_as_batch: bool = False, imports: Sequence[str] = (), render_item: RenderItemFn | None = None, migration_context: MigrationContext | None = None, user_module_prefix: str | None = None) str#
给定
UpgradeOps或DowngradeOps对象，渲染 Python 代码。这是一个便捷函数，可用于测试用户定义的
MigrationScript结构的自动生成输出。- 参数:
up_or_down_op¶ –
UpgradeOps或DowngradeOps对象sqlalchemy_module_prefix¶ – SQLAlchemy 对象的模块前缀
alembic_module_prefix¶ – Alembic 构造的模块前缀
render_as_batch¶ – 使用“批量操作”样式进行渲染
imports¶ – 要添加的导入符号序列
render_item¶ – 渲染项目的可调用对象
migration_context¶ – 可选的
MigrationContextuser_module_prefix¶ –
用户定义类型的可选字符串前缀
在 1.11.0 版本中添加。
使用重写器进行细粒度自动生成#
前一个示例说明了我们如何对操作指令的结构进行简单的更改以生成新的自动生成输出。对于我们想要影响自动生成流的非常特定部分的情况，我们可以为 EnvironmentContext.configure.process_revision_directives 创建一个函数，该函数遍历整个 MigrationScript 结构，找到我们关心的元素，并在需要时就地修改它们。但是，为了减少与此任务相关的样板，我们可以使用 Rewriter 对象来简化此操作。 Rewriter 为我们提供了一个对象，我们可以直接将其传递给 EnvironmentContext.configure.process_revision_directives，我们还可以将处理程序函数附加到该对象，这些函数以特定类型的构造为键。
下面是一个示例，其中我们重写 ops.AddColumnOp 指令；根据新列是否“可为空”，我们返回现有指令，或者返回现有指令，其中可为空标志已更改，并将其放在一个列表中，其中包含第二个指令，以在第二步中更改可为空标志
# ... fragmented env.py script ....
from alembic.autogenerate import rewriter
from alembic.operations import ops
writer = rewriter.Rewriter()
@writer.rewrites(ops.AddColumnOp)
def add_column(context, revision, op):
if op.column.nullable:
return op
else:
op.column.nullable = True
return [
op,
ops.AlterColumnOp(
op.table_name,
op.column.name,
modify_nullable=False,
existing_type=op.column.type,
)
]
# ... later ...
def run_migrations_online():
# ...
with connectable.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata,
process_revision_directives=writer
)
with context.begin_transaction():
context.run_migrations()
在完整的 ops.MigrationScript 结构中， AddColumn 指令将出现在路径 MigrationScript->UpgradeOps->ModifyTableOps 和 MigrationScript->DowngradeOps->ModifyTableOps 中。 Rewriter 负责遍历这些结构以及根据需要重写它们，以便我们只需要为我们关心的特定对象编写代码。
- 类 alembic.autogenerate.rewriter.Rewriter#
一个帮助对象，允许轻松地“重写”操作流。
Rewriter对象旨在传递给EnvironmentContext.configure.process_revision_directives参数，该参数位于env.py脚本中。一旦构建，可以将任意数量的“重写”函数与之关联，这些函数将有机会修改结构，而无需明确了解整体结构。该函数传递了
MigrationContext对象和revision元组，这些元组通常传递给Environment Context.configure.process_revision_directives函数，第三个参数是装饰器中指出的类型的单个指令。该函数可以选择返回一个单一的 op 指令，该指令通常可以是实际传递的指令，或者是一个新指令来替换它，或者一个列表，其中包含零个或多个指令来替换它。另请参阅
使用重写器进行细粒度自动生成 - 使用示例
- chain(other: ProcessRevisionDirectiveFn | Rewriter) Rewriter#
生成此
Rewriter到另一个的“链”。这允许两个或更多重写器串行操作一个流，例如
writer1 = autogenerate.Rewriter() writer2 = autogenerate.Rewriter() @writer1.rewrites(ops.AddColumnOp) def add_column_nullable(context, revision, op): op.column.nullable = True return op @writer2.rewrites(ops.AddColumnOp) def add_column_idx(context, revision, op): idx_op = ops.CreateIndexOp( "ixc", op.table_name, [op.column.name] ) return [op, idx_op] writer = writer1.chain(writer2)
- rewrites(operator: Type[AddColumnOp] | Type[MigrateOperation] | Type[AlterColumnOp] | Type[CreateTableOp] | Type[ModifyTableOps]) Callable[..., Any]#
为给定类型注册一个函数作为重写器。
该函数应接收三个参数，它们是
MigrationContext、revision元组和指示类型的 op 指令。例如：@writer1.rewrites(ops.AddColumnOp) def add_column_nullable(context, revision, op): op.column.nullable = True return op
使用多个引擎生成修订 / run_migrations() 调用#
提供的 multidb 模板中说明了一种较少使用的技术，该技术允许自动生成的修订同时针对多个数据库后端运行，并将更改生成到单个修订脚本中。此模板具有一个特殊的 env.py，它会遍历多个 Engine 实例，并为每个实例调用 MigrationContext.run_migrations()
for name, rec in engines.items():
logger.info("Migrating database %s" % name)
context.configure(
connection=rec['connection'],
upgrade_token="%s_upgrades" % name,
downgrade_token="%s_downgrades" % name,
target_metadata=target_metadata.get(name)
)
context.run_migrations(engine_name=name)
在上面，MigrationContext.run_migrations() 运行多次，每次针对一个引擎运行。在自动生成的环境中，每次调用该方法时，upgrade_token 和 downgrade_token 参数都会更改，以便模板变量集合为每个引擎获取不同的条目，然后在 script.py.mako 中明确引用这些条目。
关于 EnvironmentContext.configure.process_revision_directives 钩子，此处的行为是 process_revision_directives 钩子被调用多次，每次调用 context.run_migrations() 一次。这意味着如果要将多 run_migrations() 方法与 process_revision_directives 钩子结合使用，则必须小心适当地使用钩子。
需要注意的第一点是，当对 run_migrations() 进行第二次调用时，.upgrade_ops 和 .downgrade_ops 属性将转换为 Python 列表，并将新的 UpgradeOps 和 DowngradeOps 对象追加到这些列表中。每个 UpgradeOps 和 DowngradeOps 对象分别维护一个 .upgrade_token 和一个 .downgrade_token 属性，用于将其内容呈现为适当的模板标记。
例如，具有引擎名称 engine1 和 engine2 的多引擎运行将在运行时生成 engine1_upgrades、engine1_downgrades、engine2_upgrades 和 engine2_downgrades 的标记。生成的迁移结构将如下所示
from alembic.operations import ops
import sqlalchemy as sa
migration_script = ops.MigrationScript(
'eced083f5df',
[
ops.UpgradeOps(
ops=[
# upgrade operations for "engine1"
],
upgrade_token="engine1_upgrades"
),
ops.UpgradeOps(
ops=[
# upgrade operations for "engine2"
],
upgrade_token="engine2_upgrades"
),
],
[
ops.DowngradeOps(
ops=[
# downgrade operations for "engine1"
],
downgrade_token="engine1_downgrades"
),
ops.DowngradeOps(
ops=[
# downgrade operations for "engine2"
],
downgrade_token="engine2_downgrades"
)
],
message='migration message'
)
鉴于上述情况，当 env.py 脚本在运行自动生成时多次调用 MigrationContext.run_migrations() 时，应考虑以下准则
如果
process_revision_directives钩子旨在根据对当前数据库/连接的检查添加元素，则应在每次迭代时执行其操作。这是为了确保每次运行钩子时数据库都可用。或者，如果
process_revision_directives钩子旨在修改迁移指令列表，则应仅在最后一次迭代中调用它。这样做是为了避免每次都向钩子提供一个不断增长的结构，而该结构之前已经修改过。如果使用
Rewriter对象，则应仅在最后一次迭代中调用它，因为它每次都会提供所有指令，因此为了避免对指令进行双重/三重/等处理，应仅在结构完成后调用它。在引用
MigrationScript.upgrade_ops_list和MigrationScript.downgrade_ops_list属性时，应参考UpgradeOps和DowngradeOps对象的集合。
自动生成自定义操作指令#
在 操作插件 一节中，我们讨论了添加 MigrateOperation 的新子类，以便添加新的 op. 指令。在前面的 自定义修订版生成 一节中，我们还了解到，这些相同的 MigrateOperation 结构是自动生成系统了解要呈现哪些 Python 代码的基础。利用此知识，我们可以创建其他函数，将其插入自动生成系统，以便在运行 alembic revision --autogenerate 时，可以将我们的新操作生成到迁移脚本中。
以下部分将详细介绍一个示例，该示例使用我们在 操作插件 中创建的 CreateSequenceOp 和 DropSequenceOp 指令，它们对应于 SQLAlchemy Sequence 构造。
使用模型跟踪我们的对象#
自动生成比较函数的基本工作是检查数据库中的一系列对象，并将它们与模型中定义的一系列对象进行比较。我们所说的“在我们的模型中”是指我们希望跟踪的 Python 代码中定义的任何内容，但最常见的是我们讨论的是一系列 Table 对象，存在于 MetaData 集合中。
我们提出一种简单的方法来查看 Sequence 对象，我们希望确保它们在自动生成运行时存在于数据库中。虽然这些对象确实与 Table 和 MetaData 有一些集成，但我们假设它们没有，因为这里的示例旨在说明我们如何对大多数任何类型的自定义构造执行此操作。我们将对象与 info 集合 MetaData 关联起来，这是一个我们可以用于任何内容的字典，我们也知道它将传递给自动生成进程
from sqlalchemy.schema import Sequence
def add_sequence_to_model(sequence, metadata):
metadata.info.setdefault("sequences", set()).add(
(sequence.schema, sequence.name)
)
my_seq = Sequence("my_sequence")
add_sequence_to_model(my_seq, model_metadata)
info 字典是一个放置我们希望自动生成例程能够找到的内容的好地方，其中可以包括任何对象，例如表示视图、触发器、特殊约束的自定义 DDL 对象，或我们希望支持的任何其他内容。
注册比较函数#
我们现在需要注册一个比较钩子，它将用于将数据库与我们的模型进行比较，并生成要包含在迁移脚本中的 CreateSequenceOp 和 DropSequenceOp 指令。请注意，我们假设后端是 Postgresql
from alembic.autogenerate import comparators
@comparators.dispatch_for("schema")
def compare_sequences(autogen_context, upgrade_ops, schemas):
all_conn_sequences = set()
for sch in schemas:
all_conn_sequences.update([
(sch, row[0]) for row in
autogen_context.connection.execute(
"SELECT relname FROM pg_class c join "
"pg_namespace n on n.oid=c.relnamespace where "
"relkind='S' and n.nspname=%(nspname)s",
# note that we consider a schema of 'None' in our
# model to be the "default" name in the PG database;
# this usually is the name 'public'
nspname=autogen_context.dialect.default_schema_name
if sch is None else sch
)
])
# get the collection of Sequence objects we're storing with
# our MetaData
metadata_sequences = autogen_context.metadata.info.setdefault(
"sequences", set())
# for new names, produce CreateSequenceOp directives
for sch, name in metadata_sequences.difference(all_conn_sequences):
upgrade_ops.ops.append(
CreateSequenceOp(name, schema=sch)
)
# for names that are going away, produce DropSequenceOp
# directives
for sch, name in all_conn_sequences.difference(metadata_sequences):
upgrade_ops.ops.append(
DropSequenceOp(name, schema=sch)
)
在上面，我们构建了一个新函数 compare_sequences()，并使用 autogenerate 将其注册为“模式”级别的比较函数。它执行的工作是将每个数据库模式中存在的序列名称列表与我们维护在 MetaData 对象中的序列名称列表进行比较。
当 autogenerate 完成时，它将在“升级”操作列表中有一系列 CreateSequenceOp 和 DropSequenceOp 指令；“降级”操作列表是使用我们已在这些对象上实现的 CreateSequenceOp.reverse() 和 DropSequenceOp.reverse() 方法直接从这些指令生成的。
在“模式”范围内注册我们的函数意味着我们的 autogenerate 比较函数在任何特定表或列的上下文之外被调用。三个可用的范围是“模式”、“表”和“列”，总结如下
模式级别 - 这些钩子传递了一个
AutogenContext、一个UpgradeOps集合和一个要操作的字符串模式名称集合。如果UpgradeOps集合在运行所有钩子后包含更改，则它将包含在迁移脚本中@comparators.dispatch_for("schema") def compare_schema_level(autogen_context, upgrade_ops, schemas): pass
表级别 - 这些钩子传递了一个
AutogenContext、一个ModifyTableOps集合、一个模式名称、表名称、一个从数据库反射的Table（如果有）或None，以及一个存在于本地MetaData中的Table。如果ModifyTableOps集合在运行所有钩子后包含更改，则它将包含在迁移脚本中@comparators.dispatch_for("table") def compare_table_level(autogen_context, modify_ops, schemaname, tablename, conn_table, metadata_table): pass
列级别 - 这些挂钩会传递一个
AutogenContext、一个AlterColumnOp对象、一个架构名称、一个表名称、一个列名称、一个从数据库中反映出来的Column以及一个存在于本地表中的Column。如果在所有挂钩都运行后AlterColumnOp包含更改，则它将包含在迁移脚本中；如果任何modify_属性被设置为非默认值，或者.kw集合中存在任何前缀为"modify_"的键，则认为存在“更改”@comparators.dispatch_for("column") def compare_column_level(autogen_context, alter_column_op, schemaname, tname, cname, conn_col, metadata_col): pass
传递给这些挂钩的 AutogenContext 如下所述。
- 类 alembic.autogenerate.api.AutogenContext(migration_context: MigrationContext, metadata: MetaData | None = None, opts: Dict[str, Any] | None = None, autogenerate: bool = True)#
维护特定于自动生成操作的配置和状态。
- connection: Connection | None = None#
Connection对象当前连接到正在比较的数据库后端。这是从
MigrationContext.bind获得的，最终在env.py脚本中设置。
- 导入: 集合[字符串] = 无#
包含字符串 Python 导入指令的
set()。指令将呈现到脚本模板的
${imports}部分。该集合通常为空，并且可以在钩子中修改，例如EnvironmentContext.configure.render_item钩子。另请参阅
- 元数据: 元数据 | 无 = 无#
表示目标的
元数据对象。此对象是
env.py中传递给EnvironmentContext.configure.target_metadata参数的对象。它表示Table和其他对象的结构，如当前数据库模型中所述，并表示正在检查的数据库的目标结构。虽然
MetaData对象主要被称为Table对象的集合，但它还有一个info字典，最终用户方案可以使用该字典来存储其他模式级对象，以便在自定义自动生成方案中进行比较。
- migration_context: MigrationContext = None#
由
env.py脚本建立的MigrationContext。
- run_filters(object_: SchemaItem, name: sqla_compat._ConstraintName, type_: NameFilterType, reflected: bool, compare_to: SchemaItem | None) bool#
运行上下文的对象过滤器，如果目标应成为自动生成操作的一部分，则返回 True。
此方法应针对自动生成操作中遇到的每种对象类型运行，从而使环境有机会筛选应包含在比较中的对象。此处的筛选器直接通过
EnvironmentContext.configure.include_object参数生成。
- run_name_filters(name: str | None, type_: NameFilterType, parent_names: NameFilterParentNames) bool#
运行上下文的名称筛选器，如果目标应成为自动生成操作的一部分，则返回 True。
此方法应针对自动生成操作的反射侧中遇到的每种名称类型运行，从而使环境有机会筛选应作为数据库对象进行反射的名称。此处的筛选器直接通过
EnvironmentContext.configure.include_name参数生成。
- run_object_filters(object_: SchemaItem, name: sqla_compat._ConstraintName, type_: NameFilterType, reflected: bool, compare_to: SchemaItem | None) bool#
运行上下文的对象过滤器，如果目标应成为自动生成操作的一部分，则返回 True。
此方法应针对自动生成操作中遇到的每种对象类型运行，从而使环境有机会筛选应包含在比较中的对象。此处的筛选器直接通过
EnvironmentContext.configure.include_object参数生成。
- sorted_tables#
返回
MetaData.sorted_tables集合的聚合。对于
MetaData对象序列，它连接每个单独MetaData中的MetaData.sorted_tables集合，按照序列的顺序。它不会整理已排序的表集合。
- table_key_to_table#
返回
MetaData.tables字典的聚合。MetaData.tables集合是表键到Table的字典；此方法将多个MetaData对象中的字典聚合为一个字典。不支持重复的表键；如果两个
MetaData对象包含相同的表键，则会引发异常。
创建渲染函数#
第二个自动生成集成挂钩是提供“渲染”函数；由于自动生成系统渲染 Python 代码，我们需要构建一个函数来为我们的指令渲染正确的“op”指令
from alembic.autogenerate import renderers
@renderers.dispatch_for(CreateSequenceOp)
def render_create_sequence(autogen_context, op):
return "op.create_sequence(%r, **%r)" % (
op.sequence_name,
{"schema": op.schema}
)
@renderers.dispatch_for(DropSequenceOp)
def render_drop_sequence(autogen_context, op):
return "op.drop_sequence(%r, **%r)" % (
op.sequence_name,
{"schema": op.schema}
)
上述函数将渲染与 CreateSequenceOp 和 DropSequenceOp 指令在我们的比较函数生成的列表中出现相对应的 Python 代码。
运行它#
所有上述代码都可以按照开发人员认为合适的方式组织；使其正常工作所需的唯一条件是当调用 Alembic 环境 env.py 时，它要么导入包含所有上述例程的模块，要么它们在本地存在，或者两者结合。
如果我们在模型中（当然，当 env.py 运行时也需要调用它！）有如下代码
from sqlalchemy.schema import Sequence
my_seq_1 = Sequence("my_sequence_1")
add_sequence_to_model(my_seq_1, target_metadata)
当我们首次运行 alembic revision --autogenerate 时，我们会在迁移文件中看到以下内容
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.create_sequence('my_sequence_1', **{'schema': None})
### end Alembic commands ###
def downgrade():
### commands auto generated by Alembic - please adjust! ###
op.drop_sequence('my_sequence_1', **{'schema': None})
### end Alembic commands ###
这些是我们自定义的指令，将在运行 alembic upgrade 或 alembic downgrade 时调用。