.rst

Operation Directives

Operation Directives#

Note

this section discusses the internal API of Alembic as regards the internal system of defining migration operation directives. This section is only useful for developers who wish to extend the capabilities of Alembic. For end-user guidance on Alembic migration operations, please see Operation Reference.

Within migration scripts, actual database migration operations are handled via an instance of Operations. The Operations class lists out available migration operations that are linked to a MigrationContext, which communicates instructions originated by the Operations object into SQL that is sent to a database or SQL output stream.

Most methods on the Operations class are generated dynamically using a "plugin" system, described in the next section Operation Plugins. Additionally, when Alembic migration scripts actually run, the methods on the current Operations object are proxied out to the alembic.op module, so that they are available using module-style access.

For an overview of how to use an Operations object directly in programs, as well as for reference to the standard operation methods as well as "batch" methods, see Operation Reference.

Operation Plugins#

The Operations object is extensible using a plugin system. This system allows one to add new op.<some_operation> methods at runtime. The steps to use this system are to first create a subclass of MigrateOperation, register it using the Operations.register_operation() class decorator, then build a default "implementation" function which is established using the Operations.implementation_for() decorator.

Below we illustrate a very simple operation CreateSequenceOp which will implement a new method op.create_sequence() for use in migration scripts:

fromalembic.operationsimport Operations, MigrateOperation
@Operations.register_operation("create_sequence")
classCreateSequenceOp(MigrateOperation):
"""Create a SEQUENCE."""
 def__init__(self, sequence_name, schema=None):
 self.sequence_name = sequence_name
 self.schema = schema
 @classmethod
 defcreate_sequence(cls, operations, sequence_name, **kw):
"""Issue a "CREATE SEQUENCE" instruction."""
 op = CreateSequenceOp(sequence_name, **kw)
 return operations.invoke(op)
 defreverse(self):
 # only needed to support autogenerate
 return DropSequenceOp(self.sequence_name, schema=self.schema)
@Operations.register_operation("drop_sequence")
classDropSequenceOp(MigrateOperation):
"""Drop a SEQUENCE."""
 def__init__(self, sequence_name, schema=None):
 self.sequence_name = sequence_name
 self.schema = schema
 @classmethod
 defdrop_sequence(cls, operations, sequence_name, **kw):
"""Issue a "DROP SEQUENCE" instruction."""
 op = DropSequenceOp(sequence_name, **kw)
 return operations.invoke(op)
 defreverse(self):
 # only needed to support autogenerate
 return CreateSequenceOp(self.sequence_name, schema=self.schema)

Above, the CreateSequenceOp and DropSequenceOp classes represent new operations that will be available as op.create_sequence() and op.drop_sequence(). The reason the operations are represented as stateful classes is so that an operation and a specific set of arguments can be represented generically; the state can then correspond to different kinds of operations, such as invoking the instruction against a database, or autogenerating Python code for the operation into a script.

In order to establish the migrate-script behavior of the new operations, we use the Operations.implementation_for() decorator:

@Operations.implementation_for(CreateSequenceOp)
defcreate_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)
defdrop_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)

Above, we use the simplest possible technique of invoking our DDL, which is just to call Operations.execute() with literal SQL. If this is all a custom operation needs, then this is fine. However, options for more comprehensive support include building out a custom SQL construct, as documented at Custom SQL Constructs and Compilation Extension.

With the above two steps, a migration script can now use new methods op.create_sequence() and op.drop_sequence() that will proxy to our object as a classmethod:

defupgrade():
 op.create_sequence("my_sequence")
defdowngrade():
 op.drop_sequence("my_sequence")

The registration of new operations only needs to occur in time for the env.py script to invoke MigrationContext.run_migrations(); within the module level of the env.py script is sufficient.

See also

Autogenerating Custom Operation Directives - how to add autogenerate support to custom operations.

Built-in Operation Objects#

The migration operations present on Operations are themselves delivered via operation objects that represent an operation and its arguments. All operations descend from the MigrateOperation class, and are registered with the Operations class using the Operations.register_operation() class decorator. The MigrateOperation objects also serve as the basis for how the autogenerate system renders new migration scripts.

See also

Operation Plugins

Customizing Revision Generation

The built-in operation objects are listed below.

classalembic.operations.ops.AddColumnOp(table_name:str , column:Column[Any], *, schema:str |None =None, if_not_exists:bool |None =None, inline_references:bool |None =None, inline_primary_key:bool |None =None, **kw:Any)#

Represent an add column operation.

classmethodadd_column(operations:Operations , table_name:str , column:Column[Any], *, schema:str |None =None, if_not_exists:bool |None =None, inline_references:bool |None =None, inline_primary_key:bool |None =None) → None #: This method is proxied on the Operations class, via the Operations.add_column() method.

classmethodbatch_add_column(operations:BatchOperations , column:Column[Any], *, insert_before:str |None =None, insert_after:str |None =None, if_not_exists:bool |None =None, inline_references:bool |None =None, inline_primary_key:bool |None =None) → None #: This method is proxied on the BatchOperations class, via the BatchOperations.add_column() method.

classalembic.operations.ops.AddConstraintOp#: Represent an add constraint operation.

Represent an alter column operation.

classmethodalter_column(operations:Operations , table_name:str , column_name:str , *, nullable:bool |None =None, comment:str |Literal[False]|None =False, server_default:_ServerDefaultType|None |Literal[False]=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:_ServerDefaultType|None |Literal[False]=False, existing_nullable:bool |None =None, existing_comment:str |None =None, schema:str |None =None, **kw:Any) → None #: This method is proxied on the Operations class, via the Operations.alter_column() method.

classmethodbatch_alter_column(operations:BatchOperations , column_name:str , *, nullable:bool |None =None, comment:str |Literal[False]|None =False, server_default:_ServerDefaultType|None |Literal[False]=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:_ServerDefaultType|None |Literal[False]=False, existing_nullable:bool |None =None, existing_comment:str |None =None, insert_before:str |None =None, insert_after:str |None =None, **kw:Any) → None #: This method is proxied on the BatchOperations class, via the BatchOperations.alter_column() method.

classalembic.operations.ops.AlterTableOp(table_name:str , *, schema:str |None =None)#: Represent an alter table operation.

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

Represent a bulk insert operation.

classmethodbulk_insert(operations:Operations , table:Table|TableClause, rows:List[Dict[str ,Any]], *, multiinsert:bool =True) → None #: This method is proxied on the Operations class, via the Operations.bulk_insert() method.

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

Represent a create check constraint operation.

classmethodbatch_create_check_constraint(operations:BatchOperations , constraint_name:str , condition:str |ColumnElement[bool ]|TextClause, **kw:Any) → None #: This method is proxied on the BatchOperations class, via the BatchOperations.create_check_constraint() method.

classmethodcreate_check_constraint(operations:Operations , constraint_name:str |None , table_name:str , condition:str |ColumnElement[bool ]|TextClause, *, schema:str |None =None, **kw:Any) → None #: This method is proxied on the Operations class, via the Operations.create_check_constraint() method.

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

Represent a create foreign key constraint operation.

classmethodbatch_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 #: This method is proxied on the BatchOperations class, via the BatchOperations.create_foreign_key() method.

classmethodcreate_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 #: This method is proxied on the Operations class, via the Operations.create_foreign_key() method.

Represent a create index operation.

classmethodbatch_create_index(operations:BatchOperations , index_name:str , columns:List [str ], **kw:Any ) → None #: This method is proxied on the BatchOperations class, via the BatchOperations.create_index() method.

classmethodcreate_index(operations:Operations , index_name:str |None , table_name:str , columns:Sequence[str |TextClause|ColumnElement[Any]], *, schema:str |None =None, unique:bool =False, if_not_exists:bool |None =None, **kw:Any) → None #: This method is proxied on the Operations class, via the Operations.create_index() method.

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

Represent a create primary key operation.

classmethodbatch_create_primary_key(operations:BatchOperations , constraint_name:str |None , columns:List [str ]) → None #: This method is proxied on the BatchOperations class, via the BatchOperations.create_primary_key() method.

classmethodcreate_primary_key(operations:Operations , constraint_name:str |None , table_name:str , columns:List [str ], *, schema:str |None =None) → None #: This method is proxied on the Operations class, via the Operations.create_primary_key() method.

classalembic.operations.ops.CreateTableCommentOp(table_name:str , comment:str |None , *, schema:str |None =None, existing_comment:str |None =None)#

Represent a COMMENT ON table operation.

classmethodbatch_create_table_comment(operations:BatchOperations , comment:str |None , *, existing_comment:str |None =None) → None #: This method is proxied on the BatchOperations class, via the BatchOperations.create_table_comment() method.

classmethodcreate_table_comment(operations:Operations , table_name:str , comment:str |None , *, existing_comment:str |None =None, schema:str |None =None) → None #: This method is proxied on the Operations class, via the Operations.create_table_comment() method.

reverse() → CreateTableCommentOp |DropTableCommentOp #: Reverses the COMMENT ON operation against a table.

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

Represent a create table operation.

classmethodcreate_table(operations:Operations , table_name:str , *columns:SchemaItem, if_not_exists:bool |None =None, **kw:Any) → Table#: This method is proxied on the Operations class, via the Operations.create_table() method.

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

Represent a create unique constraint operation.

classmethodbatch_create_unique_constraint(operations:BatchOperations , constraint_name:str , columns:Sequence [str ], **kw:Any ) → Any #: This method is proxied on the BatchOperations class, via the BatchOperations.create_unique_constraint() method.

classmethodcreate_unique_constraint(operations:Operations , constraint_name:str |None , table_name:str , columns:Sequence [str ], *, schema:str |None =None, **kw:Any ) → Any #: This method is proxied on the Operations class, via the Operations.create_unique_constraint() method.

classalembic.operations.ops.DowngradeOps(ops:Sequence [MigrateOperation ]=(), downgrade_token:str ='downgrades')#: contains a sequence of operations that would apply to the ‘downgrade’ stream of a script.

See also

Customizing Revision Generation

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

Represent a drop column operation.

classmethodbatch_drop_column(operations:BatchOperations , column_name:str , **kw:Any ) → None #: This method is proxied on the BatchOperations class, via the BatchOperations.drop_column() method.

classmethoddrop_column(operations:Operations , table_name:str , column_name:str , *, schema:str |None =None, **kw:Any ) → None #: This method is proxied on the Operations class, via the Operations.drop_column() method.

Represent a drop constraint operation.

classmethodbatch_drop_constraint(operations:BatchOperations , constraint_name:str , type_:str |None =None) → None #: This method is proxied on the BatchOperations class, via the BatchOperations.drop_constraint() method.

classmethoddrop_constraint(operations:Operations , constraint_name:str , table_name:str , type_:str |None =None, *, schema:str |None =None, if_exists:bool |None =None) → None #: This method is proxied on the Operations class, via the Operations.drop_constraint() method.

Represent a drop index operation.

classmethodbatch_drop_index(operations:BatchOperations , index_name:str , **kw:Any ) → None #: This method is proxied on the BatchOperations class, via the BatchOperations.drop_index() method.

classmethoddrop_index(operations:Operations , index_name:str , table_name:str |None =None, *, schema:str |None =None, if_exists:bool |None =None, **kw:Any ) → None #: This method is proxied on the Operations class, via the Operations.drop_index() method.

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

Represent an operation to remove the comment from a table.

classmethodbatch_drop_table_comment(operations:BatchOperations , *, existing_comment:str |None =None) → None #: This method is proxied on the BatchOperations class, via the BatchOperations.drop_table_comment() method.

classmethoddrop_table_comment(operations:Operations , table_name:str , *, existing_comment:str |None =None, schema:str |None =None) → None #: This method is proxied on the Operations class, via the Operations.drop_table_comment() method.

reverse() → CreateTableCommentOp #: Reverses the COMMENT ON operation against a table.

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

Represent a drop table operation.

classmethoddrop_table(operations:Operations , table_name:str , *, schema:str |None =None, if_exists:bool |None =None, **kw:Any ) → None #: This method is proxied on the Operations class, via the Operations.drop_table() method.

classalembic.operations.ops.ExecuteSQLOp(sqltext:Executable|str , *, execution_options:dict [str ,Any]|None =None)#

Represent an execute SQL operation.

classmethodbatch_execute(operations:Operations , sqltext:Executable|str , *, execution_options:dict [str ,Any]|None =None) → None #: This method is proxied on the BatchOperations class, via the BatchOperations.execute() method.

classmethodexecute(operations:Operations , sqltext:Executable|str , *, execution_options:dict [str ,Any]|None =None) → None #: This method is proxied on the Operations class, via the Operations.execute() method.

classalembic.operations.ops.MigrateOperation#

base class for migration command and organization objects.

This system is part of the operation extensibility API.

See also

Built-in Operation Objects

Operation Plugins

Customizing Revision Generation

info#: A dictionary that may be used to store arbitrary information along with this MigrateOperation object.

classalembic.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 |os.PathLike [str ]|None =None, depends_on:_RevIdType|None =None)#

represents a migration script.

E.g. when autogenerate encounters this object, this corresponds to the production of an actual script file.

A normal MigrationScript object would contain a single UpgradeOps and a single DowngradeOps directive. These are accessible via the .upgrade_ops and .downgrade_ops attributes.

In the case of an autogenerate operation that runs multiple times, such as the multiple database example in the "multidb" template, the .upgrade_ops and .downgrade_ops attributes are disabled, and instead these objects should be accessed via the .upgrade_ops_list and .downgrade_ops_list list-based attributes. These latter attributes are always available at the very least as single-element lists.

See also

Customizing Revision Generation

propertydowngrade_ops:DowngradeOps |None #: An instance of DowngradeOps.

See also

MigrationScript.downgrade_ops_list

propertydowngrade_ops_list:List [DowngradeOps ]#

A list of DowngradeOps instances.

This is used in place of the MigrationScript.downgrade_ops attribute when dealing with a revision operation that does multiple autogenerate passes.

propertyupgrade_ops:UpgradeOps |None #: An instance of UpgradeOps.

See also

MigrationScript.upgrade_ops_list

propertyupgrade_ops_list:List [UpgradeOps ]#

A list of UpgradeOps instances.

This is used in place of the MigrationScript.upgrade_ops attribute when dealing with a revision operation that does multiple autogenerate passes.

classalembic.operations.ops.ModifyTableOps(table_name:str , ops:Sequence [MigrateOperation ], *, schema:str |None =None)#: Contains a sequence of operations that all apply to a single Table.

classalembic.operations.ops.OpContainer(ops:Sequence [MigrateOperation ]=())#: Represent a sequence of operations operation.

classalembic.operations.ops.RenameTableOp(old_table_name:str , new_table_name:str , *, schema:str |None =None)#

Represent a rename table operation.

classmethodrename_table(operations:Operations , old_table_name:str , new_table_name:str , *, schema:str |None =None) → None #: This method is proxied on the Operations class, via the Operations.rename_table() method.

classalembic.operations.ops.UpgradeOps(ops:Sequence [MigrateOperation ]=(), upgrade_token:str ='upgrades')#: contains a sequence of operations that would apply to the ‘upgrade’ stream of a script.

See also

Customizing Revision Generation

Extending Existing Operations#

Added in version 1.17.2.

The Operations.implementation_for.replace parameter allows replacement of existing operation implementations, including built-in operations such as CreateTableOp. This enables customization of migration execution for purposes such as logging operations, running integrity checks, conditionally canceling operations, or adapting operations with dialect-specific options.

The example below illustrates replacing the implementation of CreateTableOp to log each table creation to a separate metadata table:

fromalembicimport op
fromalembic.operationsimport Operations
fromalembic.operations.opsimport CreateTableOp
fromalembic.operations.toimplimport create_table as _create_table
fromsqlalchemyimport MetaData, Table, Column, String
# Define a metadata table to track table operations
log_table = Table(
 "table_metadata_log",
 MetaData(),
 Column("operation", String),
 Column("table_name", String),
)
@Operations.implementation_for(CreateTableOp, replace=True)
defcreate_table_with_logging(operations, operation):
 # First, run the original CREATE TABLE implementation
 _create_table(operations, operation)
 # Then, log the operation to the metadata table
 operations.execute(
 log_table.insert().values(
 operation="create",
 table_name=operation.table_name
 )
 )

The above code can be placed in the env.py file to ensure it is loaded before migrations run. Once registered, all op.create_table() calls within migration scripts will use the augmented implementation.

The original implementation is imported from alembic.operations.toimpl and invoked within the replacement implementation. The replace parameter also enables conditional execution or complete replacement of operation behavior. The example below demonstrates skipping a CreateTableOp based on custom logic:

fromalembic.operationsimport Operations
fromalembic.operations.opsimport CreateTableOp
fromalembic.operations.toimplimport create_table as _create_table
@Operations.implementation_for(CreateTableOp, replace=True)
defcreate_table_conditional(operations, operation):
 # Check if the table should be created based on custom logic
 if should_create_table(operation.table_name):
 _create_table(operations, operation)
 else:
 # Skip creation and optionally log
 operations.execute(
 "-- Skipped creation of table %s" % operation.table_name
 )
defshould_create_table(table_name):
 # Custom logic to determine if table should be created
 # For example, check a configuration or metadata table
 return table_name not in get_ignored_tables()

Operation Directives

Contents

Operation Directives#

Operation Plugins#

Built-in Operation Objects#

Extending Existing Operations#