SQLite 完全指南 / 18 - 驱动集成

18 - 驱动集成：Python、Go、Java、Node.js、Rust

18.1 驱动概览

语言	驱动	类型	特点
Python	`sqlite3`	标准库	内置，零依赖
Go	`modernc.org/sqlite`	纯 Go	无需 CGO
Go	`github.com/mattn/go-sqlite3`	CGO	性能最佳
Java	`org.xerial:sqlite-jdbc`	JDBC	最广泛
Rust	`rusqlite`	原生	安全高效
Node.js	`better-sqlite3`	原生	同步 API，性能最好
Node.js	`libsql`	原生	支持 Turso/LibSQL
PHP	`PDO_SQLite`	标准扩展	内置
C#	`Microsoft.Data.Sqlite`	ADO.NET	官方维护

18.2 Python

18.2.1 基本用法

import sqlite3

# 连接数据库
conn = sqlite3.connect('mydb.db')
# 内存数据库
conn = sqlite3.connect(':memory:')

# 设置 Row 工厂（返回字典）
conn.row_factory = sqlite3.Row

# 设置 PRAGMA
conn.execute('PRAGMA journal_mode = WAL')
conn.execute('PRAGMA foreign_keys = ON')
conn.execute('PRAGMA busy_timeout = 5000')

# 创建表
conn.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL,
        age INTEGER DEFAULT 0
    )
''')

# 插入数据
conn.execute('INSERT INTO users (name, email, age) VALUES (?, ?, ?)',
             ('张三', '[email protected]', 25))
conn.commit()

# 查询数据
cursor = conn.execute('SELECT * FROM users WHERE age > ?', (18,))
for row in cursor:
    print(dict(row))  # {'id': 1, 'name': '张三', 'email': '[email protected]', 'age': 25}

conn.close()

18.2.2 上下文管理器

import sqlite3

def get_connection(db_path='mydb.db'):
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    conn.execute('PRAGMA journal_mode = WAL')
    conn.execute('PRAGMA foreign_keys = ON')
    return conn

# 使用上下文管理器
with get_connection() as conn:
    users = conn.execute('SELECT * FROM users').fetchall()
    # 自动提交或回滚

18.2.3 批量操作

import sqlite3

conn = sqlite3.connect(':memory:')
conn.execute('CREATE TABLE data (id INTEGER PRIMARY KEY, val TEXT)')

# executemany 批量插入
data = [(i, f'value_{i}') for i in range(10000)]
conn.executemany('INSERT INTO data VALUES (?, ?)', data)
conn.commit()

# executemany 批量更新
updates = [(f'new_{i}', i) for i in range(10000)]
conn.executemany('UPDATE data SET val = ? WHERE id = ?', updates)
conn.commit()

# executemany 批量删除
ids = [(i,) for i in range(5000)]
conn.executemany('DELETE FROM data WHERE id = ?', ids)
conn.commit()

18.2.4 自定义类型与适配器

import sqlite3
import json
from datetime import datetime, date

# 自定义适配器：Python → SQLite
def adapt_datetime(dt):
    return dt.isoformat()

def adapt_json(obj):
    return json.dumps(obj, ensure_ascii=False)

# 自定义转换器：SQLite → Python
def convert_datetime(text):
    return datetime.fromisoformat(text.decode())

def convert_json(text):
    return json.loads(text)

# 注册适配器和转换器
sqlite3.register_adapter(datetime, adapt_datetime)
sqlite3.register_adapter(dict, adapt_json)
sqlite3.register_converter('datetime', convert_datetime)
sqlite3.register_converter('json', convert_json)

# 使用 detect_types 启用类型检测
conn = sqlite3.connect(':memory:', detect_types=sqlite3.PARSE_DECLTYPES)

18.2.5 连接池（使用 SQLAlchemy）

from sqlalchemy import create_engine

# SQLAlchemy 连接池
engine = create_engine(
    'sqlite:///mydb.db',
    pool_size=5,
    max_overflow=10,
    connect_args={
        'check_same_thread': False,  # 允许多线程
    }
)

# 使用原生连接设置 PRAGMA
from sqlalchemy import event

@event.listens_for(engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA journal_mode=WAL")
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.execute("PRAGMA busy_timeout=5000")
    cursor.close()

18.3 Go

18.3.1 使用 modernc.org/sqlite（纯 Go，推荐）

package main

import (
    "database/sql"
    "fmt"
    "log"

    _ "modernc.org/sqlite"
)

func main() {
    // 连接数据库
    db, err := sql.Open("sqlite", "mydb.db")
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()

    // 设置 PRAGMA
    db.Exec("PRAGMA journal_mode = WAL")
    db.Exec("PRAGMA foreign_keys = ON")
    db.Exec("PRAGMA busy_timeout = 5000")

    // 创建表
    _, err = db.Exec(`
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL,
            age INTEGER DEFAULT 0
        )
    `)
    if err != nil {
        log.Fatal(err)
    }

    // 插入数据
    result, err := db.Exec(
        "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
        "张三", "[email protected]", 25,
    )
    if err != nil {
        log.Fatal(err)
    }
    id, _ := result.LastInsertId()
    fmt.Printf("插入 ID: %d\n", id)

    // 查询单行
    var name string
    var email string
    var age int
    err = db.QueryRow("SELECT name, email, age FROM users WHERE id = ?", id).
        Scan(&name, &email, &age)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("用户: %s, %s, %d\n", name, email, age)

    // 查询多行
    rows, err := db.Query("SELECT id, name, email FROM users WHERE age > ?", 18)
    if err != nil {
        log.Fatal(err)
    }
    defer rows.Close()

    for rows.Next() {
        var id int
        var name, email string
        rows.Scan(&id, &name, &email)
        fmt.Printf("ID=%d, Name=%s, Email=%s\n", id, name, email)
    }
}

18.3.2 使用 go-sqlite3（CGO）

import (
    "database/sql"
    _ "github.com/mattn/go-sqlite3"
)

func main() {
    // 需要 CGO_ENABLED=1 编译
    db, err := sql.Open("sqlite3", "mydb.db?_journal_mode=WAL&_foreign_keys=ON")
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()
}

18.3.3 Go 事务示例

// 使用事务
tx, err := db.Begin()
if err != nil {
    log.Fatal(err)
}

_, err = tx.Exec("UPDATE accounts SET balance = balance - ? WHERE id = ?", 100, 1)
if err != nil {
    tx.Rollback()
    log.Fatal(err)
}

_, err = tx.Exec("UPDATE accounts SET balance = balance + ? WHERE id = ?", 100, 2)
if err != nil {
    tx.Rollback()
    log.Fatal(err)
}

err = tx.Commit()
if err != nil {
    log.Fatal(err)
}

18.4 Java

18.4.1 使用 JDBC

import java.sql.*;

public class SQLiteDemo {
    public static void main(String[] args) throws Exception {
        // 加载驱动
        Class.forName("org.sqlite.JDBC");

        // 连接数据库
        String url = "jdbc:sqlite:mydb.db";
        try (Connection conn = DriverManager.getConnection(url)) {
            // 设置 PRAGMA
            try (Statement stmt = conn.createStatement()) {
                stmt.execute("PRAGMA journal_mode = WAL");
                stmt.execute("PRAGMA foreign_keys = ON");
                stmt.execute("PRAGMA busy_timeout = 5000");
            }

            // 创建表
            try (Statement stmt = conn.createStatement()) {
                stmt.execute("""
                    CREATE TABLE IF NOT EXISTS users (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        name TEXT NOT NULL,
                        email TEXT UNIQUE NOT NULL,
                        age INTEGER DEFAULT 0
                    )
                """);
            }

            // 参数化查询
            String sql = "INSERT INTO users (name, email, age) VALUES (?, ?, ?)";
            try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
                pstmt.setString(1, "张三");
                pstmt.setString(2, "[email protected]");
                pstmt.setInt(3, 25);
                pstmt.executeUpdate();
            }

            // 查询
            try (Statement stmt = conn.createStatement();
                 ResultSet rs = stmt.executeQuery("SELECT * FROM users")) {
                while (rs.next()) {
                    System.out.printf("ID=%d, Name=%s, Email=%s, Age=%d%n",
                        rs.getInt("id"),
                        rs.getString("name"),
                        rs.getString("email"),
                        rs.getInt("age"));
                }
            }
        }
    }
}

<!-- Maven 依赖 -->
<dependency>
    <groupId>org.xerial</groupId>
    <artifactId>sqlite-jdbc</artifactId>
    <version>3.45.1.0</version>
</dependency>

18.4.2 连接池（HikariCP）

import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;

HikariConfig config = new HikariConfig();
config.setJdbcUrl("jdbc:sqlite:mydb.db");
config.setMaximumPoolSize(5);
config.setConnectionInitSql(
    "PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON; PRAGMA busy_timeout=5000;"
);

HikariDataSource ds = new HikariDataSource(config);

18.5 Node.js

18.5.1 better-sqlite3（推荐）

const Database = require('better-sqlite3');

// 连接数据库
const db = new Database('mydb.db', {
    verbose: console.log  // 可选：日志
});

// 设置 PRAGMA
db.pragma('journal_mode = WAL');
db.pragma('foreign_keys = ON');
db.pragma('busy_timeout = 5000');

// 创建表
db.exec(`
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL,
        age INTEGER DEFAULT 0
    )
`);

// 插入数据（预编译语句）
const insert = db.prepare(
    'INSERT INTO users (name, email, age) VALUES (@name, @email, @age)'
);
insert.run({ name: '张三', email: '[email protected]', age: 25 });

// 批量插入
const insertMany = db.transaction((users) => {
    for (const user of users) {
        insert.run(user);
    }
});

insertMany([
    { name: '李四', email: '[email protected]', age: 30 },
    { name: '王五', email: '[email protected]', age: 28 },
]);

// 查询
const user = db.prepare('SELECT * FROM users WHERE id = ?').get(1);
console.log(user);  // { id: 1, name: '张三', email: '[email protected]', age: 25 }

// 查询多行
const users = db.prepare('SELECT * FROM users WHERE age > ?').all(18);
console.log(users);

// 迭代查询结果
const iterator = db.prepare('SELECT * FROM users').iterate();
for (const row of iterator) {
    console.log(row);
}

// 关闭连接
db.close();

18.5.2 better-sqlite3 事务

const db = new Database('mydb.db');

// 显式事务
const transfer = db.transaction((fromId, toId, amount) => {
    const from = db.prepare('SELECT balance FROM accounts WHERE id = ?').get(fromId);
    if (from.balance < amount) {
        throw new Error('余额不足');
    }
    db.prepare('UPDATE accounts SET balance = balance - ? WHERE id = ?').run(amount, fromId);
    db.prepare('UPDATE accounts SET balance = balance + ? WHERE id = ?').run(amount, toId);
});

transfer(1, 2, 100);

18.5.3 Express.js 示例

const express = require('express');
const Database = require('better-sqlite3');

const app = express();
const db = new Database('app.db');

app.use(express.json());

app.get('/api/users', (req, res) => {
    const users = db.prepare('SELECT * FROM users').all();
    res.json(users);
});

app.post('/api/users', (req, res) => {
    const { name, email, age } = req.body;
    const result = db.prepare(
        'INSERT INTO users (name, email, age) VALUES (?, ?, ?)'
    ).run(name, email, age);
    res.json({ id: result.lastInsertRowid });
});

app.listen(3000);

18.6 Rust

18.6.1 使用 rusqlite

use rusqlite::{Connection, Result, params};

fn main() -> Result<()> {
    // 连接数据库
    let conn = Connection::open("mydb.db")?;

    // 设置 PRAGMA
    conn.execute_batch("
        PRAGMA journal_mode = WAL;
        PRAGMA foreign_keys = ON;
        PRAGMA busy_timeout = 5000;
    ")?;

    // 创建表
    conn.execute("
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL,
            age INTEGER DEFAULT 0
        )
    ", [])?;

    // 插入数据
    conn.execute(
        "INSERT INTO users (name, email, age) VALUES (?1, ?2, ?3)",
        params!["张三", "[email protected]", 25],
    )?;

    // 查询
    let mut stmt = conn.prepare("SELECT id, name, email, age FROM users WHERE age > ?1")?;
    let users = stmt.query_map([18], |row| {
        Ok((
            row.get::<_, i64>(0)?,
            row.get::<_, String>(1)?,
            row.get::<_, String>(2)?,
            row.get::<_, i64>(3)?,
        ))
    })?;

    for user in users {
        let (id, name, email, age) = user?;
        println!("ID={}, Name={}, Email={}, Age={}", id, name, email, age);
    }

    Ok(())
}

# Cargo.toml
[dependencies]
rusqlite = { version = "0.31", features = ["bundled"] }

18.6.2 Rust 结构体映射

use rusqlite::{Connection, Result, Row};

#[derive(Debug)]
struct User {
    id: i64,
    name: String,
    email: String,
    age: i64,
}

impl User {
    fn from_row(row: &Row) -> Result<Self> {
        Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
            age: row.get(3)?,
        })
    }

    fn create_table(conn: &Connection) -> Result<()> {
        conn.execute("
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                email TEXT UNIQUE NOT NULL,
                age INTEGER DEFAULT 0
            )
        ", [])?;
        Ok(())
    }

    fn insert(&self, conn: &Connection) -> Result<i64> {
        conn.execute(
            "INSERT INTO users (name, email, age) VALUES (?1, ?2, ?3)",
            rusqlite::params![self.name, self.email, self.age],
        )?;
        Ok(conn.last_insert_rowid())
    }

    fn find_all(conn: &Connection) -> Result<Vec<User>> {
        let mut stmt = conn.prepare("SELECT id, name, email, age FROM users")?;
        let users = stmt.query_map([], User::from_row)?.collect::<Result<Vec<_>>>()?;
        Ok(users)
    }
}

18.6.3 Rust 事务

use rusqlite::{Connection, Transaction, Result};

fn transfer(conn: &Connection, from: i64, to: i64, amount: f64) -> Result<()> {
    let tx = conn.unchecked_transaction()?;

    let balance: f64 = tx.query_row(
        "SELECT balance FROM accounts WHERE id = ?1",
        [from],
        |row| row.get(0),
    )?;

    if balance < amount {
        tx.rollback()?;
        return Err(rusqlite::Error::ExecuteReturnedResults);
    }

    tx.execute("UPDATE accounts SET balance = balance - ?1 WHERE id = ?2", rusqlite::params![amount, from])?;
    tx.execute("UPDATE accounts SET balance = balance + ?1 WHERE id = ?2", rusqlite::params![amount, to])?;

    tx.commit()?;
    Ok(())
}

18.7 驱动对比

特性	Python sqlite3	Go modernc	Java JDBC	better-sqlite3	rusqlite
内置	✅	❌	❌	❌	❌
CGO	—	❌	—	✅	❌ (bundled)
同步 API	✅	✅	✅	✅	✅
异步 API	❌	❌	❌	❌	❌
事务支持	✅	✅	✅	✅	✅
WAL 支持	✅	✅	✅	✅	✅
扩展加载	✅	❌	❌	✅	✅
性能	好	好	好	最好	最好

⚠️ 注意事项

SQLite 是同步的——不要使用阻塞事件循环的驱动——Node.js 中 better-sqlite3 是同步的，适合短查询
连接不要跨线程共享——每个线程一个连接，或使用连接池
WAL 模式下不要在网络文件系统上使用——NFS/SMB 上文件锁不可靠
PRAGMA foreign_keys = ON 每次连接都要设置——不持久化
批量操作一定要使用事务——否则性能差 100 倍以上
Python 的 check_same_thread=False——允许跨线程使用连接（需自行确保线程安全）

💡 技巧

Python 的 sqlite3.Row 可以像字典一样访问列
better-sqlite3 的事务 API 非常优雅——db.transaction(fn)(args)
Go 的 modernc.org/sqlite 不需要 CGO——适合交叉编译
Rust 的 rusqlite with bundled 自动编译 SQLite——无需系统依赖
所有驱动都支持参数化查询——务必使用，不要拼接 SQL

📌 业务场景

场景一：Python CLI 工具

import sqlite3
import click

@click.command()
@click.argument('db_path')
@click.option('--name', prompt='用户名')
def add_user(db_path, name):
    conn = sqlite3.connect(db_path)
    conn.execute('INSERT INTO users (name) VALUES (?)', (name,))
    conn.commit()
    click.echo(f'用户 {name} 已添加')

if __name__ == '__main__':
    add_user()

场景二：Node.js REST API

const Database = require('better-sqlite3');
const express = require('express');

const db = new Database('app.db');
db.pragma('journal_mode = WAL');

const app = express();
app.use(express.json());

app.get('/products', (req, res) => {
    const products = db.prepare(
        'SELECT * FROM products WHERE category = ? ORDER BY price'
    ).all(req.query.category || 'all');
    res.json(products);
});

app.listen(3000);

🔗 扩展阅读

📖 下一章：19 - 最佳实践 —— Schema 设计、并发策略、何时不用 SQLite