Release: 1.4.25 | Release Date: September 22, 2021

SQLAlchemy 1.4 Documentation

自定义DDL

在前面的章节中,我们讨论了各种模式构造,包括 TableForeignKeyConstraintCheckConstraintSequence . 一直以来,我们都依赖 create()create_all() 方法 TableMetaData 以便为所有构造发出数据定义语言(DDL)。当发出命令时,将调用预先确定的操作顺序,并且无条件创建每个表的DDL,包括所有约束和与之关联的其他对象。对于需要特定于数据库的DDL的更复杂的场景,SQLAlchemy提供了两种技术,可用于根据任何条件添加任何DDL,要么与标准生成的表一起使用,要么单独使用。

自定义DDL

自定义DDL短语最容易使用 DDL 构建。此构造与所有其他DDL元素一样工作,但它接受要发出的文本字符串:

event.listen(
    metadata,
    "after_create",
    DDL("ALTER TABLE users ADD CONSTRAINT "
        "cst_user_name_length "
        " CHECK (length(user_name) >= 8)")
)

创建DDL构造库的一个更全面的方法是使用自定义编译-请参见 自定义SQL构造和编译扩展 有关详细信息。

控制DDL序列

这个 DDL 前面介绍的构造还可以根据对数据库的检查有条件地调用。此功能可使用 DDLElement.execute_if() 方法。例如,如果我们只想在PostgreSQL后端上创建一个触发器,我们可以将其调用为:

mytable = Table(
    'mytable', metadata,
    Column('id', Integer, primary_key=True),
    Column('data', String(50))
)

func = DDL(
    "CREATE FUNCTION my_func() "
    "RETURNS TRIGGER AS $$ "
    "BEGIN "
    "NEW.data := 'ins'; "
    "RETURN NEW; "
    "END; $$ LANGUAGE PLPGSQL"
)

trigger = DDL(
    "CREATE TRIGGER dt_ins BEFORE INSERT ON mytable "
    "FOR EACH ROW EXECUTE PROCEDURE my_func();"
)

event.listen(
    mytable,
    'after_create',
    func.execute_if(dialect='postgresql')
)

event.listen(
    mytable,
    'after_create',
    trigger.execute_if(dialect='postgresql')
)

这个 DDLElement.execute_if.dialect 关键字还接受字符串方言名称的元组::

event.listen(
    mytable,
    "after_create",
    trigger.execute_if(dialect=('postgresql', 'mysql'))
)
event.listen(
    mytable,
    "before_drop",
    trigger.execute_if(dialect=('postgresql', 'mysql'))
)

这个 DDLElement.execute_if() 方法还可以针对将接收正在使用的数据库连接的可调用函数工作。在下面的示例中,我们使用它有条件地创建一个检查约束,首先在PostgreSQL目录中查找它是否存在:

def should_create(ddl, target, connection, **kw):
    row = connection.execute(
        "select conname from pg_constraint where conname='%s'" %
        ddl.element.name).scalar()
    return not bool(row)

def should_drop(ddl, target, connection, **kw):
    return not should_create(ddl, target, connection, **kw)

event.listen(
    users,
    "after_create",
    DDL(
        "ALTER TABLE users ADD CONSTRAINT "
        "cst_user_name_length CHECK (length(user_name) >= 8)"
    ).execute_if(callable_=should_create)
)
event.listen(
    users,
    "before_drop",
    DDL(
        "ALTER TABLE users DROP CONSTRAINT cst_user_name_length"
    ).execute_if(callable_=should_drop)
)

sqlusers.create(engine)

sqlusers.drop(engine)

使用内置的ddlElement类

这个 sqlalchemy.schema 包包含提供DDL表达式的SQL表达式构造。例如,生成 CREATE TABLE 声明:

from sqlalchemy.schema import CreateTable
with engine.connect() as conn:
sql    conn.execute(CreateTable(mytable))

上面, CreateTable 与任何其他表达式构造(例如 select()table.insert() 等)。所有SQLAlchemy的面向DDL的构造都是 DDLElement 基类;这是与创建、删除和更改相对应的所有对象的基础,不仅在SQLAlchemy中,而且在Alembic迁移中也是如此。可用构造的完整引用位于 DDL表达式构造API .

用户定义的DDL构造也可以创建为 DDLElement 本身。文件 自定义SQL构造和编译扩展 有几个这样的例子。

上一节中描述的事件驱动的DDL系统 控制DDL序列 可与其他人一起使用 DDLElement 对象也一样。但是,在处理诸如 CreateIndexCreateSequence 等,事件系统 有限的 使用,如 Table.create()MetaData.create_all() 将无条件地调用这些构造。在未来的SQLAlchemy版本中,包含条件执行的DDL事件系统将考虑当前在所有情况下调用的内置构造。

我们可以用 AddConstraintDropConstraint 构造,因为事件驱动系统将用于检查和唯一约束,使用这些方法就像我们在前面的示例中所做的那样 DDLElement.execute_if()

def should_create(ddl, target, connection, **kw):
    row = connection.execute(
        "select conname from pg_constraint where conname='%s'" %
        ddl.element.name).scalar()
    return not bool(row)

def should_drop(ddl, target, connection, **kw):
    return not should_create(ddl, target, connection, **kw)

event.listen(
    users,
    "after_create",
    AddConstraint(constraint).execute_if(callable_=should_create)
)
event.listen(
    users,
    "before_drop",
    DropConstraint(constraint).execute_if(callable_=should_drop)
)

sqlusers.create(engine)

sqlusers.drop(engine)

上面的示例与内置的 AddConstraintDropConstraint 对象,DDL事件目前的主要用途仍然集中在 DDL 构造自身以及用户定义的子类 DDLElement 这还不是 MetaData.create_all()Table.create() 以及相应的“丢弃”过程。

DDL表达式构造API

Object Name Description

_CreateDropBase

表示创建和删除或等效项的DDL构造的基类。

AddConstraint

表示alter table add约束语句。

CreateColumn

代表一个 Column 在create table语句中,通过 CreateTable 构造。

CreateIndex

表示创建索引语句。

CreateSchema

表示创建架构语句。

CreateSequence

表示创建序列语句。

CreateTable

表示create table语句。

DDL

字面DDL语句。

DDLElement

DDL表达式构造的基类。

DropConstraint

表示alter table drop约束语句。

DropIndex

表示一个DROP INDEX语句。

DropSchema

表示一个DROP SCHEMA语句。

DropSequence

表示一个DROP SEQUENCE语句。

DropTable

表示Drop Table语句。

sort_tables(tables[, skip_fn, extra_dependencies])

收集 Table 基于依赖关系的对象。

sort_tables_and_constraints(tables[, filter_fn, extra_dependencies, _warn_for_cycles])

收集 Table / ForeignKeyConstraint 物体。

function sqlalchemy.schema.sort_tables(tables, skip_fn=None, extra_dependencies=None)

收集 Table 基于依赖关系的对象。

这是一个依赖项顺序排序,将发出 Table 对象,使它们跟随其从属对象 Table 物体。表依赖于另一个基于存在的 ForeignKeyConstraint 对象以及由添加的显式依赖项 Table.add_is_dependent_on() .

警告

这个 sort_tables() 函数本身不能自动解析表之间的依赖循环,这通常是由相互依赖的外键约束引起的。当检测到这些循环时,将在排序时忽略这些表的外键。当这种情况发生时,会发出一个警告,这将在将来的版本中引发异常。不属于循环一部分的表仍将按依赖关系顺序返回。

为了解决这些周期 ForeignKeyConstraint.use_alter 参数可应用于创建循环的约束。或者 sort_tables_and_constraints() 当检测到循环时,函数将自动返回单独集合中的外键约束,以便将它们分别应用于架构。

在 1.3.17 版更改: -出现以下情况时发出警告 sort_tables() 由于循环依赖关系,无法执行正确的排序。这将是未来版本中的一个例外。此外,sort将继续按依赖关系顺序返回循环中未涉及的其他表(以前不是这样)。

参数
  • tables -- 一系列 Table 物体。

  • skip_fn -- 可选可调用,将通过 ForeignKey 对象;如果返回true,则不会将此约束视为依赖项。注意这是 不同的 来自中的相同参数 sort_tables_and_constraints() ,而不是通过拥有 ForeignKeyConstraint 对象。

  • extra_dependencies -- 表的两个元组的序列,也被认为是相互依赖的。

参见

sort_tables_and_constraints()

MetaData.sorted_tables -使用此函数进行排序

function sqlalchemy.schema.sort_tables_and_constraints(tables, filter_fn=None, extra_dependencies=None, _warn_for_cycles=False)

收集 Table / ForeignKeyConstraint 物体。

这是一个依赖项顺序排序,它将发出 (Table, [ForeignKeyConstraint, ...]) 这样每个 Table 依附于它 Table 物体。剩下的 ForeignKeyConstraint 由于排序不满足依赖关系规则而分离的对象随后作为 (None, [ForeignKeyConstraint ...]) .

表依赖于另一个基于存在的 ForeignKeyConstraint 对象,由添加的显式依赖项 Table.add_is_dependent_on() 以及此处使用 sort_tables_and_constraints.skip_fn 和/或 sort_tables_and_constraints.extra_dependencies 参数。

参数
  • tables -- 一系列 Table 物体。

  • filter_fn -- 可选可调用,将通过 ForeignKeyConstraint 对象,并返回一个值,该值基于是否应将此约束作为内联约束明确包含或排除,或者两者都不包含。如果返回false,约束将被作为一个不能更改的依赖项包含;如果为true,它将 only 在结尾处作为一个更改结果包括在内。返回none意味着约束包含在基于表的结果中,除非它作为依赖循环的一部分被检测到。

  • extra_dependencies -- 表的两个元组的序列,也被认为是相互依赖的。

1.0.0 新版功能.

参见

sort_tables()

class sqlalchemy.schema.DDLElement

DDL表达式构造的基类。

这个类是通用的基础 DDL 类以及各种create/drop子句构造,例如 CreateTableDropTableAddConstraint 等。

DDLElement 与SQLAlchemy事件紧密集成,在 事件 . 一个实例本身就是一个接收可调用的事件:

event.listen(
    users,
    'after_create',
    AddConstraint(constraint).execute_if(dialect='postgresql')
)

类签名

class sqlalchemy.schema.DDLElement (sqlalchemy.sql.roles.DDLRole, sqlalchemy.sql.expression.Executable, sqlalchemy.schema._DDLCompiles)

method sqlalchemy.schema.DDLElement.__call__(target, bind, **kw)

作为DDL侦听器执行DDL。

method sqlalchemy.schema.DDLElement.against(target)

返回此的副本 DDLElement 包括给定的目标。

这实际上是将给定的项应用于 .target 返回的属性 DDLElement 对象。然后,事件处理程序和编译例程可以使用该目标,以便提供服务,例如根据特定的 Table .

当A DDLElement 对象被建立为 DDLEvents.before_create()DDLEvents.after_create() 事件,然后针对给定目标(如 ConstraintTable ,该目标是用 DDLElement 对象,该方法随后将转到 DDLElement.execute() 方法调用实际的DDL指令。

参数

target -- 一 SchemaItem 这将是DDL操作的主题。

返回

一份副本 DDLElement.target 分配给给定的 SchemaItem .

参见

DDL -在处理DDL字符串时对“target”使用标记化。

attribute sqlalchemy.schema.DDLElement.bind
attribute sqlalchemy.schema.DDLElement.callable_ = None
attribute sqlalchemy.schema.DDLElement.dialect = None
method sqlalchemy.schema.DDLElement.execute(bind=None, target=None)

立即执行此DDL。

1.4 版后已移除: 这个 DDLElement.execute() 方法被认为是SQLAlChemy 1.x系列的遗留方法,将在2.0中删除。SQLAlChemy 2.0中的所有语句都由 Connection.execute() 一种方法 Connection ,或在ORM中由 Session.execute() 一种方法 Session 。(有关SQLAlChemy 2.0的背景信息位于: 迁移到Alchemy )

使用提供的 ConnectableConnectable 分配给 .bind 属性(如果未提供)。如果DDL有条件 on 条件,它将以“无”作为事件被调用。

参数
method sqlalchemy.schema.DDLElement.execute_if(dialect=None, callable_=None, state=None)

返回将执行此操作的可调用 DDLElement 在事件处理程序中有条件地。

用于为事件侦听提供包装:

event.listen(
            metadata,
            'before_create',
            DDL("my_ddl").execute_if(dialect='postgresql')
        )
参数
  • dialect -- 可以是字符串或字符串元组。如果是字符串,将与正在执行的数据库方言的名称进行比较(如果是元组,则指定多个方言名称::DDL(‘something’).execute_if(dialect=‘postgresql’)‘mysql’))

  • callable_ -- 可调用,将使用四个位置参数和可选关键字参数调用: :ddl: This DDL element. :target: The TableMetaData 对象,它是此事件的目标。如果显式执行DDL,则可以为“无”。 :bind: The Connection 用于DDL执行:tables:可选关键字参数-要在元数据中创建/删除的表对象列表。create_all()或drop_all()方法调用。:state:可选关键字参数-将是 state 传递给此函数的参数。:checkfirst:keyword参数,如果在调用期间设置了“checkfirst”标志,则该参数为true。 create()create_all()drop()drop_all() . 如果callable返回真值,则将执行DDL语句。

  • state -- 将传递给 callable_ 作为 state 关键字参数。

参见

DDLEvents

事件

attribute sqlalchemy.schema.DDLElement.on = None
attribute sqlalchemy.schema.DDLElement.target = None
class sqlalchemy.schema.DDL(statement, context=None, bind=None)

字面DDL语句。

指定要由数据库执行的文本SQL DDL。DDL对象用作DDL事件侦听器,可以订阅中列出的那些事件。 DDLEvents ,使用任一 TableMetaData 对象作为目标。基本模板支持允许单个DDL实例处理多个表的重复任务。

实例:

from sqlalchemy import event, DDL

tbl = Table('users', metadata, Column('uid', Integer))
event.listen(tbl, 'before_create', DDL('DROP TRIGGER users_trigger'))

spow = DDL('ALTER TABLE %(table)s SET secretpowers TRUE')
event.listen(tbl, 'after_create', spow.execute_if(dialect='somedb'))

drop_spow = DDL('ALTER TABLE users SET secretpowers FALSE')
connection.execute(drop_spow)

当操作表事件时,以下 statement 字符串替换可用:

%(table)s  - the Table name, with any required quoting applied
%(schema)s - the schema name, with any required quoting applied
%(fullname)s - the Table name including schema, quoted if needed

DDL的“上下文”(如果有的话)将与上面提到的标准替换相结合。上下文中存在的键将覆盖标准替换。

method sqlalchemy.schema.DDL.__init__(statement, context=None, bind=None)

创建DDL语句。

参数
  • statement -- 要执行的字符串或Unicode字符串。语句将使用python的字符串格式运算符进行处理。见 context 争论和 execute_at 方法。语句中的文本“%”必须作为“%”转义。SQL绑定参数在DDL语句中不可用。

  • context -- 可选字典,默认为无。这些值可用于DDL语句上的字符串替换。

  • bind -- 可选的。一 Connectable ,默认情况下在 execute() 在没有绑定参数的情况下调用。。已弃用::1.4 DDL.bind 参数已弃用,将在SQLAlchemy 2.0中删除。

参见

DDLEvents

事件

class sqlalchemy.schema._CreateDropBase(element, bind=None, if_exists=False, if_not_exists=False, _legacy_bind=None)

表示创建和删除或等效项的DDL构造的基类。

CreateDropBase的共同主题是 element 属性,它引用要创建或删除的元素。

类签名

class sqlalchemy.schema._CreateDropBase (sqlalchemy.schema.DDLElement)

class sqlalchemy.schema.CreateTable(element, bind=None, include_foreign_key_constraints=None, if_not_exists=False)

表示create table语句。

类签名

class sqlalchemy.schema.CreateTable (sqlalchemy.schema._CreateDropBase)

method sqlalchemy.schema.CreateTable.__init__(element, bind=None, include_foreign_key_constraints=None, if_not_exists=False)

创建一个 CreateTable 构造。

参数
  • element -- 一 Table 这就是创造的主题

  • on -- 请参阅中“on”的说明。 DDL .

  • bind -- 请参阅中“bind”的说明。 DDL .

1.4 版后已移除: 这个 CreateTable.bind 参数已弃用,将在SQLAlChemy 2.0中删除。

参数
  • include_foreign_key_constraints -- 可选顺序 ForeignKeyConstraint 将包含在create构造中的内联对象;如果省略,则包括所有未指定use_alter=true的外键约束。…版本已添加::1.0.0

  • if_not_exists -- 如果为True,则IF NOT EXISTS运算符将应用于构造。。。添加的版本::1.4.0b2

class sqlalchemy.schema.DropTable(element, bind=None, if_exists=False)

表示Drop Table语句。

类签名

class sqlalchemy.schema.DropTable (sqlalchemy.schema._CreateDropBase)

method sqlalchemy.schema.DropTable.__init__(element, bind=None, if_exists=False)

创建 DropTable 构造。

参数
  • element -- 一个 Table 这就是这次投放的主题。

  • on -- 请参阅中“on”的说明。 DDL .

  • bind -- 请参阅中“bind”的说明。 DDL .

1.4 版后已移除: 这个 DropTable.bind 参数已弃用,将在SQLAlChemy 2.0中删除。

参数

if_exists -- 如果为True,则IF EXISTS运算符将应用于构造。。。添加的版本::1.4.0b2

class sqlalchemy.schema.CreateColumn(element)

代表一个 Column 在create table语句中,通过 CreateTable 构造。

它通过使用中记录的编译器扩展来支持在生成create table语句的过程中自定义列DDL。 自定义SQL构造和编译扩展 延伸 CreateColumn .

典型的集成是检查传入的 Column 对象,并在找到特定标志或条件时重定向编译:

from sqlalchemy import schema
from sqlalchemy.ext.compiler import compiles

@compiles(schema.CreateColumn)
def compile(element, compiler, **kw):
    column = element.element

    if "special" not in column.info:
        return compiler.visit_create_column(element, **kw)

    text = "%s SPECIAL DIRECTIVE %s" % (
            column.name,
            compiler.type_compiler.process(column.type)
        )
    default = compiler.get_column_default_string(column)
    if default is not None:
        text += " DEFAULT " + default

    if not column.nullable:
        text += " NOT NULL"

    if column.constraints:
        text += " ".join(
                    compiler.process(const)
                    for const in column.constraints)
    return text

上述构造可应用于 Table 如下:

from sqlalchemy import Table, Metadata, Column, Integer, String
from sqlalchemy import schema

metadata = MetaData()

table = Table('mytable', MetaData(),
        Column('x', Integer, info={"special":True}, primary_key=True),
        Column('y', String(50)),
        Column('z', String(20), info={"special":True})
    )

metadata.create_all(conn)

上面,我们添加到 Column.info 我们的自定义编译方案将检测到集合:

CREATE TABLE mytable (
        x SPECIAL DIRECTIVE INTEGER NOT NULL,
        y VARCHAR(50),
        z SPECIAL DIRECTIVE VARCHAR(20),
    PRIMARY KEY (x)
)

这个 CreateColumn 构造还可用于在生成 CREATE TABLE . 这是通过创建一个有条件地返回 None . 这基本上就是如何产生与使用 system=True 争论 Column 将列标记为隐式存在的“系统”列。

例如,假设我们希望 Table 跳过了PostgreSQL的渲染 xmin 针对PostgreSQL后端的列,但在其他后端会呈现该列,以预测触发的规则。条件编译规则只能在PostgreSQL上跳过此名称:

from sqlalchemy.schema import CreateColumn

@compiles(CreateColumn, "postgresql")
def skip_xmin(element, compiler, **kw):
    if element.element.name == 'xmin':
        return None
    else:
        return compiler.visit_create_column(element, **kw)


my_table = Table('mytable', metadata,
            Column('id', Integer, primary_key=True),
            Column('xmin', Integer)
        )

以上,A CreateTable 构造将生成 CREATE TABLE 其中只包括 id 字符串中的列;列 xmin 列将被省略,但只针对PostgreSQL后端。

类签名

class sqlalchemy.schema.CreateColumn (sqlalchemy.schema._DDLCompiles)

class sqlalchemy.schema.CreateSequence(element, bind=None, if_exists=False, if_not_exists=False, _legacy_bind=None)

表示创建序列语句。

类签名

class sqlalchemy.schema.CreateSequence (sqlalchemy.schema._CreateDropBase)

class sqlalchemy.schema.DropSequence(element, bind=None, if_exists=False, if_not_exists=False, _legacy_bind=None)

表示一个DROP SEQUENCE语句。

类签名

class sqlalchemy.schema.DropSequence (sqlalchemy.schema._CreateDropBase)

class sqlalchemy.schema.CreateIndex(element, bind=None, if_not_exists=False)

表示创建索引语句。

类签名

class sqlalchemy.schema.CreateIndex (sqlalchemy.schema._CreateDropBase)

method sqlalchemy.schema.CreateIndex.__init__(element, bind=None, if_not_exists=False)

创建 Createindex 构造。

参数
  • element -- 一个 Index 这就是创造的主题。

  • on -- 请参阅中“on”的说明。 DDL .

  • bind -- 请参阅中“bind”的说明。 DDL .

1.4 版后已移除: 这个 CreateIndex.bind 参数已弃用,将在SQLAlChemy 2.0中删除。

参数

if_not_exists -- 如果为True,则IF NOT EXISTS运算符将应用于构造。。。添加的版本::1.4.0b2

class sqlalchemy.schema.DropIndex(element, bind=None, if_exists=False)

表示一个DROP INDEX语句。

类签名

class sqlalchemy.schema.DropIndex (sqlalchemy.schema._CreateDropBase)

method sqlalchemy.schema.DropIndex.__init__(element, bind=None, if_exists=False)

创建 DropIndex 构造。

参数
  • element -- 一个 Index 这就是这次投放的主题。

  • on -- 请参阅中“on”的说明。 DDL .

  • bind -- 请参阅中“bind”的说明。 DDL .

1.4 版后已移除: 这个 DropIndex.bind 参数已弃用,将在SQLAlChemy 2.0中删除。

参数

if_exists -- 如果为True,则IF EXISTS运算符将应用于构造。。。添加的版本::1.4.0b2

class sqlalchemy.schema.AddConstraint(element, *args, **kw)

表示alter table add约束语句。

类签名

class sqlalchemy.schema.AddConstraint (sqlalchemy.schema._CreateDropBase)

class sqlalchemy.schema.DropConstraint(element, cascade=False, **kw)

表示alter table drop约束语句。

类签名

class sqlalchemy.schema.DropConstraint (sqlalchemy.schema._CreateDropBase)

class sqlalchemy.schema.CreateSchema(name, quote=None, **kw)

表示创建架构语句。

这里的参数是模式的字符串名称。

类签名

class sqlalchemy.schema.CreateSchema (sqlalchemy.schema._CreateDropBase)

method sqlalchemy.schema.CreateSchema.__init__(name, quote=None, **kw)

创建新的 CreateSchema 构造。

class sqlalchemy.schema.DropSchema(name, quote=None, cascade=False, **kw)

表示一个DROP SCHEMA语句。

这里的参数是模式的字符串名称。

类签名

class sqlalchemy.schema.DropSchema (sqlalchemy.schema._CreateDropBase)

method sqlalchemy.schema.DropSchema.__init__(name, quote=None, cascade=False, **kw)

创建新的 DropSchema 构造。

Previous: 定义约束和索引 Next: 列和数据类型