SqlAlchemy 2.0 中文文档（五十）（4）-阿里云开发者社区

SqlAlchemy 2.0 中文文档（五十）（3）https://developer.aliyun.com/article/1563129

保存点支持

SQLite 支持 SAVEPOINT，仅在启动事务后才能运行。SQLAlchemy 的 SAVEPOINT 支持可在 Core 级别使用 Connection.begin_nested() 方法，在 ORM 级别使用 Session.begin_nested()。但是，除非采取解决方法，否则 SAVEPOINT 在 pysqlite 中将无法工作。

警告

pysqlite 和 aiosqlite 驱动存在未解决的问题，这些问题将 BEGIN 语句推迟到一个更大程度上比通常可行的程度。有关绕过此行为的技术，请参见 Serializable isolation / Savepoints / Transactional DDL 和 Serializable isolation / Savepoints / Transactional DDL (asyncio version) 部分。

事务性 DDL

SQLite 数据库还支持事务性 DDL。在这种情况下，pysqlite 驱动不仅在检测到 DDL 时无法启动事务，还会结束任何现有事务，因此需要采取解决方法。

警告

pysqlite 驱动中存在未解决的问题影响了 SQLite 的事务性 DDL，当遇到 DDL 时，该驱动器未发出 BEGIN 并且还强制执行 COMMIT 来取消任何事务。有关绕过此行为的技术，请参见 Serializable isolation / Savepoints / Transactional DDL 部分。

外键支持

当发出用于表的 CREATE 语句时，SQLite 支持 FOREIGN KEY 语法，但是默认情况下，这些约束对表的操作没有任何影响。

在 SQLite 上进行约束检查有三个先决条件：

必须使用至少版本 3.6.19 的 SQLite
必须在编译 SQLite 库时没有启用 SQLITE_OMIT_FOREIGN_KEY 或 SQLITE_OMIT_TRIGGER 符号。
必须在所有连接上发出 PRAGMA foreign_keys = ON 语句，包括对 MetaData.create_all() 的初始调用。

SQLAlchemy 允许通过事件的使用自动发出 PRAGMA 语句以进行新连接：

from sqlalchemy.engine import Engine
from sqlalchemy import event
@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()

警告

当启用 SQLite 外键时，不可能对包含相互依赖的外键约束的表发出 CREATE 或 DROP 语句；要发出这些表的 DDL，需要单独使用 ALTER TABLE 创建或删除这些约束，而 SQLite 不支持这一点。

另请参阅

SQLite 外键支持 - 在 SQLite 网站上。

事件 - SQLAlchemy 事件 API。

通过 ALTER 创建/删除外键约束 - 有关 SQLAlchemy 处理的更多信息

互相依赖的外键约束。

对约束的 ON CONFLICT 支持

另请参见

本节描述了 SQLite 中“ON CONFLICT”的 DDL 版本，该版本出现在 CREATE TABLE 语句中。有关应用于 INSERT 语句的“ON CONFLICT”，请参见 INSERT…ON CONFLICT (Upsert)。

SQLite 支持一个名为 ON CONFLICT 的非标准 DDL 子句，可应用于主键、唯一、检查和非空约束。在 DDL 中，它要么在“CONSTRAINT”子句中呈现，要么在目标约束的位置取决于列定义本身。要在 DDL 中呈现此子句，可以在PrimaryKeyConstraint、UniqueConstraint、CheckConstraint对象中指定扩展参数sqlite_on_conflict，并在Column对象中，有单独的参数sqlite_on_conflict_not_null、sqlite_on_conflict_primary_key、sqlite_on_conflict_unique，分别对应于可以从Column对象指示的三种相关约束类型。

另请参见

冲突时执行 - 在 SQLite 文档中

版本 1.3 中的新功能。

sqlite_on_conflict参数接受一个字符串参数，该参数只是要选择的解决方案名称，在 SQLite 上可以是 ROLLBACK、ABORT、FAIL、IGNORE 和 REPLACE 中的一个。例如，要添加一个指定 IGNORE 算法的唯一约束：

some_table = Table(
    'some_table', metadata,
    Column('id', Integer, primary_key=True),
    Column('data', Integer),
    UniqueConstraint('id', 'data', sqlite_on_conflict='IGNORE')
)

以上将 CREATE TABLE DDL 呈现为：

CREATE TABLE some_table (
    id INTEGER NOT NULL,
    data INTEGER,
    PRIMARY KEY (id),
    UNIQUE (id, data) ON CONFLICT IGNORE
)

当使用Column.unique标志向单个列添加唯一约束时，也可以向Column添加sqlite_on_conflict_unique参数，该参数将添加到 DDL 中的唯一约束中：

some_table = Table(
    'some_table', metadata,
    Column('id', Integer, primary_key=True),
    Column('data', Integer, unique=True,
           sqlite_on_conflict_unique='IGNORE')
)

渲染：

CREATE TABLE some_table (
    id INTEGER NOT NULL,
    data INTEGER,
    PRIMARY KEY (id),
    UNIQUE (data) ON CONFLICT IGNORE
)

要应用 FAIL 算法以满足 NOT NULL 约束，使用sqlite_on_conflict_not_null：

some_table = Table(
    'some_table', metadata,
    Column('id', Integer, primary_key=True),
    Column('data', Integer, nullable=False,
           sqlite_on_conflict_not_null='FAIL')
)

这将使列内联 ON CONFLICT 短语：

CREATE TABLE some_table (
    id INTEGER NOT NULL,
    data INTEGER NOT NULL ON CONFLICT FAIL,
    PRIMARY KEY (id)
)

同样，对于内联主键，使用sqlite_on_conflict_primary_key：

some_table = Table(
    'some_table', metadata,
    Column('id', Integer, primary_key=True,
           sqlite_on_conflict_primary_key='FAIL')
)

SQLAlchemy 将主键约束单独呈现，因此冲突解决算法应用于约束本身：

CREATE TABLE some_table (
    id INTEGER NOT NULL,
    PRIMARY KEY (id) ON CONFLICT FAIL
)

插入…冲突时执行（Upsert）

另请参见

本节描述了 SQLite 的“ON CONFLICT”的 DML 版本，它发生在 INSERT 语句中。有关应用于 CREATE TABLE 语句的“ON CONFLICT”，请参见约束的 ON CONFLICT 支持。

从版本 3.24.0 开始，SQLite 支持通过 INSERT 语句的 ON CONFLICT 子句进行行的“upserts”（更新或插入）到表中。仅当候选行不违反任何唯一或主键约束时才会插入该行。在唯一约束违反的情况下，可以发生次要操作，可以是“DO UPDATE”，表示应更新目标行中的数据，或者是“DO NOTHING”，表示默默地跳过此行。

冲突是使用现有唯一约束和索引的列确定的。这些约束通过说明组成索引的列和条件来确定。

SQLAlchemy 通过 SQLite 特定的 insert() 函数提供 ON CONFLICT 支持，该函数提供生成方法 Insert.on_conflict_do_update() 和 Insert.on_conflict_do_nothing()：

>>> from sqlalchemy.dialects.sqlite import insert
>>> insert_stmt = insert(my_table).values(
...     id='some_existing_id',
...     data='inserted value')
>>> do_update_stmt = insert_stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value')
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (?,  ?)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  ?
>>> do_nothing_stmt = insert_stmt.on_conflict_do_nothing(
...     index_elements=['id']
... )
>>> print(do_nothing_stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (?,  ?)
ON  CONFLICT  (id)  DO  NOTHING

新版本 1.4 中新增。

另请参阅

Upsert - SQLite 文档中的内容。

指定目标

两种方法都使用列推断冲突的“目标”：

Insert.on_conflict_do_update.index_elements 参数指定包含字符串列名称、Column 对象和/或 SQL 表达式元素的序列，这些元素将标识唯一索引或唯一约束。
当使用 Insert.on_conflict_do_update.index_elements 来推断索引时，还可以通过指定 Insert.on_conflict_do_update.index_where 参数推断出部分索引：

>>> stmt = insert(my_table).values(user_email='a@b.com', data='inserted data')
>>> do_update_stmt = stmt.on_conflict_do_update(
...     index_elements=[my_table.c.user_email],
...     index_where=my_table.c.user_email.like('%@gmail.com'),
...     set_=dict(data=stmt.excluded.data)
...     )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (data,  user_email)  VALUES  (?,  ?)
ON  CONFLICT  (user_email)
WHERE  user_email  LIKE  '%@gmail.com'
DO  UPDATE  SET  data  =  excluded.data

SET 子句

使用 ON CONFLICT...DO UPDATE 来执行已经存在行的更新，使用任何组合的新值以及来自所提议插入的值。这些值使用 Insert.on_conflict_do_update.set_ 参数指定。该参数接受一个包含直接 UPDATE 值的字典：

>>> stmt = insert(my_table).values(id='some_id', data='inserted value')
>>> do_update_stmt = stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value')
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (?,  ?)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  ?

警告

Insert.on_conflict_do_update() 方法不考虑 Python 端的默认 UPDATE 值或生成函数，例如，使用 Column.onupdate 指定的那些。除非在 Insert.on_conflict_do_update.set_ 字典中手动指定，否则这些值不会在 ON CONFLICT 类型的 UPDATE 中使用。

使用被排除的 INSERT 值进行更新

为了引用所提议的插入行，特殊别名 Insert.excluded 可以作为 Insert 对象的属性使用；该对象在列上创建了一个 “excluded.” 前缀，它通知 DO UPDATE 使用将插入的值更新行，如果约束没有失败的话将会插入的值：

>>> stmt = insert(my_table).values(
...     id='some_id',
...     data='inserted value',
...     author='jlh'
... )
>>> do_update_stmt = stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value', author=stmt.excluded.author)
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data,  author)  VALUES  (?,  ?,  ?)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  ?,  author  =  excluded.author

附加的 WHERE 条件

Insert.on_conflict_do_update() 方法还接受使用 Insert.on_conflict_do_update.where 参数的 WHERE 子句，这将限制接收 UPDATE 的行：

>>> stmt = insert(my_table).values(
...     id='some_id',
...     data='inserted value',
...     author='jlh'
... )
>>> on_update_stmt = stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value', author=stmt.excluded.author),
...     where=(my_table.c.status == 2)
... )
>>> print(on_update_stmt)
INSERT  INTO  my_table  (id,  data,  author)  VALUES  (?,  ?,  ?)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  ?,  author  =  excluded.author
WHERE  my_table.status  =  ?

使用 DO NOTHING 跳过行

ON CONFLICT 可以用来完全跳过插入行，如果任何与唯一约束发生冲突的话；下面通过使用 Insert.on_conflict_do_nothing() 方法进行了说明：

>>> stmt = insert(my_table).values(id='some_id', data='inserted value')
>>> stmt = stmt.on_conflict_do_nothing(index_elements=['id'])
>>> print(stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (?,  ?)  ON  CONFLICT  (id)  DO  NOTHING

如果使用 DO NOTHING 而没有指定任何列或约束，则会跳过发生的任何唯一性冲突的 INSERT：

>>> stmt = insert(my_table).values(id='some_id', data='inserted value')
>>> stmt = stmt.on_conflict_do_nothing()
>>> print(stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (?,  ?)  ON  CONFLICT  DO  NOTHING

指定目标

两种方法都使用列推断提供冲突的 “目标”：

Insert.on_conflict_do_update.index_elements 参数指定一个序列，包含字符串列名、Column 对象和/或 SQL 表达式元素，用于标识唯一索引或唯一约束。
当使用 Insert.on_conflict_do_update.index_elements 推断索引时，还可以通过指定 Insert.on_conflict_do_update.index_where 参数来推断部分索引。

>>> stmt = insert(my_table).values(user_email='a@b.com', data='inserted data')
>>> do_update_stmt = stmt.on_conflict_do_update(
...     index_elements=[my_table.c.user_email],
...     index_where=my_table.c.user_email.like('%@gmail.com'),
...     set_=dict(data=stmt.excluded.data)
...     )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (data,  user_email)  VALUES  (?,  ?)
ON  CONFLICT  (user_email)
WHERE  user_email  LIKE  '%@gmail.com'
DO  UPDATE  SET  data  =  excluded.data

SET 子句

ON CONFLICT...DO UPDATE 用于对已存在的行进行更新，可以使用新值与插入提议中的任意组合值。这些值使用 Insert.on_conflict_do_update.set_ 参数指定。此参数接受一个字典，其中包含 UPDATE 的直接值：

>>> stmt = insert(my_table).values(id='some_id', data='inserted value')
>>> do_update_stmt = stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value')
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (?,  ?)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  ?

警告

Insert.on_conflict_do_update() 方法不会考虑 Python 端默认的 UPDATE 值或生成函数，例如使用 Column.onupdate 指定的值。除非这些值在 Insert.on_conflict_do_update.set_ 字典中手动指定，否则这些值不会用于 ON CONFLICT 类型的 UPDATE。

使用插入的排除值进行更新

为了引用插入提议的行，特殊别名 Insert.excluded 可作为 Insert 对象的属性使用；此对象在列上创建一个“excluded.”前缀，该前缀告知 DO UPDATE 使用将在约束失败时插入的值更新行：

>>> stmt = insert(my_table).values(
...     id='some_id',
...     data='inserted value',
...     author='jlh'
... )
>>> do_update_stmt = stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value', author=stmt.excluded.author)
... )
>>> print(do_update_stmt)
INSERT  INTO  my_table  (id,  data,  author)  VALUES  (?,  ?,  ?)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  ?,  author  =  excluded.author

附加的 WHERE 条件

Insert.on_conflict_do_update() 方法还接受使用 Insert.on_conflict_do_update.where 参数的 WHERE 子句，这将限制那些接收 UPDATE 的行：

>>> stmt = insert(my_table).values(
...     id='some_id',
...     data='inserted value',
...     author='jlh'
... )
>>> on_update_stmt = stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value', author=stmt.excluded.author),
...     where=(my_table.c.status == 2)
... )
>>> print(on_update_stmt)
INSERT  INTO  my_table  (id,  data,  author)  VALUES  (?,  ?,  ?)
ON  CONFLICT  (id)  DO  UPDATE  SET  data  =  ?,  author  =  excluded.author
WHERE  my_table.status  =  ?

使用 DO NOTHING 跳过行

ON CONFLICT 可以用于完全跳过插入行，如果发生与唯一约束的冲突；以下是使用 Insert.on_conflict_do_nothing() 方法进行说明：

>>> stmt = insert(my_table).values(id='some_id', data='inserted value')
>>> stmt = stmt.on_conflict_do_nothing(index_elements=['id'])
>>> print(stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (?,  ?)  ON  CONFLICT  (id)  DO  NOTHING

如果 DO NOTHING 在不指定任何列或约束的情况下使用，则会跳过发生的任何唯一违规的 INSERT：

>>> stmt = insert(my_table).values(id='some_id', data='inserted value')
>>> stmt = stmt.on_conflict_do_nothing()
>>> print(stmt)
INSERT  INTO  my_table  (id,  data)  VALUES  (?,  ?)  ON  CONFLICT  DO  NOTHING

类型反射

SQLite 类型与大多数其他数据库后端不同，在于类型的字符串名称通常不是一对一地对应于一个“类型”。相反，SQLite 将每列的类型行为链接到五种所谓的“类型亲和性”之一，基于类型的字符串匹配模式。

当 SQLAlchemy 的反射过程检查类型时，它使用一个简单的查找表将返回的关键字链接到提供的 SQLAlchemy 类型。这个查找表存在于 SQLite 方言中，就像存在于所有其他方言中一样。然而，当某个特定类型名称未在查找映射中找到时，SQLite 方言有一个不同的“回退”例程；它实现了位于 www.sqlite.org/datatype3.html 第 2.1 节的 SQLite “类型亲和性”方案。

提供的类型映射将直接从以下类型的精确字符串名称匹配进行关联：

BIGINT、BLOB、BOOLEAN、BOOLEAN、CHAR、DATE、DATETIME、FLOAT、DECIMAL、FLOAT、INTEGER、INTEGER、NUMERIC、REAL、SMALLINT、TEXT、TIME、TIMESTAMP、VARCHAR、NVARCHAR、NCHAR

当类型名称与上述类型之一不匹配时，将使用“类型亲和性”查找：

如果类型名称包含字符串INT，则返回INTEGER类型。
如果类型名称包含字符串CHAR、CLOB或TEXT，则返回TEXT类型。
如果类型名称包含字符串BLOB，则返回NullType类型。
如果类型名称包含字符串REAL、FLOA或DOUB，则返回REAL类型。
否则，使用NUMERIC类型。

部分索引

可以使用 DDL 系统指定带有 WHERE 子句的部分索引，例如使用参数sqlite_where：

tbl = Table('testtbl', m, Column('data', Integer))
idx = Index('test_idx1', tbl.c.data,
            sqlite_where=and_(tbl.c.data > 5, tbl.c.data < 10))

在创建时，索引将被渲染为：

CREATE INDEX test_idx1 ON testtbl (data)
WHERE data > 5 AND data < 10

点分列名

不推荐使用显式带有句点的表名或列名。虽然这对于关系数据库来说通常是个坏主意，因为句点是一个语法上重要的字符，但 SQLite 驱动在 SQLite 版本 3.10.0 之前存在一个 bug，要求 SQLAlchemy 在结果集中滤掉这些句点。

这个 bug 完全不在 SQLAlchemy 的范围之内，可以用以下方式加以说明：

import sqlite3
assert sqlite3.sqlite_version_info < (3, 10, 0), "bug is fixed in this version"
conn = sqlite3.connect(":memory:")
cursor = conn.cursor()
cursor.execute("create table x (a integer, b integer)")
cursor.execute("insert into x (a, b) values (1, 1)")
cursor.execute("insert into x (a, b) values (2, 2)")
cursor.execute("select x.a, x.b from x")
assert [c[0] for c in cursor.description] == ['a', 'b']
cursor.execute('''
 select x.a, x.b from x where a=1
 union
 select x.a, x.b from x where a=2
''')
assert [c[0] for c in cursor.description] == ['a', 'b'], \
    [c[0] for c in cursor.description]

第二个断言失败：

Traceback (most recent call last):
  File "test.py", line 19, in <module>
    [c[0] for c in cursor.description]
AssertionError: ['x.a', 'x.b']

在上述情况下，驱动程序错误地报告包括表名在内的列名，这与 UNION 不在时完全不一致。

SQLAlchemy 依赖于列名在匹配原始语句时的可预测性，因此 SQLAlchemy 方言别无选择，只能将这些列名滤除：

from sqlalchemy import create_engine
eng = create_engine("sqlite://")
conn = eng.connect()
conn.exec_driver_sql("create table x (a integer, b integer)")
conn.exec_driver_sql("insert into x (a, b) values (1, 1)")
conn.exec_driver_sql("insert into x (a, b) values (2, 2)")
result = conn.exec_driver_sql("select x.a, x.b from x")
assert result.keys() == ["a", "b"]
result = conn.exec_driver_sql('''
 select x.a, x.b from x where a=1
 union
 select x.a, x.b from x where a=2
''')
assert result.keys() == ["a", "b"]

请注意，尽管 SQLAlchemy 过滤掉了句点，这两个名称仍然是可寻址的：

>>> row = result.first()
>>> row["a"]
1
>>> row["x.a"]
1
>>> row["b"]
1
>>> row["x.b"]
1

因此，SQLAlchemy 应用的解决方法只会影响公共 API 中的 CursorResult.keys() 和 Row.keys()。在强制使用包含句点的列名，并且需要 CursorResult.keys() 和 Row.keys() 返回这些带点的名称不经修改的非常特殊情况下，可以在每个 Connection 上提供 sqlite_raw_colnames 执行选项：

result = conn.execution_options(sqlite_raw_colnames=True).exec_driver_sql('''
 select x.a, x.b from x where a=1
 union
 select x.a, x.b from x where a=2
''')
assert result.keys() == ["x.a", "x.b"]

或者在每个 Engine 上：

engine = create_engine("sqlite://", execution_options={"sqlite_raw_colnames": True})

使用每个 Engine 的执行选项时，请注意使用 UNION 的 Core 和 ORM 查询可能无法正常工作。

SQLite 特定的表选项

CREATE TABLE 的一种选项直接由 SQLite 方言支持，与 Table 结构配合使用：

WITHOUT ROWID：

Table("some_table", metadata, ..., sqlite_with_rowid=False)

另请参阅

SQLite CREATE TABLE 选项

反映内部模式表

返回表列表的反射方法将省略所谓的“SQLite 内部模式对象”名称，这些对象被 SQLite 视为任何以 sqlite_ 为前缀的对象名称。这种对象的示例是在使用 AUTOINCREMENT 列参数时生成的 sqlite_sequence 表。为了返回这些对象，可以向诸如 MetaData.reflect() 或 Inspector.get_table_names() 这样的方法传递参数 sqlite_include_internal=True。

新版本 2.0 中新增了 sqlite_include_internal=True 参数。以前，这些表不被 SQLAlchemy 反射方法忽略。

注意

sqlite_include_internal 参数不引用存在于 sqlite_master 等模式中的 “系统” 表。

另请参阅

SQLite 内部模式对象 - SQLite 文档中。

SqlAlchemy 2.0 中文文档（五十）（5）https://developer.aliyun.com/article/1563131

SqlAlchemy 2.0 中文文档（五十）（4）

保存点支持

事务性 DDL

外键支持

对约束的 ON CONFLICT 支持

插入…冲突时执行（Upsert）

指定目标

SET 子句

使用被排除的 INSERT 值进行更新

附加的 WHERE 条件

使用 DO NOTHING 跳过行

指定目标

SET 子句

使用插入的排除值进行更新

附加的 WHERE 条件

使用 DO NOTHING 跳过行

类型反射

部分索引

点分列名

SQLite 特定的表选项

反映内部模式表

热门文章

最新文章

相关电子书