在前面的章节中,我们讨论了各种模式构造,包括 Table
, ForeignKeyConstraint
, CheckConstraint
和 Sequence
. 一直以来,我们都依赖 create()
和 create_all()
方法 Table
和 MetaData
以便为所有构造发出数据定义语言(DDL)。当发出命令时,将调用预先确定的操作顺序,并且无条件创建每个表的DDL,包括所有约束和与之关联的其他对象。对于需要特定于数据库的DDL的更复杂的场景,SQLAlchemy提供了两种技术,可用于根据任何条件添加任何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
前面介绍的构造还可以根据对数据库的检查有条件地调用。此功能可使用 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)
CREATE TABLE users (
user_id SERIAL NOT NULL,
user_name VARCHAR(40) NOT NULL,
PRIMARY KEY (user_id)
)
select conname from pg_constraint where conname='cst_user_name_length'
ALTER TABLE users ADD CONSTRAINT cst_user_name_length CHECK (length(user_name) >= 8)
sqlusers.drop(engine)
select conname from pg_constraint where conname='cst_user_name_length'
ALTER TABLE users DROP CONSTRAINT cst_user_name_length
DROP TABLE users
这个 sqlalchemy.schema
包包含提供DDL表达式的SQL表达式构造。例如,生成 CREATE TABLE
声明:
from sqlalchemy.schema import CreateTable
sqlengine.execute(CreateTable(mytable))
CREATE TABLE mytable (
col1 INTEGER,
col2 INTEGER,
col3 INTEGER,
col4 INTEGER,
col5 INTEGER,
col6 INTEGER
)
上面, CreateTable
与任何其他表达式构造(例如 select()
, table.insert()
等)。所有SQLAlchemy的面向DDL的构造都是 DDLElement
基类;这是与创建、删除和更改相对应的所有对象的基础,不仅在SQLAlchemy中,而且在Alembic迁移中也是如此。可用构造的完整引用位于 DDL表达式构造API .
用户定义的DDL构造也可以创建为 DDLElement
本身。文件 自定义SQL构造和编译扩展 有几个这样的例子。
上一节中描述的事件驱动的DDL系统 控制DDL序列 可与其他人一起使用 DDLElement
对象也一样。但是,在处理诸如 CreateIndex
, CreateSequence
等,事件系统 有限的 使用,如 Table.create()
和 MetaData.create_all()
将无条件地调用这些构造。在未来的SQLAlchemy版本中,包含条件执行的DDL事件系统将考虑当前在所有情况下调用的内置构造。
我们可以用 AddConstraint
和 DropConstraint
构造,因为事件驱动系统将用于检查和唯一约束,使用这些方法就像我们在前面的示例中所做的那样 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)
CREATE TABLE users (
user_id SERIAL NOT NULL,
user_name VARCHAR(40) NOT NULL,
PRIMARY KEY (user_id)
)
select conname from pg_constraint where conname='cst_user_name_length'
ALTER TABLE users ADD CONSTRAINT cst_user_name_length CHECK (length(user_name) >= 8)
sqlusers.drop(engine)
select conname from pg_constraint where conname='cst_user_name_length'
ALTER TABLE users DROP CONSTRAINT cst_user_name_length
DROP TABLE users
上面的示例与内置的 AddConstraint
和 DropConstraint
对象,DDL事件目前的主要用途仍然集中在 DDL
构造自身以及用户定义的子类 DDLElement
这还不是 MetaData.create_all()
, Table.create()
以及相应的“丢弃”过程。
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()
函数,它将分别打破循环中涉及的外键约束。
skip_fn¶ -- 可选可调用,将通过 ForeignKey
对象;如果返回true,则不会将此约束视为依赖项。注意这是 不同的 来自中的相同参数 sort_tables_and_constraints()
,而不是通过拥有 ForeignKeyConstraint
对象。
extra_dependencies¶ -- 表的两个元组的序列,也被认为是相互依赖的。
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
参数。
filter_fn¶ -- 可选可调用,将通过 ForeignKeyConstraint
对象,并返回一个值,该值基于是否应将此约束作为内联约束明确包含或排除,或者两者都不包含。如果返回false,约束将被作为一个不能更改的依赖项包含;如果为true,它将 only 在结尾处作为一个更改结果包括在内。返回none意味着约束包含在基于表的结果中,除非它作为依赖循环的一部分被检测到。
extra_dependencies¶ -- 表的两个元组的序列,也被认为是相互依赖的。
1.0.0 新版功能.
sqlalchemy.schema.
DDLElement
¶基地: sqlalchemy.sql.expression.Executable
, sqlalchemy.schema._DDLCompiles
DDL表达式构造的基类。
这个类是通用的基础 DDL
类以及各种create/drop子句构造,例如 CreateTable
, DropTable
, AddConstraint
等。
DDLElement
与SQLAlchemy事件紧密集成,在 事件 . 一个实例本身就是一个接收可调用的事件:
event.listen(
users,
'after_create',
AddConstraint(constraint).execute_if(dialect='postgresql')
)
__call__
(target, bind, **kw)¶作为DDL侦听器执行DDL。
against
(target)¶针对特定架构项返回此DDL的副本。
bind
¶返回 Engine
或 Connection
对此 Executable
是绑定的,如果未找到,则为无。
这是一个遍历,它在本地进行检查,然后在关联对象的“from”子句之间进行检查,直到找到绑定的引擎或连接为止。
callable_
= None¶dialect
= None¶execute
(bind=None, target=None)¶立即执行此DDL。
使用提供的 Connectable
或 Connectable
分配给 .bind
属性(如果未提供)。如果DDL有条件 on
条件,它将以“无”作为事件被调用。
bind¶ -- 可选的,一个 Engine
或 Connection
. 如果未提供,则 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 Table
或 MetaData
实例,在创建或删除该架构项时执行。将使用与表create/drop本身相同的连接和事务上下文来执行DDL语句。这个 .bind
忽略此语句的属性。
DDLEEMENT实例可以链接到任意数量的架构项。
execute_at
建立在 append_ddl_listener
界面 MetaData
和 Table
对象。
警告:单独创建或删除表也将触发任何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 Table
或 MetaData
对象,它是此事件的目标。如果显式执行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
关键字参数。
on
= None¶target
= None¶sqlalchemy.schema.
DDL
(statement, on=None, context=None, bind=None)¶基地: sqlalchemy.schema.DDLElement
字面DDL语句。
指定要由数据库执行的文本SQL DDL。DDL对象用作DDL事件侦听器,可以订阅中列出的那些事件。 DDLEvents
,使用任一 Table
或 MetaData
对象作为目标。基本模板支持允许单个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 Table
或 MetaData
对象,它是此事件的目标。如果显式执行DDL,则可能为“无”。:连接:The Connection
用于DDL执行:tables:可选关键字参数-要在元数据中创建/删除的表对象列表。create_all()或drop_all()方法调用。如果可调用返回一个真值,则将执行DDL语句。
context¶ -- 可选字典,默认为无。这些值可用于DDL语句上的字符串替换。
bind¶ -- 可选的。一 Connectable
,默认情况下在 execute()
在没有绑定参数的情况下调用。
sqlalchemy.schema.
_CreateDropBase
(element, on=None, bind=None)¶基地: sqlalchemy.schema.DDLElement
表示创建和删除或等效项的DDL构造的基类。
CreateDropBase的共同主题是 element
属性,它引用要创建或删除的元素。
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
构造。
sqlalchemy.schema.
DropTable
(element, on=None, bind=None)¶基地: sqlalchemy.schema._CreateDropBase
表示Drop Table语句。
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后端。
sqlalchemy.schema.
CreateSequence
(element, on=None, bind=None)¶基地: sqlalchemy.schema._CreateDropBase
表示创建序列语句。
sqlalchemy.schema.
DropSequence
(element, on=None, bind=None)¶基地: sqlalchemy.schema._CreateDropBase
表示一个DROP SEQUENCE语句。
sqlalchemy.schema.
CreateIndex
(element, on=None, bind=None)¶基地: sqlalchemy.schema._CreateDropBase
表示创建索引语句。
sqlalchemy.schema.
DropIndex
(element, on=None, bind=None)¶基地: sqlalchemy.schema._CreateDropBase
表示一个DROP INDEX语句。
sqlalchemy.schema.
AddConstraint
(element, *args, **kw)¶基地: sqlalchemy.schema._CreateDropBase
表示alter table add约束语句。
sqlalchemy.schema.
DropConstraint
(element, cascade=False, **kw)¶基地: sqlalchemy.schema._CreateDropBase
表示alter table drop约束语句。
sqlalchemy.schema.
CreateSchema
(name, quote=None, **kw)¶基地: sqlalchemy.schema._CreateDropBase
表示创建架构语句。
这里的参数是模式的字符串名称。
__init__
(name, quote=None, **kw)¶创建新的 CreateSchema
构造。
sqlalchemy.schema.
DropSchema
(name, quote=None, cascade=False, **kw)¶基地: sqlalchemy.schema._CreateDropBase
表示一个DROP SCHEMA语句。
这里的参数是模式的字符串名称。
__init__
(name, quote=None, cascade=False, **kw)¶创建新的 DropSchema
构造。