DDL 内部机制

DDL 内部#

以下是一些用于生成迁移指令的构造。这里的 API 基于 sqlalchemy.schema.DDLElement 和自定义 SQL 构造和编译扩展系统。

对于 Alembic 迁移指令的编程用法，最简单的途径是使用操作指令给出的更高级别的函数。

class alembic.ddl.base.AddColumn(name: str, column: Column[Any], schema: quoted_name | str | None = None)#

class alembic.ddl.base.AlterColumn(name: str, column_name: str, schema: str | None = None, existing_type: TypeEngine | None = None, existing_nullable: bool | None = None, existing_server_default: _ServerDefault | None = None, existing_comment: str | None = None)#

类 alembic.ddl.base.AlterTable(table_name: str, schema: quoted_name | str | None = None)#

表示 ALTER TABLE 语句。

只需要表字符串名称和可选架构名称，不需要完整的表对象。

类 alembic.ddl.base.ColumnComment(name: str, column_name: str, comment: str | None, **kw)#

类 alembic.ddl.base.ColumnDefault(name: str, column_name: str, default: _ServerDefault | None, **kw)#

类 alembic.ddl.base.ColumnName(name: str, column_name: str, newname: str, **kw)#

类 alembic.ddl.base.ColumnNullable(name: str, column_name: str, nullable: bool, **kw)#

类 alembic.ddl.base.ColumnType(name: str, column_name: str, type_: TypeEngine, **kw)#

类 alembic.ddl.base.ComputedColumnDefault(name: str, column_name: str, default: Computed | None, **kw)#

类 alembic.ddl.base.DropColumn(name: str, column: Column[Any], schema: str | None = None)#

class alembic.ddl.base.IdentityColumnDefault(name: str, column_name: str, default: Identity | None, impl: DefaultImpl, **kw)#

class alembic.ddl.base.RenameTable(old_table_name: str, new_table_name: quoted_name | str, schema: quoted_name | str | None = None)#

alembic.ddl.base.add_column(compiler: DDLCompiler, column: Column[Any], **kw) → str#

alembic.ddl.base.alter_column(compiler: DDLCompiler, name: str) → str#

alembic.ddl.base.alter_table(compiler: DDLCompiler, name: str, schema: str | None) → str#

alembic.ddl.base.drop_column(compiler: DDLCompiler, name: str, **kw) → str#

alembic.ddl.base.format_column_name(compiler: DDLCompiler, name: quoted_name | str | None) → quoted_name | str#

alembic.ddl.base.format_server_default(compiler: DDLCompiler, default: _ServerDefault | None) → str#

alembic.ddl.base.format_table_name(compiler: Compiled, name: quoted_name | str, schema: quoted_name | str | None) → quoted_name | str#

alembic.ddl.base.format_type(compiler: DDLCompiler, type_: TypeEngine) → str#

alembic.ddl.base.quote_dotted(name: quoted_name | str, quote: partial) → quoted_name | str#: 引用点分隔名称的元素

alembic.ddl.base.visit_add_column(element: AddColumn, compiler: DDLCompiler, **kw) → str#

alembic.ddl.base.visit_column_default(element: ColumnDefault, compiler: DDLCompiler, **kw) → str#

alembic.ddl.base.visit_column_name(element: ColumnName, compiler: DDLCompiler, **kw) → str#

alembic.ddl.base.visit_column_nullable(element: ColumnNullable, compiler: DDLCompiler, **kw) → str#

alembic.ddl.base.visit_column_type(element: ColumnType, compiler: DDLCompiler, **kw) → str#

alembic.ddl.base.visit_computed_column(element: ComputedColumnDefault, compiler: DDLCompiler, **kw)#

alembic.ddl.base.visit_drop_column(element: DropColumn, compiler: DDLCompiler, **kw) → str#

alembic.ddl.base.visit_identity_column(element: IdentityColumnDefault, compiler: DDLCompiler, **kw)#

alembic.ddl.base.visit_rename_table(element: RenameTable, compiler: DDLCompiler, **kw) → str#

class alembic.ddl.impl.DefaultImpl(dialect: Dialect, connection: Connection | None, as_sql: bool, transactional_ddl: bool | None, output_buffer: TextIO | None, context_opts: Dict[str, Any])#

提供主要迁移操作的入口点，包括特定于数据库的行为差异。

虽然各个 SQL/DDL 结构已经提供了特定于数据库的实现，但此处的差异允许针对特定迁移执行完全不同的操作序列，例如 SQL Server 的特殊“IDENTITY INSERT”步骤用于批量插入。

add_column(table_name: str, column: Column[Any], schema: str | quoted_name | None = None) → None#

add_constraint(const: Any) → None#

adjust_reflected_dialect_options(reflected_object: Dict[str, Any], kind: str) → Dict[str, Any]#

alter_column(table_name: str, column_name: str, nullable: bool | None = None, server_default: _ServerDefault | Literal[False] = False, name: str | None = None, type_: TypeEngine | None = None, schema: str | None = None, autoincrement: bool | None = None, comment: str | Literal[False] | None = False, existing_comment: str | None = None, existing_type: TypeEngine | None = None, existing_server_default: _ServerDefault | None = None, existing_nullable: bool | None = None, existing_autoincrement: bool | None = None, **kw: Any) → None#

autogen_column_reflect(inspector, table, column_info)#

当在自动生成过程中从数据库反射表时附加到“column_reflect”事件的钩子。

方言可以选择修改此处收集的信息。

property bind: Connection | None#

bulk_insert(table: TableClause | Table, rows: List[dict], multiinsert: bool = True) → None#

cast_for_batch_migrate(existing, existing_transfer, new_type)#

command_terminator = ';'#

compare_indexes(metadata_index: Index, reflected_index: Index) → ComparisonResult#

通过比较由 create_index_sig 生成的签名来比较两个索引。

此方法返回 ComparisonResult。

compare_server_default(inspector_column, metadata_column, rendered_metadata_default, rendered_inspector_default)#

compare_type(inspector_column: Column[Any], metadata_column: Column) → bool#: 如果两个列的类型有差异，则返回 True。考虑了回溯类型和元数据类型之间的 impl.type_synonyms

compare_unique_constraint(metadata_constraint: UniqueConstraint, reflected_constraint: UniqueConstraint) → ComparisonResult#

通过比较两个签名来比较两个唯一约束。

参数是两个元组，其中包含唯一约束和由 create_unique_constraint_sig 生成的签名。

此方法返回 ComparisonResult。

correct_for_autogen_constraints(conn_uniques: Set[UniqueConstraint], conn_indexes: Set[Index], metadata_unique_constraints: Set[UniqueConstraint], metadata_indexes: Set[Index]) → None#

correct_for_autogen_foreignkeys(conn_fks: Set[ForeignKeyConstraint], metadata_fks: Set[ForeignKeyConstraint]) → None#

create_column_comment(column: ColumnElement[Any]) → None#

create_index(index: Index, **kw: Any) → None#

create_table(table: Table) → None#

create_table_comment(table: Table) → None#

drop_column(table_name: str, column: Column[Any], schema: str | None = None, **kw) → None#

drop_constraint(const: Constraint) → None#

drop_index(index: Index, **kw: Any) → None#

drop_table(table: Table) → None#

drop_table_comment(table: Table) → None#

emit_begin() → None#

在当前连接上下文中发出字符串 BEGIN，或后端特定的等效字符串。

这在离线模式下使用，通常通过 EnvironmentContext.begin_transaction()。

emit_commit() → None#

在当前连接上下文中发出字符串 COMMIT，或后端特定的等效字符串。

这在离线模式下使用，通常通过 EnvironmentContext.begin_transaction()。

execute(sql: Executable | str, execution_options: dict[str, Any] | None = None) → None#

类方法 get_by_dialect(方言: 方言) → 类型[DefaultImpl]#

标识属性忽略: 元组[str, ...] = ('order', 'on_null')#

为批处理准备表(批处理实现: ApplyBatchImpl, 表: 表) → 无#

在批处理模式下创建新表以替换表之前，执行表上所需的任何操作。

PG 方言在新的表使用相同名称之前，使用此方法删除表上的约束。

rename_table(old_table_name: str, new_table_name: str | quoted_name, schema: str | quoted_name | None = None) → None#

render_ddl_sql_expr(expr: ClauseElement, is_server_default: bool = False, **kw: Any) → str#: 渲染通常是服务器默认值、索引表达式等的 SQL 表达式。

render_type(type_obj: TypeEngine, autogen_context: AutogenContext) → str | Literal[False]#

requires_recreate_in_batch(batch_op: BatchOperationsImpl) → bool#

如果给定的 BatchOperationsImpl 需要重新创建表并进行复制才能继续，则返回 True。

通常，仅在 SQLite 中存在 add_column 以外的操作时才返回 True。

start_migrations() → None#

当调用 EnvironmentContext.run_migrations() 时调用的钩子。

实现可以在这里设置每个迁移运行状态。

static_output(text: str) → None#

transactional_ddl = False#

type_arg_extract: Sequence[str] = ()#

type_synonyms: Tuple[Set[str], ...] = ({'DECIMAL', 'NUMERIC'},)#

class alembic.ddl.impl.ImplMeta(classname: str, bases: Tuple[Type[DefaultImpl]], dict_: Dict[str, Any])#

类 alembic.ddl.impl.Params(token0, tokens, args, kwargs)#

创建 Params(token0, tokens, args, kwargs) 的新实例

args: List[str]#: 字段编号 2 的别名

kwargs: Dict[str, str]#: 字段编号 3 的别名

token0: str#: 字段编号 0 的别名

tokens: 列表[字符串]#: 字段编号 1 的别名

MySQL#

类 alembic.ddl.mysql.MariaDBImpl(方言: 方言, 连接: 连接 | 无, as_sql: 布尔值, 事务性 DDL: 布尔值 | 无, 输出缓冲区: 文本 IO | 无, 上下文选项: 字典[字符串, 任意])#

基类：MySQLImpl

备忘录: 字典#

类 alembic.ddl.mysql.MySQLAlterDefault(name: str, column_name: str, default: _ServerDefault, schema: str | None = None)#: 基类：AlterColumn

class alembic.ddl.mysql.MySQLChangeColumn(name: str, column_name: str, schema: str | None = None, newname: str | None = None, type_: TypeEngine | None = None, nullable: bool | None = None, default: _ServerDefault | Literal[False] | None = False, autoincrement: bool | None = None, comment: str | Literal[False] | None = False)#: 基类：AlterColumn

类 alembic.ddl.mysql.MySQLImpl(方言: 方言, 连接: 连接 | 无, as_sql: 布尔, 事务性 DDL: 布尔 | 无, 输出缓冲区: 文本 IO | 无, 上下文选项: 字典[字符串, 任意])#

基类：DefaultImpl

alter_column(table_name: str, column_name: str, nullable: bool | None = None, server_default: _ServerDefault | Literal[False] = False, name: str | None = None, type_: TypeEngine | None = None, schema: str | None = None, existing_type: TypeEngine | None = None, existing_server_default: _ServerDefault | None = None, existing_nullable: bool | None = None, autoincrement: bool | None = None, existing_autoincrement: bool | None = None, comment: str | Literal[False] | None = False, existing_comment: str | None = None, **kw: Any) → None#

比较服务器默认值(检查器列, 元数据列, 呈现的元数据默认值, 呈现的检查器默认值)#

correct_for_autogen_constraints(conn_unique_constraints, conn_indexes, metadata_unique_constraints, metadata_indexes)#

correct_for_autogen_foreignkeys(conn_fks, metadata_fks)#

drop_constraint(const: Constraint) → None#

memo: dict#

transactional_ddl = False#

type_arg_extract: Sequence[str] = ['character set ([\\w\\-_]+)', 'collate ([\\w\\-_]+)']#

type_synonyms: Tuple[Set[str], ...] = ({'DECIMAL', 'NUMERIC'}, {'BOOL', 'TINYINT'}, {'JSON', 'LONGTEXT'})#

class alembic.ddl.mysql.MySQLModifyColumn(name: str, column_name: str, schema: str | None = None, newname: str | None = None, type_: TypeEngine | None = None, nullable: bool | None = None, default: _ServerDefault | Literal[False] | None = False, autoincrement: bool | None = None, comment: str | Literal[False] | None = False)#: 基类：MySQLChangeColumn

MS-SQL#

class alembic.ddl.mssql.MSSQLImpl(*arg, **kw)#

基类：DefaultImpl

adjust_reflected_dialect_options(reflected_object: Dict[str, Any], kind: str) → Dict[str, Any]#

alter_column(table_name: str, column_name: str, nullable: bool | None = None, server_default: _ServerDefault | Literal[False] | None = False, name: str | None = None, type_: TypeEngine | None = None, schema: str | None = None, existing_type: TypeEngine | None = None, existing_server_default: _ServerDefault | None = None, existing_nullable: bool | None = None, **kw: Any) → None#

batch_separator = 'GO'#

bulk_insert(table: TableClause | Table, rows: List[dict], **kw: Any) → None#

compare_server_default(inspector_column, metadata_column, rendered_metadata_default, rendered_inspector_default)#

create_index(index: Index, **kw: Any) → None#

drop_column(table_name: str, column: Column[Any], schema: str | None = None, **kw) → None#

emit_begin() → None#

在当前连接上下文中发出字符串 BEGIN，或后端特定的等效字符串。

这在离线模式下使用，通常通过 EnvironmentContext.begin_transaction()。

emit_commit() → None#

在当前连接上下文中发出字符串 COMMIT，或后端特定的等效字符串。

这在离线模式下使用，通常通过 EnvironmentContext.begin_transaction()。

identity_attrs_ignore: Tuple[str, ...] = ('order', 'on_null', 'minvalue', 'maxvalue', 'nominvalue', 'nomaxvalue', 'cycle', 'cache')#

memo: dict#

transactional_ddl = True#

type_synonyms: Tuple[Set[str], ...] = ({'DECIMAL', 'NUMERIC'}, {'NVARCHAR', 'VARCHAR'})#

alembic.ddl.mssql.mssql_add_column(compiler: MSDDLCompiler, column: Column[Any], **kw) → str#

alembic.ddl.mssql.visit_add_column(element: AddColumn, compiler: MSDDLCompiler, **kw) → str#

alembic.ddl.mssql.visit_column_default(element: ColumnDefault, compiler: MSDDLCompiler, **kw) → str#

alembic.ddl.mssql.visit_column_nullable(element: ColumnNullable, compiler: MSDDLCompiler, **kw) → str#

alembic.ddl.mssql.visit_column_type(element: ColumnType, compiler: MSDDLCompiler, **kw) → str#

alembic.ddl.mssql.visit_rename_column(element: ColumnName, compiler: MSDDLCompiler, **kw) → str#

alembic.ddl.mssql.visit_rename_table(element: RenameTable, compiler: MSDDLCompiler, **kw) → str#

Postgresql#

class alembic.ddl.postgresql.CreateExcludeConstraintOp(constraint_name: sqla_compat._ConstraintName, table_name: str | quoted_name, elements: Sequence[Tuple[str, str]] | Sequence[Tuple[ColumnClause[Any], str]], where: ColumnElement[bool] | str | None = None, schema: str | None = None, _orig_constraint: ExcludeConstraint | None = None, **kw)#

基类：AddConstraintOp

表示创建排除约束操作。

classmethod batch_create_exclude_constraint(operations: BatchOperations, constraint_name: str, *elements: Any, **kw: Any) → Table | None#: 此方法通过BatchOperations.create_exclude_constraint()方法代理到BatchOperations类上。

constraint_type = 'exclude'#

classmethod create_exclude_constraint(operations: Operations, constraint_name: str, table_name: str, *elements: Any, **kw: Any) → Table | None#: 此方法在 Operations 类上通过 Operations.create_exclude_constraint() 方法进行代理。

classmethod from_constraint(constraint: ExcludeConstraint) → CreateExcludeConstraintOp#

to_constraint(migration_context: MigrationContext | None = None) → ExcludeConstraint#

class alembic.ddl.postgresql.PostgresqlColumnType(name: str, column_name: str, type_: TypeEngine, **kw)#: 基类：AlterColumn

类 alembic.ddl.postgresql.PostgresqlImpl(方言: 方言, 连接: 连接 | 无, 作为 SQL: 布尔值, 事务 DDL: 布尔值 | 无, 输出缓冲区: 文本 IO | 无, 上下文选项: 字典[字符串, 任意])#

基类：DefaultImpl

调整反射方言选项(反射选项: 字典[字符串, 任意], 类型: 字符串) → 字典[字符串, 任意]#

alter_column(table_name: str, column_name: str, nullable: bool | None = None, server_default: _ServerDefault | Literal[False] = False, name: str | None = None, type_: TypeEngine | None = None, schema: str | None = None, autoincrement: bool | None = None, existing_type: TypeEngine | None = None, existing_server_default: _ServerDefault | None = None, existing_nullable: bool | None = None, existing_autoincrement: bool | None = None, **kw: Any) → None#

autogen_column_reflect(inspector, table, column_info)#

当在自动生成过程中从数据库反射表时附加到“column_reflect”事件的钩子。

方言可以选择修改此处收集的信息。

compare_indexes(metadata_index: Index, reflected_index: Index) → ComparisonResult#

通过比较由 create_index_sig 生成的签名来比较两个索引。

此方法返回 ComparisonResult。

compare_server_default(inspector_column, metadata_column, rendered_metadata_default, rendered_inspector_default)#

compare_unique_constraint(metadata_constraint: UniqueConstraint, reflected_constraint: UniqueConstraint) → ComparisonResult#

通过比较两个签名来比较两个唯一约束。

参数是两个元组，其中包含唯一约束和由 create_unique_constraint_sig 生成的签名。

此方法返回 ComparisonResult。

correct_for_autogen_constraints(conn_unique_constraints, conn_indexes, metadata_unique_constraints, metadata_indexes)#

create_index(index: Index, **kw: Any) → None#

memo: dict#

prep_table_for_batch(batch_impl, table)#

在批处理模式下创建新表以替换表之前，执行表上所需的任何操作。

PG 方言在新的表使用相同名称之前，使用此方法删除表上的约束。

render_ddl_sql_expr(expr: ClauseElement, is_server_default: bool = False, is_index: bool = False, **kw: Any) → str#: 渲染通常是服务器默认值、索引表达式等的 SQL 表达式。

render_type(type_: TypeEngine, autogen_context: AutogenContext) → str | Literal[False]#

transactional_ddl = True#

type_synonyms: Tuple[Set[str], ...] = ({'DECIMAL', 'NUMERIC'}, {'DOUBLE PRECISION', 'FLOAT'})#

alembic.ddl.postgresql.visit_column_comment(element: ColumnComment, compiler: PGDDLCompiler, **kw) → str#

alembic.ddl.postgresql.visit_column_type(element: PostgresqlColumnType, compiler: PGDDLCompiler, **kw) → str#

alembic.ddl.postgresql.visit_identity_column(element: IdentityColumnDefault, compiler: PGDDLCompiler, **kw)#

alembic.ddl.postgresql.visit_rename_table(element: RenameTable, compiler: PGDDLCompiler, **kw) → str#

SQLite#

类 alembic.ddl.sqlite.SQLiteImpl(dialect: Dialect, connection: Connection | None, as_sql: bool, transactional_ddl: bool | None, output_buffer: TextIO | None, context_opts: Dict[str, Any])#

基类：DefaultImpl

add_constraint(const: Constraint)#

autogen_column_reflect(inspector: Inspector, table: Table, column_info: Dict[str, Any]) → None#

当在自动生成过程中从数据库反射表时附加到“column_reflect”事件的钩子。

方言可以选择修改此处收集的信息。

cast_for_batch_migrate(existing: Column[Any], existing_transfer: Dict[str, TypeEngine | Cast], new_type: TypeEngine) → None#

compare_server_default(inspector_column: Column[Any], metadata_column: Column[Any], rendered_metadata_default: str | None, rendered_inspector_default: str | None) → bool#

correct_for_autogen_constraints(conn_unique_constraints, conn_indexes, metadata_unique_constraints, metadata_indexes)#

drop_constraint(const: Constraint)#

memo: dict#

render_ddl_sql_expr(expr: ClauseElement, is_server_default: bool = False, **kw) → str#: 渲染通常是服务器默认值、索引表达式等的 SQL 表达式。

requires_recreate_in_batch(batch_op: BatchOperationsImpl) → bool#

如果给定的 BatchOperationsImpl 需要重新创建表并进行复制才能继续，则返回 True。

通常，仅在 SQLite 中存在 add_column 以外的操作时才返回 True。

transactional_ddl = False#: SQLite 支持事务 DDL，但 pysqlite 不支持：请参阅：http://bugs.python.org/issue10740

alembic.ddl.sqlite.visit_rename_table(element: RenameTable, compiler: DDLCompiler, **kw) → str#

DDL 内部机制

目录

DDL 内部#

MySQL#

MS-SQL#

Postgresql#

SQLite#