SQLite¶

对 SQLite 数据库的支持。

下表总结了当前数据库发布版本的支持级别。

**支持的 SQLite 版本**¶
支持类型	版本
受支持版本	3.12+
尽力而为	3.7.16+

DBAPI 支持¶

以下方言/DBAPI 选项可用。有关连接信息，请参阅各个 DBAPI 部分。

pysqlite
aiosqlite
pysqlcipher

日期和时间类型¶

SQLite 没有内置的 DATE、TIME 或 DATETIME 类型，并且 pysqlite 没有提供开箱即用的功能来在 Python datetime 对象和 SQLite 支持的格式之间转换值。当使用 SQLite 时，SQLAlchemy 自己的 DateTime 和相关类型提供了日期格式化和解析功能。实现类是 DATETIME、DATE 和 TIME。这些类型将日期和时间表示为 ISO 格式的字符串，这也很好地支持排序。这些函数不依赖于典型的 “libc” 内部结构，因此完全支持历史日期。

确保文本亲和性¶

为这些类型呈现的 DDL 是标准的 DATE、TIME 和 DATETIME 指示符。但是，自定义存储格式也可以应用于这些类型。当检测到存储格式不包含字母字符时，这些类型的 DDL 将呈现为 DATE_CHAR、TIME_CHAR 和 DATETIME_CHAR，以便列继续具有文本亲和性。

另请参阅

类型亲和性 - 在 SQLite 文档中

SQLite 自动递增行为¶

有关 SQLite 自动递增的背景信息，请访问： https://sqlite.ac.cn/autoinc.html

关键概念

SQLite 具有隐式 “自动递增” 功能，该功能适用于任何非复合主键列，该列专门使用 “INTEGER PRIMARY KEY” 为类型 + 主键创建。
SQLite 还有一个显式 “AUTOINCREMENT” 关键字，它不等同于隐式自动递增功能；不建议通常使用此关键字。除非使用特定的 SQLite 特定指令（见下文），否则 SQLAlchemy 不会呈现此关键字。但是，它仍然要求列的类型名为 “INTEGER”。

使用 AUTOINCREMENT 关键字¶

要在呈现 DDL 时在主键列上专门呈现 AUTOINCREMENT 关键字，请将标志 sqlite_autoincrement=True 添加到 Table 构造中

Table(
    "sometable",
    metadata,
    Column("id", Integer, primary_key=True),
    sqlite_autoincrement=True,
)

允许除 Integer/INTEGER 之外的 SQLAlchemy 类型使用自动递增行为¶

SQLite 的类型模型基于命名约定。除此之外，这意味着任何包含子字符串 "INT" 的类型名称都将被确定为具有 “integer affinity”（整数亲和性）。名为 "BIGINT"、"SPECIAL_INT" 甚至 "XYZINTQPR" 的类型都将被 SQLite 视为具有 “integer” 亲和性。但是，SQLite 自动递增功能，无论是隐式还是显式启用，都要求列的类型名称必须完全是字符串 “INTEGER”。因此，如果应用程序对主键使用诸如 BigInteger 之类的类型，则在 SQLite 上，此类型需要呈现为名称 "INTEGER"，以便在发出初始 CREATE TABLE 语句时可以使用自动递增行为。

实现此目的的一种方法是在仅在 SQLite 上使用 Integer，方法是使用 TypeEngine.with_variant()

table = Table(
    "my_table",
    metadata,
    Column(
        "id",
        BigInteger().with_variant(Integer, "sqlite"),
        primary_key=True,
    ),
)

另一种方法是使用 BigInteger 的子类，该子类覆盖其 DDL 名称，使其在针对 SQLite 编译时为 INTEGER

from sqlalchemy import BigInteger
from sqlalchemy.ext.compiler import compiles


class SLBigInteger(BigInteger):
    pass


@compiles(SLBigInteger, "sqlite")
def bi_c(element, compiler, **kw):
    return "INTEGER"


@compiles(SLBigInteger)
def bi_c(element, compiler, **kw):
    return compiler.visit_BIGINT(element, **kw)


table = Table(
    "my_table", metadata, Column("id", SLBigInteger(), primary_key=True)
)

另请参阅

TypeEngine.with_variant()

自定义 SQL 构造和编译扩展

SQLite 版本 3 中的数据类型

数据库锁定行为 / 并发¶

SQLite 并非设计用于高水平的写入并发。数据库本身（作为一个文件）在事务内的写入操作期间被完全锁定，这意味着在此期间只有一个 “连接”（实际上是文件句柄）具有对数据库的独占访问权 - 所有其他 “连接” 将在此期间被阻止。

Python DBAPI 规范还要求连接模型始终处于事务中；没有 connection.begin() 方法，只有 connection.commit() 和 connection.rollback()，之后应立即开始新的事务。这似乎意味着理论上 SQLite 驱动程序在任何时候只允许在特定数据库文件上使用单个文件句柄；但是，SQLite 本身以及 pysqlite 驱动程序中都有几个因素大大放宽了此限制。

但是，无论使用哪种锁定模式，一旦事务启动并且至少已发出 DML（例如 INSERT、UPDATE、DELETE），SQLite 仍将始终锁定数据库文件，这将至少在其他事务也尝试发出 DML 时阻止它们。默认情况下，此阻塞的持续时间非常短，之后会超时并显示错误。

当与 SQLAlchemy ORM 结合使用时，此行为变得更加关键。SQLAlchemy 的 Session 对象默认在事务中运行，并且由于其自动刷新模型，可能会在任何 SELECT 语句之前发出 DML。这可能会导致 SQLite 数据库比预期更快地锁定。可以一定程度上操纵 SQLite 和 pysqlite 驱动程序的锁定模式，但应该注意的是，使用 SQLite 实现高程度的写入并发是一场注定失败的战斗。

有关 SQLite 在设计上缺乏写入并发的更多信息，请参阅页面底部的在哪些情况下其他 RDBMS 可能工作得更好 - 高并发。

以下小节介绍了受 SQLite 基于文件的架构影响的区域，并且通常在使用 pysqlite 驱动程序时也需要解决方法才能工作。

事务隔离级别 / 自动提交¶

SQLite 以非标准方式支持 “事务隔离”，沿着两个轴。一个是 PRAGMA read_uncommitted 指令。此设置可以基本上在 SQLite 的默认 SERIALIZABLE 隔离模式和通常称为 READ UNCOMMITTED 的 “脏读” 隔离模式之间切换 SQLite。

SQLAlchemy 使用 create_engine.isolation_level 参数连接到此 PRAGMA 语句，该参数是 create_engine() 的参数。与 SQLite 一起使用时，此参数的有效值为 "SERIALIZABLE" 和 "READ UNCOMMITTED"，分别对应于值 0 和 1。SQLite 默认为 SERIALIZABLE，但其行为受 pysqlite 驱动程序的默认行为影响。

当使用 pysqlite 驱动程序时，"AUTOCOMMIT" 隔离级别也可用，这将使用 DBAPI 连接上的 .isolation_level 属性更改 pysqlite 连接，并将其设置为 None 以保持设置的持续时间。

1.3.16 版本新增: 在使用 pysqlite / sqlite3 SQLite 驱动程序时，添加了对 SQLite AUTOCOMMIT 隔离级别的支持。

SQLite 事务锁定的另一个轴是通过使用的 BEGIN 语句的性质。三种变体是 “deferred”、“immediate” 和 “exclusive”，如 BEGIN TRANSACTION 中所述。直 BEGIN 语句使用 “deferred” 模式，其中数据库文件在第一次读取或写入操作之前不会被锁定，并且读取访问权限在第一次写入操作之前对其他事务保持开放。但同样，关键要注意的是，pysqlite 驱动程序通过甚至在第一次写入操作之前都不发出 BEGIN 来干扰此行为。

警告

SQLite 的事务范围受到 pysqlite 驱动程序中未解决问题的影响，这些问题将 BEGIN 语句推迟到通常不可行的程度。有关解决此行为的技术，请参阅可串行化隔离 / 保存点 / 事务性 DDL 或可串行化隔离 / 保存点 / 事务性 DDL (asyncio 版本) 部分。

另请参阅

设置事务隔离级别，包括 DBAPI 自动提交

INSERT/UPDATE/DELETE…RETURNING¶

SQLite 方言支持 SQLite 3.35 的 INSERT|UPDATE|DELETE..RETURNING 语法。在某些情况下，可能会自动使用 INSERT..RETURNING 来获取新生成的标识符，以代替使用 cursor.lastrowid 的传统方法，但是，对于简单的单语句情况，目前仍然首选 cursor.lastrowid，因为它具有更好的性能。

要指定显式的 RETURNING 子句，请在每个语句的基础上使用 _UpdateBase.returning() 方法

# INSERT..RETURNING
result = connection.execute(
    table.insert().values(name="foo").returning(table.c.col1, table.c.col2)
)
print(result.all())

# UPDATE..RETURNING
result = connection.execute(
    table.update()
    .where(table.c.name == "foo")
    .values(name="bar")
    .returning(table.c.col1, table.c.col2)
)
print(result.all())

# DELETE..RETURNING
result = connection.execute(
    table.delete()
    .where(table.c.name == "foo")
    .returning(table.c.col1, table.c.col2)
)
print(result.all())

2.0 版本新增: 添加了对 SQLite RETURNING 的支持

SAVEPOINT 支持¶

SQLite 支持 SAVEPOINT，它仅在事务开始后才起作用。SQLAlchemy 的 SAVEPOINT 支持可通过核心级别的 Connection.begin_nested() 方法和 ORM 级别的 Session.begin_nested() 方法获得。但是，除非采取解决方法，否则 SAVEPOINT 根本无法与 pysqlite 一起使用。

警告

SQLite 的 SAVEPOINT 功能受到 pysqlite 和 aiosqlite 驱动程序中未解决问题的影响，这些问题将 BEGIN 语句推迟到通常不可行的程度。有关解决此行为的技术，请参阅可串行化隔离 / 保存点 / 事务性 DDL 和可串行化隔离 / 保存点 / 事务性 DDL (asyncio 版本) 部分。

事务性 DDL¶

SQLite 数据库也支持事务性 DDL。在这种情况下，pysqlite 驱动程序不仅未能启动事务，而且还在检测到 DDL 时结束任何现有事务，因此再次需要解决方法。

警告

SQLite 的事务性 DDL 受到 pysqlite 驱动程序中未解决问题的影响，该驱动程序无法发出 BEGIN，并且在遇到 DDL 时还会强制执行 COMMIT 以取消任何事务。有关解决此行为的技术，请参阅可串行化隔离 / 保存点 / 事务性 DDL 部分。

外键支持¶

SQLite 在发出表的 CREATE 语句时支持 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 对象指示的三种相关约束类型。

另请参阅

ON CONFLICT - 在 SQLite 文档中

1.3 版本新增。

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

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 标志向单列添加 UNIQUE 约束时，也可以将 sqlite_on_conflict_unique 参数添加到 Column 中，这将添加到 DDL 中的 UNIQUE 约束中

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
)

要为 NOT NULL 约束应用 FAIL 算法，请使用 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 单独呈现 PRIMARY KEY 约束，因此冲突解决算法应用于约束本身

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

INSERT…ON CONFLICT (Upsert)¶

另请参阅

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

从 3.24.0 版本开始，SQLite 支持通过 INSERT 语句的 ON CONFLICT 子句将行 “upsert”（更新或插入）到表中。只有当候选行不违反任何唯一约束或主键约束时，才会插入该行。如果发生唯一约束冲突，则可能发生辅助操作，该操作可以是 “DO UPDATE”，表示应更新目标行中的数据，也可以是 “DO NOTHING”，表示静默跳过此行。

冲突是使用作为现有唯一约束和索引一部分的列来确定的。这些约束通过声明构成索引的列和条件来识别。

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

>>> fromsqlalchemy.dialects.sqliteimportinsert>>> 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 文档中。

指定目标¶

两种方法都使用列推断来提供冲突的 “target”（目标）

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 指定的那些值。这些值不会为 ON CONFLICT 样式的 UPDATE 执行，除非它们在 Insert.on_conflict_do_update.set_ 字典中手动指定。

使用排除的 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

类型反射¶

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

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

提供的类型映射将直接关联以下类型的确切字符串名称匹配

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 系统和参数 sqlite_where 来指定部分索引，例如，使用 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 都有一个错误，需要 SQLAlchemy 在结果集中过滤掉这些点。

这个完全在 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() 的功能来返回这些未修改的带点名称，则可以提供 sqlite_raw_colnames 执行选项，可以在每个 Connection 的基础上提供

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 特定的表选项¶

SQLite 方言直接支持 CREATE TABLE 的一个选项，与 Table 构造结合使用

WITHOUT ROWID:

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

STRICT:

Table("some_table", metadata, ..., sqlite_strict=True)

2.0.37 版本新增。

另请参阅

SQLite CREATE TABLE 选项

反射内部模式表¶

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

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

注意

sqlite_include_internal 参数不指模式（如 sqlite_master）中存在的“系统”表。

另请参阅

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

SQLite 数据类型¶

与所有 SQLAlchemy 方言一样，所有已知对 SQLite 有效的大写类型都可以从顶级方言导入，无论它们是源自 sqlalchemy.types 还是来自本地方言

from sqlalchemy.dialects.sqlite import (
    BLOB,
    BOOLEAN,
    CHAR,
    DATE,
    DATETIME,
    DECIMAL,
    FLOAT,
    INTEGER,
    NUMERIC,
    JSON,
    SMALLINT,
    TEXT,
    TIME,
    TIMESTAMP,
    VARCHAR,
)

对象名称	描述
DATE	在 SQLite 中使用字符串表示 Python date 对象。
DATETIME	在 SQLite 中使用字符串表示 Python datetime 对象。
JSON	SQLite JSON 类型。
TIME	在 SQLite 中使用字符串表示 Python time 对象。

class sqlalchemy.dialects.sqlite.DATETIME¶

在 SQLite 中使用字符串表示 Python datetime 对象。

默认的字符串存储格式是

"%(year)04d-%(month)02d-%(day)02d %(hour)02d:%(minute)02d:%(second)02d.%(microsecond)06d"

例如

2021-03-15 12:05:57.105542

默认情况下，传入的存储格式使用 Python datetime.fromisoformat() 函数解析。

2.0 版本更改: datetime.fromisoformat() 用于默认的 datetime 字符串解析。

可以使用 storage_format 和 regexp 参数在一定程度上自定义存储格式，例如

import re
from sqlalchemy.dialects.sqlite import DATETIME

dt = DATETIME(
    storage_format=(
        "%(year)04d/%(month)02d/%(day)02d %(hour)02d:%(minute)02d:%(second)02d"
    ),
    regexp=r"(\d+)/(\d+)/(\d+) (\d+)-(\d+)-(\d+)",
)

参数:

storage_format¶ – 格式字符串，将应用于包含 year、month、day、hour、minute、second 和 microsecond 键的字典。
regexp¶ – 正则表达式，将应用于传入的结果行，替换使用 datetime.fromisoformat() 来解析传入的字符串。如果正则表达式包含命名组，则生成的匹配字典将作为关键字参数应用于 Python datetime() 构造函数。否则，如果使用位置组，则通过 *map(int, match_obj.groups(0)) 使用位置参数调用 datetime() 构造函数。

类签名

class sqlalchemy.dialects.sqlite.DATETIME (sqlalchemy.dialects.sqlite.base._DateTimeMixin, sqlalchemy.types.DateTime)

class sqlalchemy.dialects.sqlite.DATE¶

在 SQLite 中使用字符串表示 Python date 对象。

默认的字符串存储格式是

"%(year)04d-%(month)02d-%(day)02d"

例如

2011-03-15

默认情况下，传入的存储格式使用 Python date.fromisoformat() 函数解析。

2.0 版本更改: date.fromisoformat() 用于默认的 date 字符串解析。

可以使用 storage_format 和 regexp 参数在一定程度上自定义存储格式，例如

import re
from sqlalchemy.dialects.sqlite import DATE

d = DATE(
    storage_format="%(month)02d/%(day)02d/%(year)04d",
    regexp=re.compile("(?P<month>\d+)/(?P<day>\d+)/(?P<year>\d+)"),
)

参数:

storage_format¶ – 格式字符串，将应用于包含 year、month 和 day 键的字典。
regexp¶ – 正则表达式，将应用于传入的结果行，替换使用 date.fromisoformat() 来解析传入的字符串。如果正则表达式包含命名组，则生成的匹配字典将作为关键字参数应用于 Python date() 构造函数。否则，如果使用位置组，则通过 *map(int, match_obj.groups(0)) 使用位置参数调用 date() 构造函数。

类签名

class sqlalchemy.dialects.sqlite.DATE (sqlalchemy.dialects.sqlite.base._DateTimeMixin, sqlalchemy.types.Date)

class sqlalchemy.dialects.sqlite.JSON¶

SQLite JSON 类型。

SQLite 从 3.9 版本开始通过其 JSON1 扩展支持 JSON。请注意，JSON1 是一个可加载的扩展，因此可能不可用，或者可能需要运行时加载。

每当对 SQLite 后端使用基本 JSON 数据类型时，都会自动使用 JSON。

另请参阅

JSON - 通用跨平台 JSON 数据类型的主要文档。

JSON 类型支持 JSON 值的持久化以及 JSON 数据类型提供的核心索引操作，通过调整操作以在数据库级别呈现包装在 JSON_QUOTE 函数中的 JSON_EXTRACT 函数。提取的值被引用，以确保结果始终是 JSON 字符串值。

1.3 版本新增。

成员

__init__()

类签名

class sqlalchemy.dialects.sqlite.JSON (sqlalchemy.types.JSON)

method sqlalchemy.dialects.sqlite.JSON.__init__(none_as_null: bool = False)¶

继承自 sqlalchemy.types.JSON.__init__ 方法，属于 JSON

构造一个 JSON 类型。

参数:

none_as_null=False¶ –

如果为 True，则将值 None 持久化为 SQL NULL 值，而不是 null 的 JSON 编码。请注意，当此标志为 False 时，null() 构造仍然可以用于持久化 NULL 值，该值可以作为参数值直接传递，由 JSON 类型特别解释为 SQL NULL

from sqlalchemy import null

conn.execute(table.insert(), {"data": null()})

注意

JSON.none_as_null 不适用于传递给 Column.default 和 Column.server_default 的值；为这些参数传递 None 值表示“不存在默认值”。

此外，当在 SQL 比较表达式中使用时，Python 值 None 继续引用 SQL null，而不是 JSON NULL。JSON.none_as_null 标志明确指的是 INSERT 或 UPDATE 语句中值的持久化。JSON.NULL 值应用于希望与 JSON null 进行比较的 SQL 表达式。

另请参阅

JSON.NULL

class sqlalchemy.dialects.sqlite.TIME¶

在 SQLite 中使用字符串表示 Python time 对象。

默认的字符串存储格式是

"%(hour)02d:%(minute)02d:%(second)02d.%(microsecond)06d"

例如

12:05:57.10558

默认情况下，传入的存储格式使用 Python time.fromisoformat() 函数解析。

2.0 版本更改: time.fromisoformat() 用于默认的 time 字符串解析。

可以使用 storage_format 和 regexp 参数在一定程度上自定义存储格式，例如

import re
from sqlalchemy.dialects.sqlite import TIME

t = TIME(
    storage_format="%(hour)02d-%(minute)02d-%(second)02d-%(microsecond)06d",
    regexp=re.compile("(\d+)-(\d+)-(\d+)-(?:-(\d+))?"),
)

参数:

storage_format¶ – 格式字符串，将应用于包含 hour、minute、second 和 microsecond 键的字典。
regexp¶ – 正则表达式，将应用于传入的结果行，替换使用 datetime.fromisoformat() 来解析传入的字符串。如果正则表达式包含命名组，则生成的匹配字典将作为关键字参数应用于 Python time() 构造函数。否则，如果使用位置组，则通过 *map(int, match_obj.groups(0)) 使用位置参数调用 time() 构造函数。

类签名

class sqlalchemy.dialects.sqlite.TIME (sqlalchemy.dialects.sqlite.base._DateTimeMixin, sqlalchemy.types.Time)

SQLite DML 构造¶

对象名称	描述
insert(table)	构造一个 SQLite 特定的 `Insert` 构造。
Insert	SQLite 特定的 INSERT 实现。

function sqlalchemy.dialects.sqlite.insert(table: _DMLTableArgument) → Insert¶

构造一个 SQLite 特定的 Insert 构造。

sqlalchemy.dialects.sqlite.insert() 函数创建一个 sqlalchemy.dialects.sqlite.Insert。此类基于与方言无关的 Insert 构造，该构造可以使用 SQLAlchemy Core 中的 insert() 函数构造。

Insert 构造包含附加方法 Insert.on_conflict_do_update(), Insert.on_conflict_do_nothing()。

class sqlalchemy.dialects.sqlite.Insert¶

SQLite 特定的 INSERT 实现。

为 SQLite 特定的语法（如 ON CONFLICT）添加方法。

Insert 对象是使用 sqlalchemy.dialects.sqlite.insert() 函数创建的。

1.4 版本新增。

另请参阅

INSERT…ON CONFLICT (Upsert)

成员

excluded, inherit_cache, on_conflict_do_nothing(), on_conflict_do_update()

类签名

class sqlalchemy.dialects.sqlite.Insert (sqlalchemy.sql.expression.Insert)

attribute sqlalchemy.dialects.sqlite.Insert.excluded¶

为 ON CONFLICT 语句提供 excluded 命名空间

SQLite 的 ON CONFLICT 子句允许引用将要插入的行，称为 excluded。此属性提供此行中的所有列以供引用。

提示

Insert.excluded 属性是 ColumnCollection 的一个实例，它提供的接口与 Table.c 集合（在访问表和列中描述）相同。使用此集合，普通名称可以像属性一样访问（例如，stmt.excluded.some_column），但特殊名称和字典方法名称应使用索引访问，例如 stmt.excluded["column name"] 或 stmt.excluded["values"]。有关更多示例，请参阅 ColumnCollection 的文档字符串。

attribute sqlalchemy.dialects.sqlite.Insert.inherit_cache: bool | None = False¶

指示此 HasCacheKey 实例是否应使用其直接超类使用的缓存键生成方案。

该属性默认为 None，表示构造尚未考虑其是否适合参与缓存；这在功能上等同于将值设置为 False，但也会发出警告。

如果与对象对应的 SQL 不会基于此类的局部属性（而不是其超类）而更改，则可以在特定类上将此标志设置为 True。

另请参阅

为自定义构造启用缓存支持 - 设置第三方或用户定义的 SQL 构造的 HasCacheKey.inherit_cache 属性的通用指南。

method sqlalchemy.dialects.sqlite.Insert.on_conflict_do_nothing(index_elements: Iterable[Column[Any] | str | DDLConstraintColumnRole] | None = None, index_where: WhereHavingRole | None = None) → Self¶

为 ON CONFLICT 子句指定 DO NOTHING 操作。

参数:

index_elements¶ – 一个序列，由字符串列名、Column 对象或其他列表达式对象组成，这些对象将用于推断目标索引或唯一约束。
index_where¶ – 可用于推断条件目标索引的附加 WHERE 条件。

指定 ON CONFLICT 子句的 DO UPDATE SET 操作。

参数:

index_elements¶ – 由字符串列名、Column 对象或其他列表达式对象组成的序列，这些对象将用于推断目标索引或唯一约束。
index_where¶ – 可用于推断条件目标索引的附加 WHERE 标准。
set_¶ –
字典或其他映射对象，其中键是目标表中的列名，或者是与目标表匹配的 Column 对象或其他 ORM 映射列，值是表达式或字面量，用于指定要执行的 SET 操作。

1.4 版本新增: Insert.on_conflict_do_update.set_ 参数支持来自目标 Table 的 Column 对象作为键。

警告

此字典不考虑 Python 指定的默认 UPDATE 值或生成函数，例如使用 Column.onupdate 指定的值。这些值不会为 ON CONFLICT 样式的 UPDATE 执行，除非它们在 Insert.on_conflict_do_update.set_ 字典中手动指定。
where¶ – 可选参数。表示 WHERE 子句的表达式对象，该子句限制受 DO UPDATE SET 影响的行。不满足 WHERE 条件的行将不会被更新（对于这些行，实际上是 DO NOTHING）。

Pysqlite¶

通过 pysqlite 驱动程序支持 SQLite 数据库。

请注意，pysqlite 与 Python 发行版中包含的 sqlite3 模块是同一个驱动程序。

DBAPI¶

有关 pysqlite 的文档和下载信息（如果适用）可在以下网址获取：https://docs.pythonlang.cn/library/sqlite3.html

连接¶

连接字符串

sqlite+pysqlite:///file_path

驱动程序¶

sqlite3 Python DBAPI 是所有现代 Python 版本上的标准；对于 cPython 和 Pypy，无需额外安装。

连接字符串¶

SQLite 数据库的文件规范被视为 URL 的“database”部分。请注意，SQLAlchemy url 的格式为

driver://user:pass@host/database

这意味着要使用的实际文件名以第三个斜杠右侧的字符开头。因此，连接到相对文件路径如下所示

# relative path
e = create_engine("sqlite:///path/to/database.db")

绝对路径以斜杠开头表示，这意味着您需要四个斜杠

# absolute path
e = create_engine("sqlite:////path/to/database.db")

要使用 Windows 路径，可以使用常规驱动器规范和反斜杠。可能需要双反斜杠

# absolute path on Windows
e = create_engine("sqlite:///C:\\path\\to\\database.db")

要使用 sqlite :memory: 数据库，请将其指定为文件名，使用 sqlite:///:memory:。如果不存在文件路径，这也是默认设置，仅指定 sqlite:// 而不指定其他内容

# in-memory database (note three slashes)
e = create_engine("sqlite:///:memory:")
# also in-memory database
e2 = create_engine("sqlite://")

URI 连接¶

现代版本的 SQLite 支持使用驱动程序级别 URI 进行连接的替代系统，该系统具有可以传递其他驱动程序级别参数的优势，包括“只读”等选项。Python sqlite3 驱动程序在现代 Python 3 版本下支持此模式。SQLAlchemy pysqlite 驱动程序通过在 URL 查询字符串中指定“uri=true”来支持此使用模式。SQLite 级别的“URI”保留为 SQLAlchemy url 的“database”部分（即，在斜杠之后）

e = create_engine("sqlite:///file:path/to/database?mode=ro&uri=true")

注意

“uri=true”参数必须出现在 URL 的查询字符串中。如果仅存在于 create_engine.connect_args 参数字典中，则当前无法按预期工作。

该逻辑通过分离属于 Python sqlite3 驱动程序的参数与属于 SQLite URI 的参数来调和 SQLAlchemy 查询字符串和 SQLite 查询字符串的共存。这是通过使用已知 Python 端驱动程序接受的固定参数列表来实现的。例如，要包含指示 Python sqlite3 “timeout”和 “check_same_thread” 参数以及 SQLite “mode” 和 “nolock” 参数的 URL，它们可以全部在查询字符串中一起传递

e = create_engine(
    "sqlite:///file:path/to/database?"
    "check_same_thread=true&timeout=10&mode=ro&nolock=1&uri=true"
)

在上面，pysqlite / sqlite3 DBAPI 将作为参数传递

sqlite3.connect(
    "file:path/to/database?mode=ro&nolock=1",
    check_same_thread=True,
    timeout=10,
    uri=True,
)

关于未来添加到 Python 或本机驱动程序的参数。添加到 SQLite URI 方案的新参数名称应自动适应此方案。添加到 Python 驱动程序端的新参数名称可以通过在 create_engine.connect_args 字典中指定它们来容纳，直到 SQLAlchemy 添加方言支持为止。对于本机 SQLite 驱动程序添加与现有已知 Python 驱动程序参数之一（例如 “timeout”）重叠的新参数名称的较不可能情况，SQLAlchemy 的方言将需要调整 URL 方案以继续支持它。

与所有 SQLAlchemy 方言的情况一样，可以通过使用 create_engine.creator 参数在 create_engine() 中绕过整个 “URL” 过程，该参数允许自定义可调用对象直接创建 Python sqlite3 驱动程序级别连接。

1.3.9 版本新增。

另请参阅

统一资源标识符 - 在 SQLite 文档中

正则表达式支持¶

1.4 版本新增。

使用 Python 的 re.search 函数提供对 ColumnOperators.regexp_match() 运算符的支持。SQLite 本身不包含可用的正则表达式运算符；相反，它包含一个未实现的占位符运算符 REGEXP，该运算符调用必须提供的用户定义函数。

SQLAlchemy 的实现利用 pysqlite create_function 钩子，如下所示

def regexp(a, b):
    return re.search(a, b) is not None


sqlite_connection.create_function(
    "regexp",
    2,
    regexp,
)

目前不支持将正则表达式标志作为单独的参数，因为 SQLite 的 REGEXP 运算符不支持这些标志，但是这些标志可以内联包含在正则表达式字符串中。有关详细信息，请参阅 Python 正则表达式。

另请参阅

Python 正则表达式：Python 正则表达式语法的文档。

与 sqlite3 “原生” 日期和日期时间类型的兼容性¶

pysqlite 驱动程序包括 sqlite3.PARSE_DECLTYPES 和 sqlite3.PARSE_COLNAMES 选项，这些选项的效果是将任何显式强制转换为 “date” 或 “timestamp” 的列或表达式都将转换为 Python date 或 datetime 对象。pysqlite 方言提供的日期和日期时间类型目前与这些选项不兼容，因为它们呈现包括微秒的 ISO 日期/日期时间，而 pysqlite 的驱动程序不这样做。此外，SQLAlchemy 目前不会自动呈现自由函数 “current_timestamp” 和 “current_date” 返回本机 datetime/date 类型所需的 “cast” 语法。遗憾的是，pysqlite 不在 cursor.description 中提供标准 DBAPI 类型，这使得 SQLAlchemy 无法在没有昂贵的每行类型检查的情况下动态检测这些类型。

请记住，不建议使用 pysqlite 的解析选项，也不应该在使用 SQLAlchemy 时使用，如果配置 create_engine() 上的 “native_datetime=True”，则可以强制使用 PARSE_DECLTYPES

engine = create_engine(
    "sqlite://",
    connect_args={
        "detect_types": sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES
    },
    native_datetime=True,
)

启用此标志后，DATE 和 TIMESTAMP 类型（但请注意 - 不是 DATETIME 或 TIME 类型...是否仍然感到困惑？）将不执行任何绑定参数或结果处理。“func.current_date()” 的执行将返回一个字符串。“func.current_timestamp()” 注册为在 SQLAlchemy 中返回 DATETIME 类型，因此此函数仍然接收 SQLAlchemy 级别的结果处理。

线程/池行为¶

sqlite3 DBAPI 默认情况下禁止在不是创建连接的线程中使用特定连接。随着 SQLite 的成熟，它在多线程下的行为得到了改进，甚至包括用于在多线程中使用的仅内存数据库的选项。

线程禁止称为 “check same thread”，可以使用 sqlite3 参数 check_same_thread 控制，这将禁用或启用此检查。SQLAlchemy 在此处的默认行为是在使用基于文件的数据库时自动将 check_same_thread 设置为 False，以建立与默认池类 QueuePool 的兼容性。

SQLAlchemy pysqlite DBAPI 根据请求的 SQLite 数据库的种类，以不同的方式建立连接池

当指定 :memory: SQLite 数据库时，方言默认将使用 SingletonThreadPool。此池为每个线程维护一个连接，以便在当前线程内对引擎的所有访问都使用相同的 :memory: 数据库 - 其他线程将访问不同的 :memory: 数据库。check_same_thread 参数默认为 True。
当指定基于文件的数据库时，方言将使用 QueuePool 作为连接源。同时，除非被覆盖，否则 check_same_thread 标志默认设置为 False。

2.0 版本更改: SQLite 文件数据库引擎现在默认使用 QueuePool。以前，使用 NullPool。NullPool 类可以通过 create_engine.poolclass 参数指定来使用。

禁用文件数据库的连接池¶

可以通过为 poolclass() 参数指定 NullPool 实现来禁用基于文件的数据库的池

from sqlalchemy import NullPool

engine = create_engine("sqlite:///myfile.db", poolclass=NullPool)

据观察，由于 NullPool 实现缺乏 QueuePool 实现的连接重用，NullPool 实现对重复签出产生极小的性能开销。但是，如果应用程序遇到文件被锁定的问题，则仍然可能有利于使用此类。

在多线程中使用内存数据库¶

要在多线程场景中使用 :memory: 数据库，必须在线程之间共享相同的连接对象，因为数据库仅存在于该连接的范围内。StaticPool 实现将全局维护单个连接，并且可以将 check_same_thread 标志作为 False 传递给 Pysqlite

from sqlalchemy.pool import StaticPool

engine = create_engine(
    "sqlite://",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)

请注意，在多线程中使用 :memory: 数据库需要较新版本的 SQLite。

将临时表与 SQLite 一起使用¶

由于 SQLite 处理临时表的方式，如果您希望在基于文件的 SQLite 数据库中的多个连接池签出中使用临时表，例如在使用 ORM Session 时，临时表应在调用 Session.commit() 或 Session.rollback() 之后继续保留，则必须使用维护单个连接的池。如果仅在当前线程中需要范围，则使用 SingletonThreadPool，如果在此情况下需要在多个线程中需要范围，则使用 StaticPool

# maintain the same connection per thread
from sqlalchemy.pool import SingletonThreadPool

engine = create_engine("sqlite:///mydb.db", poolclass=SingletonThreadPool)


# maintain the same connection across all threads
from sqlalchemy.pool import StaticPool

engine = create_engine("sqlite:///mydb.db", poolclass=StaticPool)

请注意，SingletonThreadPool 应配置为要使用的线程数；超出该数量，连接将以非确定性方式关闭。

处理混合字符串/二进制列¶

SQLite 数据库是弱类型的，因此在使用二进制值（在 Python 中表示为 b'some string'）时，特定的 SQLite 数据库可能在不同行中具有数据值，其中某些数据值将由 Pysqlite 驱动程序作为 b'' 值返回，而另一些数据值将作为 Python 字符串返回，例如 '' 值。如果始终如一地使用 SQLAlchemy LargeBinary 数据类型，则不会发生这种情况，但是如果特定的 SQLite 数据库具有使用 Pysqlite 驱动程序直接插入的数据，或者在使用 SQLAlchemy String 类型（后来更改为 LargeBinary）时，该表将无法一致地读取，因为 SQLAlchemy 的 LargeBinary 数据类型不处理字符串，因此它无法 “编码” 字符串格式的值。

要处理在同一列中具有混合字符串/二进制数据的 SQLite 表，请使用自定义类型，该类型将单独检查每一行

from sqlalchemy import String
from sqlalchemy import TypeDecorator


class MixedBinary(TypeDecorator):
    impl = String
    cache_ok = True

    def process_result_value(self, value, dialect):
        if isinstance(value, str):
            value = bytes(value, "utf-8")
        elif value is not None:
            value = bytes(value)

        return value

然后在通常使用 LargeBinary 的位置使用上述 MixedBinary 数据类型。

可序列化隔离/保存点/事务性 DDL¶

在数据库锁定行为/并发部分中，我们提到了 pysqlite 驱动程序的一系列问题，这些问题阻止了 SQLite 的几个功能正常工作。pysqlite DBAPI 驱动程序存在多个长期存在的错误，这些错误会影响其事务行为的正确性。在其默认操作模式下，SQLite 的可序列化隔离、事务性 DDL 和保存点支持等功能无法正常工作，为了使用这些功能，必须采取解决方法。

问题本质上是驱动程序试图猜测用户的意图，未能启动事务，有时甚至过早结束事务，以尽量减少 SQLite 数据库的文件锁定行为，即使 SQLite 本身对只读活动使用 “共享” 锁。

SQLAlchemy 选择默认情况下不更改此行为，因为这是 pysqlite 驱动程序长期以来的预期行为；如果 pysqlite 驱动程序尝试修复这些问题，那将更倾向于 SQLAlchemy 的默认值。

好消息是，通过几个事件，我们可以通过完全禁用 pysqlite 的功能并自行发出 BEGIN 来完全实现事务支持。这是通过两个事件侦听器实现的

from sqlalchemy import create_engine, event

engine = create_engine("sqlite:///myfile.db")


@event.listens_for(engine, "connect")
def do_connect(dbapi_connection, connection_record):
    # disable pysqlite's emitting of the BEGIN statement entirely.
    # also stops it from emitting COMMIT before any DDL.
    dbapi_connection.isolation_level = None


@event.listens_for(engine, "begin")
def do_begin(conn):
    # emit our own BEGIN
    conn.exec_driver_sql("BEGIN")

警告

当使用上述配方时，建议不要在 Connection.execution_options.isolation_level 设置以及带有 SQLite 驱动程序的 create_engine() 上使用，因为此功能也必然会更改 “.isolation_level” 设置。

在上面，我们拦截了一个新的 pysqlite 连接并禁用了任何事务集成。然后，在 SQLAlchemy 知道事务范围即将开始时，我们自行发出 "BEGIN"。

当我们控制 "BEGIN" 时，我们还可以直接控制 SQLite 的锁定模式，该模式在 BEGIN TRANSACTION 中引入，方法是将所需的锁定模式添加到我们的 "BEGIN" 中

@event.listens_for(engine, "begin")
def do_begin(conn):
    conn.exec_driver_sql("BEGIN EXCLUSIVE")

另请参阅

BEGIN TRANSACTION - 在 SQLite 站点上

sqlite3 SELECT does not BEGIN a transaction - 在 Python 错误跟踪器上

sqlite3 module breaks transactions and potentially corrupts data - 在 Python 错误跟踪器上

用户定义函数¶

pysqlite 支持 create_function() 方法，该方法允许我们在 Python 中创建自己的用户定义函数 (UDF)，并在 SQLite 查询中直接使用它们。这些函数在特定的 DBAPI 连接中注册。

SQLAlchemy 将连接池与基于文件的 SQLite 数据库一起使用，因此我们需要确保在创建连接时将 UDF 附加到连接。这可以通过事件侦听器完成

from sqlalchemy import create_engine
from sqlalchemy import event
from sqlalchemy import text


def udf():
    return "udf-ok"


engine = create_engine("sqlite:///./db_file")


@event.listens_for(engine, "connect")
def connect(conn, rec):
    conn.create_function("udf", 0, udf)


for i in range(5):
    with engine.connect() as conn:
        print(conn.scalar(text("SELECT UDF()")))

Aiosqlite¶

通过 aiosqlite 驱动程序支持 SQLite 数据库。

DBAPI¶

有关 aiosqlite 的文档和下载信息（如果适用）可在以下网址获取：https://pypi.ac.cn/project/aiosqlite/

连接¶

连接字符串

sqlite+aiosqlite:///file_path

aiosqlite 方言提供对在 pysqlite 之上运行的 SQLAlchemy asyncio 接口的支持。

aiosqlite 是 pysqlite 的包装器，每个连接使用一个后台线程。它实际上不使用非阻塞 IO，因为 SQLite 数据库不是基于套接字的。但是，它确实提供了一个可用的 asyncio 接口，该接口对于测试和原型设计目的很有用。

通过使用特殊的 asyncio 调解层，aiosqlite 方言可用作 SQLAlchemy asyncio 扩展包的后端。

此方言通常应仅与 create_async_engine() 引擎创建函数一起使用

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("sqlite+aiosqlite:///filename")

URL 将所有参数传递给 pysqlite 驱动程序，因此所有连接参数都与 Pysqlite 的参数相同。

用户定义函数¶

aiosqlite 扩展了 pysqlite 以支持 async，因此我们可以创建自己的用户定义函数 (UDF) 在 Python 中，并在 SQLite 查询中直接使用它们，如此处所述：用户定义函数。

可序列化隔离/保存点/事务性 DDL（asyncio 版本）¶

与 pysqlite 类似，aiosqlite 不支持 SAVEPOINT 功能。

解决方案类似于可序列化隔离/保存点/事务性 DDL。这是通过 async 中的事件侦听器实现的

from sqlalchemy import create_engine, event
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("sqlite+aiosqlite:///myfile.db")


@event.listens_for(engine.sync_engine, "connect")
def do_connect(dbapi_connection, connection_record):
    # disable aiosqlite's emitting of the BEGIN statement entirely.
    # also stops it from emitting COMMIT before any DDL.
    dbapi_connection.isolation_level = None


@event.listens_for(engine.sync_engine, "begin")
def do_begin(conn):
    # emit our own BEGIN
    conn.exec_driver_sql("BEGIN")

警告

当使用上述配方时，建议不要在 Connection.execution_options.isolation_level 设置以及带有 SQLite 驱动程序的 create_engine() 上使用，因为此功能也必然会更改 “.isolation_level” 设置。

池行为¶

SQLAlchemy aiosqlite DBAPI 根据请求的 SQLite 数据库的种类，以不同的方式建立连接池

当指定 :memory: SQLite 数据库时，方言默认将使用 StaticPool。此池维护单个连接，以便对引擎的所有访问都使用相同的 :memory: 数据库。
当指定基于文件的数据库时，方言将使用 AsyncAdaptedQueuePool 作为连接源。

2.0.38 版本更改: SQLite 文件数据库引擎现在默认使用 AsyncAdaptedQueuePool。以前，使用 NullPool。NullPool 类可以通过 create_engine.poolclass 参数指定来使用。

Pysqlcipher¶

通过 pysqlcipher 驱动程序支持 SQLite 数据库。

用于支持使用 SQLCipher 后端的 DBAPI 的方言。

连接¶

连接字符串

sqlite+pysqlcipher://:passphrase@/file_path[?kdf_iter=<iter>]

驱动¶

当前方言选择逻辑是

如果 create_engine.module 参数提供了 DBAPI 模块，则使用该模块。
否则，对于 Python 3，选择 https://pypi.ac.cn/project/sqlcipher3/
如果不可用，则回退到 https://pypi.ac.cn/project/pysqlcipher3/
对于 Python 2，使用 https://pypi.ac.cn/project/pysqlcipher/。

警告

pysqlcipher3 和 pysqlcipher DBAPI 驱动程序已不再维护；截至撰写本文时，sqlcipher3 驱动程序似乎是最新的。为了未来的兼容性，任何兼容 pysqlcipher 的 DBAPI 都可以按如下方式使用

import sqlcipher_compatible_driver

from sqlalchemy import create_engine

e = create_engine(
    "sqlite+pysqlcipher://:password@/dbname.db",
    module=sqlcipher_compatible_driver,
)

这些驱动程序使用 SQLCipher 引擎。该系统本质上向 SQLite 引入了新的 PRAGMA 命令，允许设置密码和其他加密参数，从而允许数据库文件被加密。

连接字符串¶

连接字符串的格式在各方面都与 pysqlite 驱动程序相同，只是现在接受 “password” 字段，该字段应包含密码

e = create_engine("sqlite+pysqlcipher://:testing@/foo.db")

对于绝对文件路径，数据库名称应使用两个前导斜杠

e = create_engine("sqlite+pysqlcipher://:testing@//path/to/foo.db")

SQLCipher 支持的其他加密相关编译指示的选择（如 https://www.zetetic.net/sqlcipher/sqlcipher-api/ 中所述）可以在查询字符串中传递，并且将导致为每个新连接调用该 PRAGMA。目前，支持 cipher、kdf_iter、cipher_page_size 和 cipher_use_hmac

e = create_engine(
    "sqlite+pysqlcipher://:testing@/foo.db?cipher=aes-256-cfb&kdf_iter=64000"
)

警告

以前版本的 sqlalchemy 没有考虑在 URL 字符串中传递的加密相关编译指示，这些编译指示会被静默忽略。如果加密选项不匹配，这可能会导致打开由以前的 sqlalchemy 版本保存的文件时出错。

连接池行为¶

该驱动程序更改了 pysqlite 的默认连接池行为，如线程/连接池行为中所述。已观察到 pysqlcipher 驱动程序在连接时比 pysqlite 驱动程序慢得多，这很可能是由于加密开销，因此此处的方言默认使用 SingletonThreadPool 实现，而不是 pysqlite 使用的 NullPool 连接池。与往常一样，连接池实现完全可以使用 create_engine.poolclass 参数进行配置； StaticPool 可能更适合单线程使用，或者可以使用 NullPool 来防止未加密的连接长时间保持打开状态，但这会以新连接的启动时间变慢为代价。

SQLAlchemy 2.0 文档

SQLAlchemy 2.0 文档

方言

项目版本

SQLite¶

DBAPI 支持¶

日期和时间类型¶

确保文本亲和性¶

SQLite 自动递增行为¶

使用 AUTOINCREMENT 关键字¶

允许除 Integer/INTEGER 之外的 SQLAlchemy 类型使用自动递增行为¶

数据库锁定行为 / 并发¶

事务隔离级别 / 自动提交¶

INSERT/UPDATE/DELETE…RETURNING¶

SAVEPOINT 支持¶

事务性 DDL¶

外键支持¶

约束的 ON CONFLICT 支持¶

INSERT…ON CONFLICT (Upsert)¶

指定目标¶

SET 子句¶

使用排除的 INSERT 值更新¶

附加 WHERE 条件¶

使用 DO NOTHING 跳过行¶

类型反射¶

部分索引¶

带点的列名¶

SQLite 特定的表选项¶

反射内部模式表¶

SQLite 数据类型¶

SQLite DML 构造¶

Pysqlite¶

DBAPI¶

连接¶

驱动程序¶

连接字符串¶

URI 连接¶

正则表达式支持¶

与 sqlite3 “原生” 日期和日期时间类型的兼容性¶

线程/池行为¶

禁用文件数据库的连接池¶

在多线程中使用内存数据库¶

将临时表与 SQLite 一起使用¶

处理混合字符串/二进制列¶

可序列化隔离/保存点/事务性 DDL¶

用户定义函数¶

Aiosqlite¶

DBAPI¶

连接¶

用户定义函数¶

可序列化隔离/保存点/事务性 DDL（asyncio 版本）¶

池行为¶

Pysqlcipher¶

连接¶

驱动¶

连接字符串¶

连接池行为¶