自定义 DDL¶
在前面的部分中,我们讨论了各种 schema 结构,包括 Table
、
ForeignKeyConstraint
和
CheckConstraint
和
序列
。自始至终,我们都依赖于
create()
和 create_all()
方法
Table
和 MetaData
来为所有结构发布数据定义语言 (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
构造还能够根据数据库检查有条件地调用。此功能可通过 ExecutableDDLElement.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"))
该 ExecutableDDLElement.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"))
)
该方法 ExecutableDDLElement.execute_if()
还可以针对将接收正在使用的数据库连接的可调用函数工作。在下面的示例中,我们使用它来有条件地创建一个 CHECK 约束,首先在 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
),
)
users.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)
users.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
使用内置的 DDLElement 类¶
sqlalchemy.schema
包包含 SQL 表达式构造,这些构造
提供 DDL 表达式,所有这些表达式都从公共基础扩展而来
ExecutableDDLElement 的 ExecutableDDLElement
中。例如,要生成 CREATE TABLE
语句,可以使用 CreateTable
构造:
from sqlalchemy.schema import CreateTable
with engine.connect() as conn:
conn.execute(CreateTable(mytable))
CREATE TABLE mytable (
col1 INTEGER,
col2 INTEGER,
col3 INTEGER,
col4 INTEGER,
col5 INTEGER,
col6 INTEGER
)
在上面,CreateTable
构造的工作方式与任何其他表达式构造(例如 select()、
table.insert()
等)类似。所有 SQLAlchemy 面向 DDL 的构造都是 ExecutableDDLElement
基类的子类;这是与 CREATE 和 DROP 以及 ALTER 对应的所有对象的基础,不仅在 SQLAlchemy 中,而且在 Alembic 迁移中也是如此。可用构造的完整参考位于 DDL 表达式构造 API 中。
用户定义的 DDL 结构也可以创建为
ExecutableDDLElement
本身。 中的文档
Custom SQL Constructs and Compilation Extension 提供了几个示例。
控制约束和索引的 DDL 生成¶
2.0 版的新Function。
虽然前面提到 ExecutableDDLElement.execute_if()
的方法对于需要有条件调用的自定义 DDL
类很有用,但通常与特定 Table
相关的元素(即约束和索引)也需要遵守“条件”规则,例如包含特定于特定后端(如 PostgreSQL 或 SQL Server)的功能的索引。对于此用例,Constraint.ddl_if()
和 Index.ddl_if()
方法可用于 CheckConstraint
、
UniqueConstraint
和 Index
,接受与 ExecutableDDLElement.execute_if()
该方法相同的参数,以便控制
他们的 DDL 是否会根据其父级发出
Table
对象。在为 Table
创建定义时,可以内联使用这些方法
(或类似地,在 ORM 声明式映射中使用 __table_args__
集合时),例如:
from sqlalchemy import CheckConstraint, Index
from sqlalchemy import MetaData, Table, Column
from sqlalchemy import Integer, String
meta = MetaData()
my_table = Table(
"my_table",
meta,
Column("id", Integer, primary_key=True),
Column("num", Integer),
Column("data", String),
Index("my_pg_index", "data").ddl_if(dialect="postgresql"),
CheckConstraint("num > 5").ddl_if(dialect="postgresql"),
)
在上面的示例中,Table
结构引用了
Index
和 CheckConstraint
构造,两者都表示 .ddl_if(dialect=“postgresql”)
,这表示这些元素将仅包含在 PostgreSQL 方言的 CREATE TABLE 序列中。例如,如果我们对 SQLite 方言运行 meta.create_all(),
则不会包含任何构造:
>>> from sqlalchemy import create_engine
>>> sqlite_engine = create_engine("sqlite+pysqlite://", echo=True)
>>> meta.create_all(sqlite_engine)
BEGIN (implicit)
PRAGMA main.table_info("my_table")
[raw sql] ()
PRAGMA temp.table_info("my_table")
[raw sql] ()
CREATE TABLE my_table (
id INTEGER NOT NULL,
num INTEGER,
data VARCHAR,
PRIMARY KEY (id)
)
但是,如果我们对 PostgreSQL 数据库运行相同的命令,我们将看到 CHECK 约束的内联 DDL 以及为索引发出的单独 CREATE 语句:
>>> from sqlalchemy import create_engine
>>> postgresql_engine = create_engine(
... "postgresql+psycopg2://scott:tiger@localhost/test", echo=True
... )
>>> meta.create_all(postgresql_engine)
BEGIN (implicit)
select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
[generated in 0.00009s] {'name': 'my_table'}
CREATE TABLE my_table (
id SERIAL NOT NULL,
num INTEGER,
data VARCHAR,
PRIMARY KEY (id),
CHECK (num > 5)
)
[no key 0.00007s] {}
CREATE INDEX my_pg_index ON my_table (data)
[no key 0.00013s] {}
COMMIT
Constraint.ddl_if()
和 Index.ddl_if()
方法创建一个事件钩子,该钩子不仅可以在 DDL 执行时查询,就像 ExecutableDDLElement.execute_if()
的行为一样,还可以在 CreateTable
对象的 SQL 编译阶段进行查询,该对象负责呈现 CHECK (num > 5)
DDL 内联在 CREATE TABLE 语句中。因此,ddl_if.callable_()
接收的事件钩子
parameter 存在更丰富的参数集,包括
传递的 dialect
关键字参数,以及 DDLCompiler
的实例
通过序列的 “inline rendering” 部分的 compiler
关键字参数。在 DDLCompiler
序列中触发事件时,bind
参数不存在,因此希望
要检查数据库版本控制信息,最好使用给定的
Dialect
对象,例如测试 PostgreSQL 版本控制:
def only_pg_14(ddl_element, target, bind, dialect, **kw):
return dialect.name == "postgresql" and dialect.server_version_info >= (14,)
my_table = Table(
"my_table",
meta,
Column("id", Integer, primary_key=True),
Column("num", Integer),
Column("data", String),
Index("my_pg_index", "data").ddl_if(callable_=only_pg_14),
)
DDL 表达式构造 API¶
对象名称 |
描述 |
---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
-
函数 sqlalchemy.schema。sort_tables(tables: Iterable[TableClause], skip_fn:Callable[[ForeignKeyConstraint],bool]None=None, extra_dependencies:typing_Sequence[Tuple[TableClause,TableClause]]None=None)List[Table] ¶
根据依赖关系对Table
对象的集合进行排序。
这是一个依赖关系排序的排序,它将发出Table
对象,以便它们遵循其依赖的 Table
对象。 表依赖于另一个表,具体取决于ForeignKeyConstraint
的 对象以及显式依赖项 由Table.add_is_dependent_on()
添加。
警告sort_tables()
函数本身无法容纳 table 之间依赖关系循环的自动解析,这通常是由相互依赖的外键约束引起的。当检测到这些循环时,这些表的外键将在排序中被忽略。发生这种情况时,会发出警告,这将是将来版本中的异常引发。不属于 cycle 的 table 仍将按依赖关系 Sequences 返回。
为了解决这些循环,ForeignKeyConstraint.use_alter
parameter 可以应用于那些创建循环的约束。或者,当检测到周期时,sort_tables_and_constraints()
函数将自动在单独的集合中返回外键约束,以便它们可以单独应用于架构。
在 1.3.17 版本发生变更: - 当sort_tables()
由于循环依赖关系而无法执行正确的排序。这将是未来版本中的例外。此外,排序将继续按依赖关系 Sequences 返回周期中不涉及的其他 table,这在以前是不是这种情况。
参数
skip_fn¶ —— 可选的可调用对象,它将传递一个ForeignKeyConstraint
对象;如果返回 True,则 this constraint 将不被视为依赖项。 请注意,这是 与 中的相同参数不同sort_tables_and_constraints()
中,它被传递给拥有的ForeignKeyConstraint
对象。
extra_dependencies¶ – 一个 2 元组的表格序列,这些表格也将被视为彼此依赖。
-
函数 sqlalchemy.schema。sort_tables_and_constraints(表, filter_fn=无, extra_dependencies=无, _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
参数。
参数
filter_fn¶ —— 可选的可调用对象,它将传递一个ForeignKeyConstraint
对象,并返回一个值,该值基于此约束是绝对应该作为内联约束包含还是排除,或者两者都不包含。如果它返回 False,则约束肯定会作为依赖项包含在内,该依赖项不能受 ALTER 的约束;如果为 True,则它只会作为 ALTER 结果包含在末尾。返回 None 表示约束包含在基于表的结果中,除非它被检测为依赖关系循环的一部分。
extra_dependencies¶ – 一个 2 元组的表格序列,这些表格也将被视为彼此依赖。
另请参阅
-
类 sqlalchemy.schema 中。BaseDDLEle文件¶
DDL 结构的根,包括那些作为 “create table” 和其他进程中的子元素的 DDL 结构。
2.0 版的新Function。
-
类 sqlalchemy.schema 中。ExecutableDDLElement¶
独立可执行 DDL 表达式构造的基类。
该类是通用DDL 类 (DDL
类) 的基础 以及各种 create/drop 子句结构,例如CreateTable
、DropTable
、AddConstraint
等。
在 2.0 版本发生变更:ExecutableDDLElement
已从DDLElement 的 DDLElement
中,它仍然存在以实现向后兼容性。ExecutableDDLElement
与 Events中介绍的 SQLAlchemy 事件紧密集成。一个 1 的实例本身就是一个接收 callable 的事件:event.listen( users, "after_create", AddConstraint(constraint).execute_if(dialect="postgresql"), )
类签名
类sqlalchemy.schema.ExecutableDDLElement
(sqlalchemy.sql.roles.DDLRole
,sqlalchemy.sql.expression.Executable
,sqlalchemy.schema.BaseDDLElement
)-
方法sqlalchemy.schema.ExecutableDDLElement.
__call__(target, bind, **kw)¶
将 DDL 作为 ddl_listener 执行。
-
methodsqlalchemy.schema.ExecutableDDLElement.
against(target: SchemaItem)Self ¶
返回此ExecutableDDLElement
的副本,该副本将包含给定的目标。
这实质上是将给定的项应用于返回的ExecutableDDLElement
对象的.target
属性。然后,事件处理程序和编译例程可以使用此目标,以便根据特定Table
提供服务,例如 DDL 字符串的标记化。
当ExecutableDDLElement
对象建立为DDLEvents.before_create()
或DDLEvents.after_create()
事件,然后该事件会针对给定目标(例如Constraint
或表中
,该目标是使用ExecutableDDLElement
对象,然后转到ExecutableDDLElement.execute()
方法调用实际的 DDL 指令。
参数
target¶—— 将成为 DDL作主题的SchemaItem
。
结果
此ExecutableDDLElement
的副本,其中包含.target
属性分配给给定的SchemaItem 的 SchemaItem
中。
另请参阅DDL
- 在处理 DDL 字符串时对 “目标” 使用分词。
-
方法sqlalchemy.schema.ExecutableDDLElement.
execute_if(dialect:strNone=None, callable_:DDLIfCallableNone=None, state:AnyNone=None)Self ¶
返回一个将执行此ExecutableDDLElement
的实例。
用于为事件侦听提供包装器:event.listen( metadata, "before_create", DDL("my_ddl").execute_if(dialect="postgresql"), )
参数
方言¶ –
可以是字符串或字符串的 Tuples。如果是字符串,则会将其与正在执行的数据库方言的名称进行比较:DDL("something").execute_if(dialect="postgresql")
如果是 tuples,则指定多个方言名称:DDL("something").execute_if(dialect=("postgresql", "mysql"))
callable_¶ –
一个可调用对象,它将使用三个位置参数和可选的关键字参数进行调用:
DDL 中:
这个 DDL 元素。
目标:
绑定:
用于 DDL 执行的Connection
。如果此构造是在表中内联创建的,则可能是 None,在这种情况下,将出现compiler
。
表:
可选关键字参数 - 要在 MetaData.create_all() 或 drop_all() 方法调用中创建/删除的 Table 对象列表。
方言:
keyword 参数,但始终存在 - 该 作涉及的方言
。
编译器:
keyword 参数。对于引擎级 DDL 调用,将为None
,但将引用DDLCompiler
如果此 DDL 元素是在表中内联创建的。
状态:
可选关键字参数 - 将是传递给此函数的状态
参数。
checkfirst 的
Keyword 参数,如果在调用create()、
create_all()
、drop()
和drop_all()
的
如果可调用对象返回 True 值,则将执行 DDL 语句。
state¶ —— 任何将作为state
关键字参数传递给 callable_ 的值。
-
-
类 sqlalchemy.schema 中。DDL (DDL) ¶
文本 DDL 语句。
指定要由数据库执行的文本 SQL DDL。DDL 对象充当 DDL 事件侦听器,可以使用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)
对 Table 事件进行作时,以下语句
可以使用字符串替换:%(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 的 “context” (如果有) 将与上述标准替换合并。上下文中存在的键将覆盖标准替换。
成员-
方法sqlalchemy.schema.DDL 中。
__init__(statement, context=None)¶
创建 DDL 语句。
参数
语句¶ –
要执行的字符串或 unicode 字符串。语句将使用 Python 的字符串格式化运算符处理,使用一组固定的字符串替换,以及可选的DDL.context
提供的其他替换 参数。
语句中的文本 '%' 必须转义为 '%%'。
SQL 绑定参数在 DDL 语句中不可用。
context¶——可选字典,默认为 None。这些值将可用于 DDL 语句的字符串替换。
-
-
类 sqlalchemy.schema 中。_CreateDropBase¶
表示 CREATE 和 DROP 或等效项的 DDL 构造的基类。
_CreateDropBase 的共同主题是单个element
属性,该属性引用要创建或删除的元素。
类签名
类sqlalchemy.schema._CreateDropBase
(sqlalchemy.schema.ExecutableDDLElement
)
-
类 sqlalchemy.schema 中。创建表¶
表示 CREATE TABLE 语句。
成员
类签名
类sqlalchemy.schema.CreateTable
(sqlalchemy.schema._CreateBase
)-
方法sqlalchemy.schema.CreateTable.
__init__(element: Table, include_foreign_key_constraints:typing_Sequence[ForeignKeyConstraint]None=None, if_not_exists: bool = False)¶
创建CreateTable
构造。
-
-
类 sqlalchemy.schema 中。DropTable(放置表)¶
表示 DROP TABLE 语句。
成员
类签名
类sqlalchemy.schema.DropTable
(sqlalchemy.schema._DropBase
)
-
类 sqlalchemy.schema 中。CreateColumn(创建列)¶
表示列
如 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
参数,它将列标记为隐式存在的 “system” 列。
例如,假设我们希望生成一个Table
哪个跳过 针对 PostgreSQL 后端呈现 PostgreSQLxmin
列,但在其他后端上确实呈现它,以预期触发的规则。条件编译规则只能在 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), )
在上面,CreateTable
构造将生成一个CREATE TABLE
它只包括id
列在字符串中;xmin
列将被省略,但仅限于 PostgreSQL 后端。
-
类 sqlalchemy.schema 中。CreateSequence(创建序列)¶
表示 CREATE SEQUENCE 语句。
类签名
类sqlalchemy.schema.CreateSequence
(sqlalchemy.schema._CreateBase
)
-
类 sqlalchemy.schema 中。DropSequence (放置序列)¶
表示 DROP SEQUENCE 语句。
类签名
类sqlalchemy.schema.DropSequence
(sqlalchemy.schema._DropBase
)
-
类 sqlalchemy.schema 中。创建索引¶
表示 CREATE INDEX 语句。
成员
类签名
类sqlalchemy.schema.CreateIndex
(sqlalchemy.schema._CreateBase
)-
方法sqlalchemy.schema.CreateIndex.
__init__(element, if_not_exists=False)¶
创建Createindex
构造。
-
-
类 sqlalchemy.schema 中。DropIndex(放置索引)¶
表示 DROP INDEX 语句。
成员
类签名
类sqlalchemy.schema.DropIndex
(sqlalchemy.schema._DropBase
)
-
类 sqlalchemy.schema 中。AddConstraint 插件¶
表示 ALTER TABLE ADD CONSTRAINT 语句。
类签名
类sqlalchemy.schema.AddConstraint
(sqlalchemy.schema._CreateBase
)
-
类 sqlalchemy.schema 中。DropConstraint 的
表示 ALTER TABLE DROP CONSTRAINT 语句。
类签名
类sqlalchemy.schema.DropConstraint
(sqlalchemy.schema._DropBase
)
-
类 sqlalchemy.schema 中。CreateSchema(创建架构)¶
表示 CREATE SCHEMA 语句。
此处的参数是 schema 的字符串名称。
成员
类签名
类sqlalchemy.schema.CreateSchema
(sqlalchemy.schema._CreateBase
)-
方法sqlalchemy.schema.CreateSchema.
__init__(name: str, if_not_exists: bool = False)¶
创建新的CreateSchema
构造。
-
-
类 sqlalchemy.schema 中。DropSchema¶
表示 DROP SCHEMA 语句。
此处的参数是 schema 的字符串名称。
成员
类签名
类sqlalchemy.schema.DropSchema
(sqlalchemy.schema._DropBase
)-
方法sqlalchemy.schema.DropSchema 中。
__init__(名称: str, cascade: bool = False, if_exists: bool = 假)¶
创建新的DropSchema
构造。
-