自定义 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_alterparameter 可以应用于那些创建循环的约束。或者,当检测到周期时,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构造。
-