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

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

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,
)

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()。

可以使用 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() 构造函数进行位置参数传递。
  

类签名

类 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() 来解析传入字符串。如果 regexp 包含命名组，则生成的匹配字典将作为关键字参数应用于 Python date() 构造函数。否则，如果使用位置组，则通过 *map(int, match_obj.groups(0)) 将调用 date() 构造函数以位置参数形式。
  

类签名

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

class sqlalchemy.dialects.sqlite.JSON

SQLite JSON 类型。

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

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

另请参阅

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

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

版本 1.3 中的新内容。

成员

init()

类签名

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

method __init__(none_as_null: bool = False)

继承自 JSON 的 sqlalchemy.types.JSON.__init__ 方法

构造一个 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)) 将调用时间()构造函数以位置参数的方式。
  

类签名

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

SQLite DML Constructs

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。

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

新版本 1.4 中新增。

另请参见

INSERT…ON CONFLICT (Upsert)

成员

excluded, inherit_cache, on_conflict_do_nothing(), on_conflict_do_update()

类签名

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

attribute 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 inherit_cache: bool | None = False

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

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

如果与此类本地属性而不是其超类有关的属性不会改变与对象相对应的 SQL，则可以将此标志设置为 True。

See also

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

method on_conflict_do_nothing(index_elements: _OnConflictIndexElementsT = None, index_where: _OnConflictIndexWhereT = None) → Self

指定了 ON CONFLICT 子句的 DO NOTHING 操作。

参数：

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

method 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 的文档和下载信息（如果适用）可在此处获取：docs.python.org/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
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 驱动程序接受的一组固定参数来实现的。例如，要包含指示  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()中的create_engine.creator参数来绕过整个“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 日期或日期时间对象。pysqlite  方言提供的日期和日期时间类型目前与这些选项不兼容，因为它们呈现包括微秒的 ISO 日期/日期时间，而 pysqlite  的驱动程序不包括。此外，SQLAlchemy  目前不会自动呈现“cast”语法，以便使独立函数“current_timestamp”和“current_date”返回本地的日期时间/日期类型。不幸的是，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 的成熟，它在多线程下的行为已经改进，甚至包括选项，允许在多个线程中使用仅内存数据库。

线程禁止被称为“检查同一线程”，可以使用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类来使用NullPool类。
  

为文件数据库禁用连接池

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

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

已观察到由于QueuePool没有连接重用，因此NullPool实现在重复检出时产生极小的性能开销。然而，如果应用程序出现文件被锁定的问题，仍然可能有利于使用此类。

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

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

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  的功能，如 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")

警告

在使用上述方法时，建议不要在 SQLite 驱动程序上使用 Connection.execution_options.isolation_level 设置，因为这个函数必然也会改变“.isolation_level”设置。

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

当我们控制 "BEGIN" 时，我们也可以直接控制 SQLite 的锁定模式，通过在我们的 "BEGIN" 中添加所需的锁定模式：

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

另请参阅

BEGIN TRANSACTION - 在 SQLite 网站上

sqlite3 SELECT 不会开始事务 - 在 Python 缺陷跟踪器上

sqlite3 模块破坏事务并可能损坏数据 - 在 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()")))

数据库 API

pysqlite 的文档和下载信息（如果适用）可在此处获取：docs.python.org/library/sqlite3.html

连接

连接字符串：

sqlite+pysqlite:///file_path

驱动程序

在所有现代 Python 版本上，sqlite3 Python 数据库 API 是标准的；对于 cPython 和 Pypy，不需要额外安装。

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