SQLite¶

对 SQLite 数据库的支持。

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

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

DBAPI 支持¶

以下方言/DBAPI 选项可用。请参考各个 DBAPI 部分以获取连接信息。

pysqlite
aiosqlite
pysqlcipher

日期和时间类型¶

SQLite 没有内置的 DATE、TIME 或 DATETIME 类型，并且 pysqlite 没有提供开箱即用的功能来在 Python datetime 对象和 SQLite 支持的格式之间转换值。SQLAlchemy 自己的 DateTime 及相关类型在使用 SQLite 时提供日期格式化和解析功能。实现类是 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" 的类型名称将被确定为“整数亲和性”。名为 "BIGINT"、"SPECIAL_INT" 甚至 "XYZINTQPR" 的类型将被 SQLite 视为“整数”亲和性。但是，SQLite 自动递增功能（无论是隐式还是显式启用）都要求列类型的名称与字符串“INTEGER”完全相同。因此，如果应用程序使用诸如 BigInteger 这样的类型作为主键，则在 SQLite 上，此类型需要在发出初始 CREATE TABLE 语句时渲染为名称 "INTEGER"，以便自动递增行为可用。

实现此目的的一种方法是在 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 默认情况下缺乏写入并发的更多信息，请参阅 Situations Where Another RDBMS May Work Better - High Concurrency 页面底部。

以下小节介绍受 SQLite 文件系统架构影响的区域，此外，在使用 pysqlite 驱动程序时，这些区域通常需要变通方法才能正常工作。

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

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

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 语句延迟的程度远远超过通常可行的情况。请参阅部分 Serializable isolation / Savepoints / Transactional DDL 或 Serializable isolation / Savepoints / Transactional DDL (asyncio version)，了解解决此行为的技巧。

另请参阅

设置事务隔离级别，包括 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() 方法在核心级别使用，并且可以通过 Session.begin_nested() 方法在 ORM 级别使用。但是，除非采取变通方法，否则 SAVEPOINT 根本无法与 pysqlite 一起使用。

警告

SQLite 的 SAVEPOINT 功能受到 pysqlite 和 aiosqlite 驱动程序中未解决的问题的影响，这些问题将 BEGIN 语句延迟的程度远远超过通常可行的情况。请参阅部分 Serializable isolation / Savepoints / Transactional DDL 和 Serializable isolation / Savepoints / Transactional DDL (asyncio version)，了解解决此行为的技巧。

事务性 DDL¶

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

警告

SQLite 的事务性 DDL 受 pysqlite 驱动程序中未解决的问题的影响，该驱动程序无法发出 BEGIN，此外，在遇到 DDL 时，它还会强制执行 COMMIT 以取消任何事务。请参阅部分 Serializable isolation / Savepoints / Transactional DDL，了解解决此行为的技巧。

外键支持¶

SQLite 在为表发出 CREATE 语句时支持 FOREIGN KEY 语法，但是默认情况下，这些约束对表的运行没有影响。

SQLite 上的约束检查有三个先决条件

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

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 Foreign Key Support - 在 SQLite 网站上。

Events - SQLAlchemy 事件 API。

Creating/Dropping Foreign Key Constraints via ALTER - 有关 SQLAlchemy 处理的更多信息: 相互依赖的外键约束。

约束的 ON CONFLICT 支持¶

另请参阅

本节介绍 SQLite 的“ON CONFLICT”的 DDL 版本，它出现在 CREATE TABLE 语句中。有关应用于 INSERT 语句的“ON CONFLICT”，请参阅 INSERT…ON CONFLICT (Upsert)。

SQLite 支持一种非标准的 DDL 子句，称为 ON CONFLICT，它可以应用于主键、唯一、检查和非空约束。在 DDL 中，它根据目标约束的位置在“CONSTRAINT”子句内或列定义内呈现。要在 DDL 内呈现此子句，可以将扩展参数 sqlite_on_conflict 与 PrimaryKeyConstraint、UniqueConstraint、CheckConstraint 对象中的字符串冲突解决算法一起指定。在 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 的 DML 版“ON CONFLICT”，它发生在 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 文档中。

指定目标¶

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

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='[email protected]',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 类型与大多数其他数据库后端的类型不同，因为类型的字符串名称通常不以一对一的方式对应于“类型”。相反，SQLite 将每列的类型行为链接到五种所谓的“类型亲和性”之一，这些亲和性基于类型的字符串匹配模式。

SQLAlchemy 的反射过程在检查类型时，使用一个简单的查找表将返回的关键字链接到提供的 SQLAlchemy 类型。该查找表存在于 SQLite 方言中，就像在所有其他方言中一样。但是，SQLite 方言对于在查找映射中找不到特定类型名称时具有不同的“回退”例程；它改为实现位于 https://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

当类型名称与上述类型不匹配时，将使用“类型亲和力”查找。

INTEGER 如果类型名称包含字符串 INT，则返回。
TEXT 如果类型名称包含字符串 CHAR，CLOB 或 TEXT，则返回。
NullType 如果类型名称包含字符串 BLOB，则返回。
REAL 如果类型名称包含字符串 REAL，FLOA 或 DOUB，则返回。
否则，将使用 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 方言与 Table 构造一起直接支持 CREATE TABLE 的一个选项。

WITHOUT ROWID:

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

另请参阅

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 有效的 UPPERCASE 类型都可从顶级方言导入，无论它们来自 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 日期对象。
DATETIME	使用字符串在 SQLite 中表示 Python 日期时间对象。
JSON	SQLite JSON 类型。
TIME	使用字符串在 SQLite 中表示 Python 时间对象。

class sqlalchemy.dialects.sqlite.DATETIME¶

使用字符串在 SQLite 中表示 Python 日期时间对象。

默认的字符串存储格式为

"%(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() 用于默认的日期时间字符串解析。

可以使用 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¶ – 将应用于具有年、月、日、时、分、秒和微秒键的字典的格式字符串。
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 日期对象。

默认的字符串存储格式为

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

例如

2011-03-15

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

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

可以使用 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¶ – 将应用于具有年、月和日键的字典的格式字符串。
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 是一个可加载扩展，因此可能不可用，或者可能需要在运行时加载。

JSON 在基础 JSON 数据类型用于 SQLite 后端时会自动使用。

另请参阅

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 时间对象。

默认的字符串存储格式为

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

例如

12:05:57.10558

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

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

可以使用 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¶ – 将应用于具有时、分、秒和微秒键的字典的格式字符串。
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` 结构。
插入	SQLite 特定的 INSERT 实现。

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

构建一个 SQLite 特定的变体 Insert 结构。

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

该 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: _OnConflictIndexElementsT = None, index_where: _OnConflictIndexWhereT = None) → Self¶

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

参数：:

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

method sqlalchemy.dialects.sqlite.Insert.on_conflict_do_update(index_elements: _OnConflictIndexElementsT = None, index_where: _OnConflictIndexWhereT = None, set_: _OnConflictSetT = None, where: _OnConflictWhereT = None) → Self¶

指定 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¶ – 可选参数。如果存在，可以是字面量 SQL 字符串或 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 的“数据库”部分。请注意，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 的“数据库”部分保留（即，在斜杠之后）

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 驱动程序一侧接受的固定参数列表来实现的。例如，要包含一个 URL，该 URL 指示 Python sqlite3 的“timeout”和“check_same_thread”参数，以及 SQLite 的“mode”和“nolock”参数，它们都可以一起传递到查询字符串上

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() 中可以使用 create_engine.creator 参数绕过整个“URL”过程，该参数允许使用自定义可调用对象，该对象直接创建 Python sqlite3 驱动程序级连接。

新版功能，版本 1.3.9。

另请参阅

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

正则表达式支持¶

1.4 版中的新增功能。

对 ColumnOperators.regexp_match() 运算符的支持是使用 Python 的 re.search 函数提供的。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 日期或时间对象。pysqlite 方言提供的日期和时间类型目前与这些选项不兼容，因为它们呈现包含微秒的 ISO 日期/时间，而 pysqlite 的驱动程序不支持。此外，SQLAlchemy 目前不会自动呈现独立函数“current_timestamp”和“current_date”所需的“cast”语法，以便它们原生返回 datetime/date 类型。不幸的是，pysqlite 不会在 cursor.description 中提供标准 DBAPI 类型，这使得 SQLAlchemy 无法在没有昂贵 per-row 类型检查的情况下动态检测这些类型。

请记住，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 的成熟，它在多线程下的行为有所改善，甚至包括内存中数据库在多线程中使用的选项。

线程禁止称为“检查同一线程”，可以使用 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。通过使用 create_engine.poolclass 参数指定它可以使用 NullPool 类。

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

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

from sqlalchemy import NullPool
engine = create_engine("sqlite:///myfile.db", poolclass=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

然后使用上面的 MixedBinary 数据类型替换通常使用 LargeBinary 的地方。

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

在数据库锁定行为 / 并发部分，我们提到了 pysqlite 驱动程序的各种问题，这些问题会阻止 SQLite 的几个功能正常工作。pysqlite DBAPI 驱动程序存在几个长期存在的错误，这些错误会影响其事务行为的正确性。在默认操作模式下，SQLite 的功能（如 SERIALIZABLE 隔离、事务性 DDL 和 SAVEPOINT 支持）不起作用，要使用这些功能，必须采取变通方法。

问题本质上是驱动程序试图猜测用户的意图，无法启动事务，有时会过早地结束事务，以尽量减少 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 设置上使用 Connection 和 create_engine()，以及 SQLite 驱动程序，因为此函数必然还会更改“.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 以支持异步，因此我们可以使用 Python 创建自己的用户定义函数 (UDF)，并在 SQLite 查询中直接使用它们，如下所述：用户定义函数。

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

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

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

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 设置上使用 Connection 和 create_engine()，以及 SQLite 驱动程序，因为此函数必然还会更改“.isolation_level”设置。

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，这将导致对每个新连接调用该 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 字符串中传递的与加密相关的 pragma，这些 pragma 被静默忽略。如果加密选项不匹配，这可能会导致打开以前版本的 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¶

连接¶

驱动程序¶

连接字符串¶

池行为¶