Python sqlite3.Connection.create_function 怎样注册自定义 SQL 函数

如果查询条件里需要一段 Python 才能表达的计算，最直接的做法是把 callable 注册到当前 sqlite3.Connection，然后像普通 SQL 函数一样在 SELECT 中调用。create_function(name, narg, func, deterministic=False) 负责建立这层绑定：name 是 SQL 中使用的函数名，narg 是参数个数，func 是 Python 回调。

官方地址：https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function

先把函数契约定窄：固定参数就传明确的 narg，确实需要任意参数时才使用 -1；回调只返回 SQLite 能识别的基础类型，纯函数才标记为 deterministic。

先判断三个参数是否匹配

create_function 是连接级注册，不是全局注册。函数只对这条连接上的后续 SQL 可见，换一条连接后需要重新注册。回调参数数量必须和 narg 一致，返回值则应是 SQLite 原生支持的类型，例如整数、浮点数、字符串、字节串或空值。

因此，排查“函数不存在”时先看查询是否使用了同一个连接；排查“参数数量错误”时再对照注册时的 narg，不要先怀疑 SQL 的列名。

固定参数的函数最容易维护

以把用户名规范化为查询字段为例，函数只接受一个字符串，并返回一个新的字符串。示例把连接放在 try/finally 中，避免文章里的最小示例把连接生命周期省略掉。

import sqlite3

def normalize_name(value):
    # 让回调只处理一个输入，并把空值原样交给 SQL 层判断。
    if value is None:
        return None
    # 返回字符串、整数、浮点数、字节串或 None 等 SQLite 可接收的类型。
    return str(value).strip().casefold()

con = sqlite3.connect(":memory:")
try:
    con.execute("CREATE TABLE users(name TEXT)")
    con.executemany("INSERT INTO users(name) VALUES (?)", [(" Alice ",), ("Bob",)])

    # narg=1 表示 SQL 调用 normalize_name(?) 时必须提供一个参数。
    con.create_function("normalize_name", 1, normalize_name)
    rows = con.execute(
        "SELECT normalize_name(name) FROM users ORDER BY name"
    ).fetchall()
finally:
    # 连接不再使用时显式关闭，避免把资源生命周期交给垃圾回收。
    con.close()

这里的 SQL 函数名可以和 Python 函数名不同。真正的契约是注册时的三个位置参数：SQL 使用的名字、参数数量和回调对象。回调返回不受 SQLite 支持的 Python 对象时，查询会在调用阶段失败，因此不要直接返回自定义类实例。

可变参数只在确有需要时使用

把 narg 设为 -1 后，SQLite 允许函数接收任意数量的参数。它适合“把任意多个片段拼接起来”这类明确的可变参数需求，但也会放宽调用约束，后续维护者更难从注册处看出正常输入形状。

def join_parts(*parts):
    # 可变参数函数自行规定空值和分隔符的语义。
    return " / ".join(str(part) for part in parts if part is not None)

con.create_function("join_parts", -1, join_parts)
# -1 允许下面的 SQL 传入两个或三个参数；参数仍按位置进入 *parts。
row = con.execute(
    "SELECT join_parts(?, ?, ?)",
    ("go", "python", None),
).fetchone()

可变参数不是“自动把一列展开成多个参数”。SQL 传几个表达式，Python 回调才收到几个位置参数；如果要处理一列的多行数据，应考虑聚合函数，而不是把 narg=-1 当成聚合能力。

deterministic 只给真正纯粹的函数

deterministic=True 会把函数标记为确定性函数，SQLite 可以据此进行额外优化。只有当相同输入始终得到相同输出，而且函数没有读取时间、随机数、外部文件或可变全局状态时，才适合打开这个标记。

import hashlib

def digest_text(value):
    # 固定输入得到固定摘要，适合声明为 deterministic。
    if value is None:
        return None
    return hashlib.sha256(str(value).encode("utf-8")).hexdigest()

# deterministic=True 是优化提示，不是异常处理或缓存开关。
con.create_function("digest_text", 1, digest_text, deterministic=True)
result = con.execute(
    "SELECT digest_text(?)", ("hello",)
).fetchone()

如果回调可能抛出异常，可在开发阶段调用 sqlite3.enable_callback_tracebacks(True) 让回调 traceback 输出到标准错误；这只是诊断辅助，不能把任意异常吞掉后继续返回一个看似成功的值。

需要撤销时传入 None

同一连接上可以用相同的函数名和参数数量注册新的回调，也可以把回调设为 None 来移除已有 SQL 函数。迁移或热更新时，最好把注册和撤销写成清晰的生命周期动作，避免同一个连接在不同模块之间留下隐式状态。

# 传入 None 移除同名、同参数数量的用户自定义函数。
con.create_function("normalize_name", 1, None)

若只是关闭连接，则无需单独撤销；关闭连接后这条连接上的注册自然不再可用。若连接由连接池复用，就要在池初始化或连接创建钩子中统一注册，否则不同连接会出现行为不一致。

发布前的排查清单

相关问题

create_function 能注册聚合函数吗？不能。聚合场景应使用 create_aggregate，不要用可变参数函数模拟跨行累积。

注册后另一条连接为什么找不到？注册绑定在具体 Connection 上，连接池需要对每个新连接重复初始化。

回调报错为什么不像普通 Python 函数那样直接显示？SQLite 通过回调调用 Python 函数，开发阶段可启用 enable_callback_tracebacks(True) 辅助观察回调异常。

把 Python 逻辑注册为 SQL 函数时，最重要的不是把函数做得“万能”，而是把连接范围、参数个数、返回类型和确定性声明写清楚。固定契约能让查询更容易复用，也让连接池和故障排查更可控。