自定义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))
)

trigger = DDL(
    "CREATE TRIGGER dt_ins BEFORE INSERT ON mytable "
    "FOR EACH ROW BEGIN SET NEW.data='ins'; END"
)

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
sqlengine.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

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 参数可以应用于这些约束,或者使用 sql.sort_tables_and_constraints() 函数,它将分别打破循环中涉及的外键约束。

参数
  • tables -- 一系列 Table 对象。

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

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

参见

sort_tables_and_constraints()

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

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

排序的集合 Table / ForeignKeyConstraint 对象。

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

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

参数
  • tables -- 一系列 Table 对象。

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

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

1.0.0 新版功能.

参见

sort_tables()

class sqlalchemy.schema.DDLElement

基地: sqlalchemy.sql.expression.Executablesqlalchemy.schema._DDLCompiles

DDL表达式构造的基类。

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

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

event.listen(
    users,
    'after_create',
    AddConstraint(constraint).execute_if(dialect='postgresql')
)
__call__(target, bind, **kw)

作为DDL侦听器执行DDL。

against(target)

针对特定架构项返回此DDL的副本。

bind

返回 EngineConnection 对此 Executable 是绑定的,如果未找到,则为无。

这是一个遍历,它在本地进行检查,然后在关联对象的“from”子句之间进行检查,直到找到绑定的引擎或连接为止。

callable_ = None
dialect = None
execute(bind=None, target=None)

立即执行此DDL。

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

参数
  • bind -- 可选的,一个 EngineConnection . 如果未提供,则 Connectable 必须存在于 .bind 财产。

  • target -- 可选,默认为无。执行调用的目标SchemaItem。将传递给 on 可调用(如果有),也可以为语句提供字符串扩展数据。见 execute_at 更多信息。

execute_at(event_name, target)

将此DDL的执行链接到架构项的DDL生命周期。

0.7 版后已移除: 这个 DDLElement.execute_at() 方法已弃用,将在将来的版本中删除。请使用 DDLEvents 侦听器接口与 DDLElement.execute_if() 方法。

链接此 DDLElement 到A TableMetaData 实例,在创建或删除该架构项时执行。将使用与表create/drop本身相同的连接和事务上下文来执行DDL语句。这个 .bind 忽略此语句的属性。

参数
  • event -- 架构项中定义的事件之一 .ddl_events ;例如“创建前”、“创建后”、“删除前”或“删除后”

  • target -- 将与此ddlElement关联的表或元数据实例。

DDLEEMENT实例可以链接到任意数量的架构项。

execute_at 建立在 append_ddl_listener 界面 MetaDataTable 对象。

警告:单独创建或删除表也将触发任何DDL设置为 execute_at 该表的元数据。这在将来的版本中可能会改变。

execute_if(dialect=None, callable_=None, state=None)

返回将有条件地执行此ddlElement的可调用文件。

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

event.listen(
            metadata,
            'before_create',
            DDL("my_ddl").execute_if(dialect='postgresql')
        )
参数
  • dialect -- 可以是字符串、元组或可调用谓词。如果是字符串,则将其与执行数据库dialect::ddl(“something”)的名称进行比较。如果是元组,则执行“if”(dialect='postgresql'),指定多个方言名称::ddl(“something”)。如果是(dialect=“'postgresql”,“mysql”),则执行“if”

  • 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() . 如果可调用返回一个真值,则将执行DDL语句。

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

参见

DDLEvents

事件

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

基地: sqlalchemy.schema.DDLElement

字面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的“上下文”(如果有的话)将与上面提到的标准替换相结合。上下文中存在的键将覆盖标准替换。

__init__(statement, on=None, context=None, bind=None)

创建DDL语句。

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

  • on -- …已弃用::0.7 DDL.on 参数已弃用,将在将来的版本中删除。请参考 DDLElement.execute_if() . 可选筛选条件。可以是字符串、元组或可调用谓词。如果一个字符串,它将与执行数据库方言的名称进行比较:ddl(“something”,on='postgresql');如果一个元组,指定多个方言名称:ddl(“something”,on=“'postgresql”,“mysql”);如果一个可调用的字符串,它将使用四个位置参数以及可选关键字参数调用:ddl:this ddl元素。:event:如果显式执行DDL,则触发此DDL的事件的名称(如“创建后”)将为“无”。:目标:The TableMetaData 对象,它是此事件的目标。如果显式执行DDL,则可能为“无”。:连接:The Connection 用于DDL执行:tables:可选关键字参数-要在元数据中创建/删除的表对象列表。create_all()或drop_all()方法调用。如果可调用返回一个真值,则将执行DDL语句。

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

  • bind -- 可选的。一 Connectable ,默认情况下在 execute() 在没有绑定参数的情况下调用。

参见

DDLEvents

事件

class sqlalchemy.schema._CreateDropBase(element, on=None, bind=None)

基地: sqlalchemy.schema.DDLElement

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

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

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

基地: sqlalchemy.schema._CreateDropBase

表示create table语句。

__init__(element, on=None, bind=None, include_foreign_key_constraints=None)

创建一个 CreateTable 构造。

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

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

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

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

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

基地: sqlalchemy.schema._CreateDropBase

表示Drop Table语句。

class sqlalchemy.schema.CreateColumn(element)

基地: sqlalchemy.schema._DDLCompiles

代表一个 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.CreateSequence(element, on=None, bind=None)

基地: sqlalchemy.schema._CreateDropBase

表示创建序列语句。

class sqlalchemy.schema.DropSequence(element, on=None, bind=None)

基地: sqlalchemy.schema._CreateDropBase

表示一个DROP SEQUENCE语句。

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

基地: sqlalchemy.schema._CreateDropBase

表示创建索引语句。

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

基地: sqlalchemy.schema._CreateDropBase

表示一个DROP INDEX语句。

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

基地: sqlalchemy.schema._CreateDropBase

表示alter table add约束语句。

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

基地: sqlalchemy.schema._CreateDropBase

表示alter table drop约束语句。

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

基地: sqlalchemy.schema._CreateDropBase

表示创建架构语句。

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

__init__(name, quote=None, **kw)

创建新的 CreateSchema 构造。

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

基地: sqlalchemy.schema._CreateDropBase

表示一个DROP SCHEMA语句。

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

__init__(name, quote=None, cascade=False, **kw)

创建新的 DropSchema 构造。