操作指令

内容

操作指令#

注意

本节讨论了Alembic 的内部 API，涉及定义迁移操作指令的内部系统。本节仅对希望扩展 Alembic 功能的开发人员有用。有关 Alembic 迁移操作的最终用户指南，请参阅操作参考。

在迁移脚本中，实际的数据库迁移操作通过 Operations 的实例进行处理。 Operations 类列出了可用的迁移操作，这些操作链接到 MigrationContext，后者将 Operations 对象发出的指令传达为发送到数据库或 SQL 输出流的 SQL。

Operations 类上的大多数方法都是使用“插件”系统动态生成的，该系统在下一节操作插件中进行了描述。此外，当 Alembic 迁移脚本实际运行时，当前 Operations 对象上的方法将代理到 alembic.op 模块，以便可以使用模块样式的访问方式访问它们。

有关如何在程序中直接使用 Operations 对象的概述，以及有关标准操作方法和“批处理”方法的参考，请参阅操作参考。

操作插件#

Operations 对象可使用插件系统进行扩展。此系统允许在运行时添加新的 op.<some_operation> 方法。使用此系统的步骤首先是创建 MigrateOperation 的子类，使用 Operations.register_operation() 类装饰器注册它，然后构建一个使用 Operations.implementation_for() 装饰器建立的默认“实现”函数。

下面我们说明一个非常简单的操作 CreateSequenceOp，它将在迁移脚本中实现一个新方法 op.create_sequence()

from alembic.operations import Operations, MigrateOperation

@Operations.register_operation("create_sequence")
class CreateSequenceOp(MigrateOperation):
    """Create a SEQUENCE."""

    def __init__(self, sequence_name, schema=None):
        self.sequence_name = sequence_name
        self.schema = schema

    @classmethod
    def create_sequence(cls, operations, sequence_name, **kw):
        """Issue a "CREATE SEQUENCE" instruction."""

        op = CreateSequenceOp(sequence_name, **kw)
        return operations.invoke(op)

    def reverse(self):
        # only needed to support autogenerate
        return DropSequenceOp(self.sequence_name, schema=self.schema)

@Operations.register_operation("drop_sequence")
class DropSequenceOp(MigrateOperation):
    """Drop a SEQUENCE."""

    def __init__(self, sequence_name, schema=None):
        self.sequence_name = sequence_name
        self.schema = schema

    @classmethod
    def drop_sequence(cls, operations, sequence_name, **kw):
        """Issue a "DROP SEQUENCE" instruction."""

        op = DropSequenceOp(sequence_name, **kw)
        return operations.invoke(op)

    def reverse(self):
        # only needed to support autogenerate
        return CreateSequenceOp(self.sequence_name, schema=self.schema)

在上面，CreateSequenceOp 和 DropSequenceOp 类表示将作为 op.create_sequence() 和 op.drop_sequence() 可用的新操作。将操作表示为有状态类的原因是，可以通用地表示操作和一组特定参数；然后，状态可以对应于不同类型的操作，例如针对数据库调用指令，或将操作的 Python 代码自动生成到脚本中。

为了建立新操作的迁移脚本行为，我们使用 Operations.implementation_for() 装饰器

@Operations.implementation_for(CreateSequenceOp)
def create_sequence(operations, operation):
    if operation.schema is not None:
        name = "%s.%s" % (operation.schema, operation.sequence_name)
    else:
        name = operation.sequence_name
    operations.execute("CREATE SEQUENCE %s" % name)


@Operations.implementation_for(DropSequenceOp)
def drop_sequence(operations, operation):
    if operation.schema is not None:
        name = "%s.%s" % (operation.schema, operation.sequence_name)
    else:
        name = operation.sequence_name
    operations.execute("DROP SEQUENCE %s" % name)

上面，我们使用最简单的技术来调用我们的 DDL，它只是使用带文字 SQL 的 Operations.execute() 来调用。如果这是自定义操作所需要的全部，那么这很好。然而，更全面支持的选项包括构建自定义 SQL 构造，如在自定义 SQL 构造和编译扩展中所记录。

通过上述两个步骤，迁移脚本现在可以使用新方法 op.create_sequence() 和 op.drop_sequence()，它们将作为类方法代理到我们的对象

def upgrade():
    op.create_sequence("my_sequence")

def downgrade():
    op.drop_sequence("my_sequence")

新操作的注册只需要在 env.py 脚本调用 MigrationContext.run_migrations() 时及时发生；在 env.py 脚本的模块级别就足够了。

另请参见

自动生成自定义操作指令 - 如何向自定义操作添加自动生成支持。

内置操作对象#

在 Operations 上显示的迁移操作本身是通过表示操作及其参数的操作对象传递的。所有操作都从 MigrateOperation 类派生，并使用 Operations.register_operation() 类装饰器向 Operations 类注册。 MigrateOperation 对象还作为自动生成系统如何呈现新迁移脚本的基础。

另请参见

操作插件

自定义修订版生成

内置操作对象如下所示。

类 alembic.operations.ops.AddColumnOp(table_name: str, column: Column[Any], *, schema: str | None = None, **kw: Any)#

表示添加列操作。

类方法 add_column(operations: Operations, table_name: str, column: Column[Any], *, schema: str | None = None) → None#: 此方法通过 Operations 类中的 Operations.add_column() 方法进行代理。

classmethod batch_add_column(operations: BatchOperations, column: Column[Any], *, insert_before: str | None = None, insert_after: str | None = None) → None#: 此方法通过 BatchOperations 类中的 BatchOperations.add_column() 方法进行代理。

class alembic.operations.ops.AddConstraintOp#: 表示添加约束操作。

class alembic.operations.ops.AlterColumnOp(table_name: str, column_name: str, *, schema: str | None = None, existing_type: Any | None = None, existing_server_default: Any = False, existing_nullable: bool | None = None, existing_comment: str | None = None, modify_nullable: bool | None = None, modify_comment: str | Literal[False] | None = False, modify_server_default: Any = False, modify_name: str | None = None, modify_type: Any | None = None, **kw: Any)#

表示更改列操作。

classmethod alter_column(operations: Operations, table_name: str, column_name: str, *, nullable: bool | None = None, comment: str | Literal[False] | None = False, server_default: Any = False, new_column_name: str | None = None, type_: TypeEngine[Any] | Type[TypeEngine[Any]] | None = None, existing_type: TypeEngine[Any] | Type[TypeEngine[Any]] | None = None, existing_server_default: str | bool | Identity | Computed | None = False, existing_nullable: bool | None = None, existing_comment: str | None = None, schema: str | None = None, **kw: Any) → None#: 此方法通过 Operations 类中的 Operations.alter_column() 方法进行代理。

classmethod batch_alter_column(operations: BatchOperations, column_name: str, *, nullable: bool | None = None, comment: str | Literal[False] | None = False, server_default: Any = False, new_column_name: str | None = None, type_: TypeEngine[Any] | Type[TypeEngine[Any]] | None = None, existing_type: TypeEngine[Any] | Type[TypeEngine[Any]] | None = None, existing_server_default: str | bool | Identity | Computed | None = False, existing_nullable: bool | None = None, existing_comment: str | None = None, insert_before: str | None = None, insert_after: str | None = None, **kw: Any) → None#: 此方法通过 BatchOperations 类中的 BatchOperations.alter_column() 方法进行代理。

class alembic.operations.ops.AlterTableOp(table_name: str, *, schema: str | None = None)#: 表示更改表操作。

类 alembic.operations.ops.BulkInsertOp(table: Table | TableClause, rows: List[Dict[str, Any]], *, multiinsert: bool = True)#

表示批量插入操作。

类方法 bulk_insert(operations: Operations, table: Table | TableClause, rows: List[Dict[str, Any]], *, multiinsert: bool = True) → None#: 此方法在 Operations 类中通过 Operations.bulk_insert() 方法代理。

类 alembic.operations.ops.CreateCheckConstraintOp(constraint_name: sqla_compat._ConstraintNameDefined | None, table_name: str, condition: str | TextClause | ColumnElement[Any], *, schema: str | None = None, **kw: Any)#

表示创建检查约束操作。

类方法 batch_create_check_constraint(operations: BatchOperations, constraint_name: str, condition: str | ColumnElement[bool] | TextClause, **kw: Any) → None#: 此方法在 BatchOperations 类上通过 BatchOperations.create_check_constraint() 方法进行代理。

classmethod create_check_constraint(operations: Operations, constraint_name: str | None, table_name: str, condition: str | ColumnElement[bool] | TextClause, *, schema: str | None = None, **kw: Any) → None#: 此方法在 Operations 类上通过 Operations.create_check_constraint() 方法进行代理。

class alembic.operations.ops.CreateForeignKeyOp(constraint_name: str | _NoneName | None, source_table: str, referent_table: str, local_cols: List[str], remote_cols: List[str], **kw: Any)#

表示创建外键约束操作。

classmethod batch_create_foreign_key(operations: BatchOperations, constraint_name: str | None, referent_table: str, local_cols: List[str], remote_cols: List[str], *, referent_schema: str | None = None, onupdate: str | None = None, ondelete: str | None = None, deferrable: bool | None = None, initially: str | None = None, match: str | None = None, **dialect_kw: Any) → None#: 此方法在 BatchOperations 类上通过 BatchOperations.create_foreign_key() 方法进行代理。

classmethod create_foreign_key(operations: Operations, constraint_name: str | None, source_table: str, referent_table: str, local_cols: List[str], remote_cols: List[str], *, onupdate: str | None = None, ondelete: str | None = None, deferrable: bool | None = None, initially: str | None = None, match: str | None = None, source_schema: str | None = None, referent_schema: str | None = None, **dialect_kw: Any) → None#: 此方法通过 Operations 类中的 Operations.create_foreign_key() 方法代理。

表示创建索引操作。

classmethod batch_create_index(operations: BatchOperations, index_name: str, columns: List[str], **kw: Any) → None#: 此方法通过 BatchOperations 类中的 BatchOperations.create_index() 方法代理。

classmethod create_index(operations: Operations, index_name: str | None, table_name: str, columns: Sequence[str | TextClause | Function[Any]], *, schema: str | None = None, unique: bool = False, if_not_exists: bool | None = None, **kw: Any) → None#: 此方法通过 Operations 类中的 Operations.create_index() 方法代理。

类 alembic.operations.ops.CreatePrimaryKeyOp(constraint_name: str | _NoneName | None, table_name: str, columns: Sequence[str], *, schema: str | None = None, **kw: Any)#

表示创建主键操作。

类方法 batch_create_primary_key(operations: BatchOperations, constraint_name: str | None, columns: List[str]) → None#: 此方法在 BatchOperations 类中通过 BatchOperations.create_primary_key() 方法进行代理。

classmethod create_primary_key(operations: Operations, constraint_name: str | None, table_name: str, columns: List[str], *, schema: str | None = None) → None#: 此方法在 Operations 类中通过 Operations.create_primary_key() 方法进行代理。

类方法 alembic.operations.ops.CreateTableCommentOp(table_name: str, comment: str | None, *, schema: str | None = None, existing_comment: str | None = None)#

表示 COMMENT ON table 操作。

类方法 batch_create_table_comment(operations: BatchOperations, comment: str | None, *, existing_comment: str | None = None) → None#: 此方法在 BatchOperations 类中通过 BatchOperations.create_table_comment() 方法进行代理。

classmethod create_table_comment(operations: Operations, table_name: str, comment: str | None, *, existing_comment: str | None = None, schema: str | None = None) → None#: 此方法在 Operations 类中通过 Operations.create_table_comment() 方法进行代理。

reverse() → CreateTableCommentOp | DropTableCommentOp#: 撤销针对表的 COMMENT ON 操作。

class alembic.operations.ops.CreateTableOp(table_name: str, columns: Sequence[SchemaItem], *, schema: str | None = None, _namespace_metadata: MetaData | None = None, _constraints_included: bool = False, **kw: Any)#

表示创建表操作。

classmethod create_table(operations: Operations, table_name: str, *columns: SchemaItem, **kw: Any) → Table#: 此方法在 Operations 类中通过 Operations.create_table() 方法进行代理。

class alembic.operations.ops.CreateUniqueConstraintOp(constraint_name: str | _NoneName | None, table_name: str, columns: Sequence[str], *, schema: str | None = None, **kw: Any)#

表示创建唯一约束操作。

classmethod batch_create_unique_constraint(operations: BatchOperations, constraint_name: str, columns: Sequence[str], **kw: Any) → Any#: 此方法在 BatchOperations 类中通过 BatchOperations.create_unique_constraint() 方法代理。

classmethod create_unique_constraint(operations: Operations, constraint_name: str | None, table_name: str, columns: Sequence[str], *, schema: str | None = None, **kw: Any) → Any#: 此方法在 Operations 类中通过 Operations.create_unique_constraint() 方法代理。

class alembic.operations.ops.DowngradeOps(ops: Sequence[MigrateOperation] = (), downgrade_token: str = 'downgrades')#: 包含一系列操作，这些操作将应用于脚本的“降级”流。

另请参见

自定义修订版生成

类 alembic.operations.ops.DropColumnOp(table_name: str, column_name: str, *, schema: str | None = None, _reverse: AddColumnOp | None = None, **kw: Any)#

表示删除列操作。

类方法 batch_drop_column(operations: BatchOperations, column_name: str, **kw: Any) → None#: 此方法通过 BatchOperations.drop_column() 方法代理到 BatchOperations 类上。

类方法 drop_column(operations: Operations, table_name: str, column_name: str, *, schema: str | None = None, **kw: Any) → None#: 此方法通过 Operations.drop_column() 方法在 Operations 类上进行代理。

表示删除约束操作。

类方法 batch_drop_constraint(operations: BatchOperations, constraint_name: str, type_: str | None = None) → None#: 此方法通过 BatchOperations.drop_constraint() 方法代理到 BatchOperations 类上。

类方法 drop_constraint(operations: Operations, constraint_name: str, table_name: str, type_: str | None = None, *, schema: str | None = None) → None#: 此方法代理在 Operations 类中，通过 Operations.drop_constraint() 方法。

表示删除索引操作。

类方法 batch_drop_index(operations: BatchOperations, index_name: str, **kw: Any) → None#: 此方法通过 BatchOperations.drop_index() 方法代理到 BatchOperations 类。

classmethod drop_index(operations: Operations, index_name: str, table_name: str | None = None, *, schema: str | None = None, if_exists: bool | None = None, **kw: Any) → None#: 此方法通过 Operations.drop_index() 方法代理到 Operations 类。

类 alembic.operations.ops.DropTableCommentOp(table_name: str, *, schema: str | None = None, existing_comment: str | None = None)#

表示从表中删除注释的操作。

类方法 batch_drop_table_comment(operations: BatchOperations, *, existing_comment: str | None = None) → None#: 此方法在 BatchOperations 类上通过 BatchOperations.drop_table_comment() 方法进行代理。

classmethod drop_table_comment(operations: Operations, table_name: str, *, existing_comment: str | None = None, schema: str | None = None) → None#: 此方法在 Operations 类上通过 Operations.drop_table_comment() 方法进行代理。

reverse() → CreateTableCommentOp#: 撤销针对表的 COMMENT ON 操作。

class alembic.operations.ops.DropTableOp(table_name: str, *, schema: str | None = None, table_kw: MutableMapping[Any, Any] | None = None, _reverse: CreateTableOp | None = None)#

表示删除表操作。

classmethod drop_table(operations: Operations, table_name: str, *, schema: str | None = None, **kw: Any) → None#: 此方法通过 Operations.drop_table() 方法在 Operations 类上被代理。

类 alembic.operations.ops.ExecuteSQLOp(sqltext: 可执行 | str, *, execution_options: dict[str, 任何] | 无 = 无)#

表示执行 SQL 操作。

类方法 batch_execute(operations: Operations, sqltext: Executable | str, *, execution_options: dict[str, Any] | None = None) → None#: 此方法通过 BatchOperations.execute() 方法代理到 BatchOperations 类上。

类方法 execute(operations: Operations, sqltext: Executable | str, *, execution_options: dict[str, Any] | None = None) → None#: 此方法在 Operations 类上通过 Operations.execute() 方法代理。

类 alembic.operations.ops.MigrateOperation#

迁移命令和组织对象的基类。

此系统是操作扩展性 API 的一部分。

另请参见

内置操作对象

操作插件

自定义修订版生成

info#: 一个字典，可用于存储与此 MigrateOperation 对象一起的任意信息。

class alembic.operations.ops.MigrationScript(rev_id: str | None, upgrade_ops: UpgradeOps, downgrade_ops: DowngradeOps, *, message: str | None = None, imports: Set[str] = {}, head: str | None = None, splice: bool | None = None, branch_label: _RevIdType | None = None, version_path: str | None = None, depends_on: _RevIdType | None = None)#

表示迁移脚本。

例如，当自动生成遇到此对象时，这对应于实际脚本文件的生成。

一个普通的 MigrationScript 对象将包含一个 UpgradeOps 和一个 DowngradeOps 指令。这些可以通过 .upgrade_ops 和 .downgrade_ops 属性访问。

对于多次运行的自动生成操作（例如“multidb”模板中的多个数据库示例），.upgrade_ops 和 .downgrade_ops 属性被禁用，而应通过 .upgrade_ops_list 和 .downgrade_ops_list 列表属性来访问这些对象。这些后一个属性至少始终可用作单元素列表。

另请参见

自定义修订版生成

property downgrade_ops: DowngradeOps | None#: DowngradeOps 的实例。

另请参见

MigrationScript.downgrade_ops_list

property downgrade_ops_list: List[DowngradeOps]#

DowngradeOps 实例的列表。

在处理执行多个自动生成传递的修订操作时，这用于代替 MigrationScript.downgrade_ops 属性。

property upgrade_ops: UpgradeOps | None#: UpgradeOps 的实例。

另请参见

MigrationScript.upgrade_ops_list

属性 upgrade_ops_list: 列表[UpgradeOps]#

UpgradeOps 实例的列表。

在处理执行多个自动生成传递的修订操作时，这将用于 MigrationScript.upgrade_ops 属性中。

类 alembic.operations.ops.ModifyTableOps(table_name: str, ops: 序列[MigrateOperation], *, schema: str | 无 = 无)#: 包含一系列操作，所有这些操作都应用于单个表。

类 alembic.operations.ops.OpContainer(ops: 序列[MigrateOperation] = ())#: 表示一系列操作操作。

类 alembic.operations.ops.RenameTableOp(old_table_name: 字符串, new_table_name: 字符串, *, schema: 字符串 | 无 = 无)#

表示重命名表操作。

类方法 rename_table(operations: Operations, old_table_name: str, new_table_name: str, *, schema: str | None = None) → None#: 此方法通过 Operations.rename_table() 方法，在 Operations 类上进行代理。

类 alembic.operations.ops.UpgradeOps(ops: Sequence[MigrateOperation] = (), upgrade_token: str = 'upgrades')#: 包含一系列操作，这些操作将应用于脚本的“升级”流。

另请参见

自定义修订版生成