SQLite Database Expert by martinholovsky/claude-skills-generator
npx skills add https://github.com/martinholovsky/claude-skills-generator --skill 'SQLite Database Expert'关键提示:在执行任何数据库操作之前,您必须阅读相关的参考文件:
在以下情况下阅读 references/advanced-patterns.md:
在以下情况下阅读 references/security-examples.md:
风险等级:中等
理由:桌面应用程序中的 SQLite 数据库在本地处理用户数据,如果查询未正确参数化,会存在 SQL 注入风险,并且需要仔细的迁移管理以防止数据丢失。
您是 SQLite 嵌入式数据库开发专家,专长于:
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
PRAGMA foreign_keys = ON| 组件 | 推荐版本 | 最低版本 | 备注 |
|---|---|---|---|
| SQLite | 3.45+ | 3.35 | FTS5, JSON 函数 |
| rusqlite | 0.31+ | 0.29 | 捆绑 SQLite 支持 |
| sea-query | 0.30+ | 0.28 | 查询构建器 |
| r2d2 | 0.8+ | 0.8 | 连接池 |
[dependencies]
rusqlite = { version = "0.31", features = ["bundled", "backup", "functions"] }
sea-query = "0.30"
sea-query-rusqlite = "0.5"
r2d2 = "0.8"
r2d2_sqlite = "0.24"
use rusqlite::{Connection, Result};
use std::path::Path;
pub struct Database {
conn: Connection,
}
impl Database {
pub fn new(path: &Path) -> Result<Self> {
let conn = Connection::open(path)?;
// 启用安全和性能特性
conn.execute_batch("
PRAGMA foreign_keys = ON;
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA temp_store = MEMORY;
PRAGMA mmap_size = 30000000000;
PRAGMA page_size = 4096;
")?;
Ok(Self { conn })
}
}
// 正确:参数化查询
pub fn get_user_by_id(&self, user_id: i64) -> Result<Option<User>> {
let mut stmt = self.conn.prepare(
"SELECT id, name, email FROM users WHERE id = ?1"
)?;
let user = stmt.query_row([user_id], |row| {
Ok(User {
id: row.get(0)?,
name: row.get(1)?,
email: row.get(2)?,
})
}).optional()?;
Ok(user)
}
// 正确:使用命名参数以提高清晰度
pub fn search_users(&self, name: &str, status: &str) -> Result<Vec<User>> {
let mut stmt = self.conn.prepare(
"SELECT id, name, email FROM users
WHERE name LIKE :name AND status = :status"
)?;
let users = stmt.query_map(
&[(":name", &format!("%{}%", name)), (":status", &status)],
|row| Ok(User {
id: row.get(0)?,
name: row.get(1)?,
email: row.get(2)?,
})
)?.collect::<Result<Vec<_>>>()?;
Ok(users)
}
// 错误:SQL 注入漏洞
pub fn get_user_unsafe(&self, user_id: &str) -> Result<Option<User>> {
// 切勿这样做 - 存在 SQL 注入风险
let query = format!("SELECT * FROM users WHERE id = {}", user_id);
// ...
}
pub fn transfer_funds(
&mut self,
from_id: i64,
to_id: i64,
amount: f64
) -> Result<()> {
let tx = self.conn.transaction()?;
// 从源账户扣款
tx.execute(
"UPDATE accounts SET balance = balance - ?1 WHERE id = ?2",
[amount, from_id as f64],
)?;
// 向目标账户存款
tx.execute(
"UPDATE accounts SET balance = balance + ?1 WHERE id = ?2",
[amount, to_id as f64],
)?;
tx.commit()?;
Ok(())
}
// 使用触发器创建 FTS5 虚拟表
pub fn setup_fts(&self) -> Result<()> {
self.conn.execute_batch("
CREATE VIRTUAL TABLE IF NOT EXISTS docs_fts USING fts5(
title, content, tags, content=documents, content_rowid=id
);
CREATE TRIGGER IF NOT EXISTS docs_ai AFTER INSERT ON documents BEGIN
INSERT INTO docs_fts(rowid, title, content, tags)
VALUES (new.id, new.title, new.content, new.tags);
END;
")?;
Ok(())
}
// 带高亮显示的搜索
pub fn search_documents(&self, query: &str) -> Result<Vec<Document>> {
let mut stmt = self.conn.prepare(
"SELECT d.*, highlight(docs_fts, 1, '<mark>', '</mark>') as snippet
FROM documents d JOIN docs_fts ON d.id = docs_fts.rowid
WHERE docs_fts MATCH ?1 ORDER BY rank"
)?;
stmt.query_map([query], |row| Ok(Document { /* ... */ }))?.collect()
}
缓解措施:更新到 SQLite 3.44.0+ 并始终使用参数化查询。
| OWASP 类别 | 风险 | 关键控制措施 |
|---|---|---|
| A03 - 注入 | 严重 | 参数化查询,输入验证 |
| A04 - 不安全设计 | 中等 | 模式约束,外键 |
| A05 - 错误配置 | 中等 | 安全的 PRAGMA 设置,文件权限 (600) |
关键规则 (参见 references/security-examples.md):
切勿使用字符串格式化来构建 SQL 查询
始终使用 ? 位置参数或 :name 命名参数
对于动态查询,对列名/表名使用白名单
// 动态列选择 - 安全方法 pub fn get_user_fields(&self, user_id: i64, fields: &[&str]) -> Result<HashMap<String, String>> { const ALLOWED: &[&str] = &["id", "name", "email", "created_at"]; let safe_fields: Vec<&str> = fields.iter() .filter(|f| ALLOWED.contains(f)).copied().collect(); if safe_fields.is_empty() { return Err(rusqlite::Error::InvalidQuery); } let query = format!("SELECT {} FROM users WHERE id = ?1", safe_fields.join(", ")); let mut stmt = self.conn.prepare(&query)?; // ... }
#[cfg(test)]
mod tests {
use super::*;
use rusqlite::Connection;
fn setup_test_db() -> Database {
let conn = Connection::open_in_memory().unwrap();
let db = Database { conn };
db.run_migrations().unwrap();
db
}
#[test]
fn test_sql_injection_prevented() {
let db = setup_test_db();
let result = db.search_users("'; DROP TABLE users; --", "active");
assert!(result.is_ok());
assert!(db.get_user_by_id(1).is_ok()); // 表仍然存在
}
}
# tests/test_user_repository.py
import pytest
import sqlite3
@pytest.fixture
def db():
"""用于快速测试的内存 SQLite 数据库。"""
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
yield conn
conn.close()
class TestUserRepository:
def test_create_user_returns_id(self, db):
repo = UserRepository(db)
repo.initialize_schema()
user_id = repo.create_user("test@example.com", "Test User")
assert user_id > 0
def test_sql_injection_prevented(self, db):
repo = UserRepository(db)
repo.initialize_schema()
malicious = "'; DROP TABLE users; --"
user_id = repo.create_user(malicious, "Hacker")
assert repo.get_by_id(user_id)["email"] == malicious
# app/repositories/user.py
class UserRepository:
def __init__(self, conn):
self.conn = conn
def initialize_schema(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email TEXT NOT NULL UNIQUE,
name TEXT NOT NULL
)""")
self.conn.commit()
def create_user(self, email: str, name: str) -> int:
cursor = self.conn.execute(
"INSERT INTO users (email, name) VALUES (?, ?)", (email, name))
self.conn.commit()
return cursor.lastrowid
def get_by_id(self, user_id: int):
return self.conn.execute(
"SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
pytest tests/test_*_repository.py -v --cov=app/repositories
# 良好:启用 WAL 以支持并发读写
conn.execute("PRAGMA journal_mode = WAL")
conn.execute("PRAGMA synchronous = NORMAL")
conn.execute("PRAGMA cache_size = -64000") # 64MB
# 不良:默认的 DELETE 模式在写入时会阻塞读取
# 良好:批量操作使用单个事务
conn.executemany("INSERT INTO items (name) VALUES (?)", records)
conn.commit()
# 不良:每行都提交 (慢 100 倍)
for r in records:
conn.execute("INSERT INTO items (name) VALUES (?)", (r,))
conn.commit()
# 良好:复用连接
from queue import Queue
class ConnectionPool:
def __init__(self, db_path, size=5):
self.pool = Queue(size)
for _ in range(size):
conn = sqlite3.connect(db_path, check_same_thread=False)
conn.execute("PRAGMA journal_mode = WAL")
self.pool.put(conn)
# 不良:每个查询都新建连接
conn = sqlite3.connect(db_path) # 代价高昂!
# 良好:覆盖索引和部分索引
conn.executescript("""
CREATE INDEX idx_users_email ON users(email, name);
CREATE INDEX idx_active ON items(created_at) WHERE status='active';
ANALYZE;
""")
# 不良:在未索引的列上进行全表扫描
# 良好:在空闲时间进行维护
def nightly_maintenance(conn):
conn.execute("PRAGMA optimize")
freelist = conn.execute("PRAGMA freelist_count").fetchone()[0]
if freelist > 1000:
conn.execute("VACUUM")
# 不良:在高峰使用时段执行 VACUUM 或从不执行
| 错误 | 错误做法 | 正确做法 |
|---|---|---|
| SQL 注入 | format!("...WHERE name = '{}'", input) | "...WHERE name = ?1" 配合参数 |
| 无事务 | 单独的 execute 调用 | 包装在 transaction() 和 commit() 中 |
| 无外键 | 默认连接 | PRAGMA foreign_keys = ON |
| 使用 LIKE 搜索 | LIKE '%term%' | FTS5 MATCH 'term' |
references/security-examples.mdPRAGMA foreign_keys = ONexecutemany() 进行多次插入pytest tests/test_*_repository.py -v创建 安全 (参数化查询)、可靠 (事务、外键) 且 高性能 (WAL 模式、索引、FTS5) 的 SQLite 实现。
安全提醒:切勿将用户输入拼接到 SQL 中。始终使用参数化查询。
每周安装次数
–
仓库
GitHub 星标数
29
首次出现时间
–
安全审计
CRITICAL : Before implementing ANY database operation, you MUST read the relevant reference files:
Readreferences/advanced-patterns.md WHEN:
Readreferences/security-examples.md WHEN:
Risk Level: MEDIUM
Justification : SQLite databases in desktop applications handle user data locally, present SQL injection risks if queries aren't properly parameterized, and require careful migration management to prevent data loss.
You are an expert in SQLite embedded database development, specializing in:
PRAGMA foreign_keys = ON| Component | Recommended | Minimum | Notes |
|---|---|---|---|
| SQLite | 3.45+ | 3.35 | FTS5, JSON functions |
| rusqlite | 0.31+ | 0.29 | Bundled SQLite support |
| sea-query | 0.30+ | 0.28 | Query builder |
| r2d2 | 0.8+ | 0.8 | Connection pooling |
[dependencies]
rusqlite = { version = "0.31", features = ["bundled", "backup", "functions"] }
sea-query = "0.30"
sea-query-rusqlite = "0.5"
r2d2 = "0.8"
r2d2_sqlite = "0.24"
use rusqlite::{Connection, Result};
use std::path::Path;
pub struct Database {
conn: Connection,
}
impl Database {
pub fn new(path: &Path) -> Result<Self> {
let conn = Connection::open(path)?;
// Enable security and performance features
conn.execute_batch("
PRAGMA foreign_keys = ON;
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA temp_store = MEMORY;
PRAGMA mmap_size = 30000000000;
PRAGMA page_size = 4096;
")?;
Ok(Self { conn })
}
}
// CORRECT: Parameterized query
pub fn get_user_by_id(&self, user_id: i64) -> Result<Option<User>> {
let mut stmt = self.conn.prepare(
"SELECT id, name, email FROM users WHERE id = ?1"
)?;
let user = stmt.query_row([user_id], |row| {
Ok(User {
id: row.get(0)?,
name: row.get(1)?,
email: row.get(2)?,
})
}).optional()?;
Ok(user)
}
// CORRECT: Named parameters for clarity
pub fn search_users(&self, name: &str, status: &str) -> Result<Vec<User>> {
let mut stmt = self.conn.prepare(
"SELECT id, name, email FROM users
WHERE name LIKE :name AND status = :status"
)?;
let users = stmt.query_map(
&[(":name", &format!("%{}%", name)), (":status", &status)],
|row| Ok(User {
id: row.get(0)?,
name: row.get(1)?,
email: row.get(2)?,
})
)?.collect::<Result<Vec<_>>>()?;
Ok(users)
}
// INCORRECT: SQL Injection vulnerability
pub fn get_user_unsafe(&self, user_id: &str) -> Result<Option<User>> {
// NEVER DO THIS - SQL injection risk
let query = format!("SELECT * FROM users WHERE id = {}", user_id);
// ...
}
pub fn transfer_funds(
&mut self,
from_id: i64,
to_id: i64,
amount: f64
) -> Result<()> {
let tx = self.conn.transaction()?;
// Debit from source
tx.execute(
"UPDATE accounts SET balance = balance - ?1 WHERE id = ?2",
[amount, from_id as f64],
)?;
// Credit to destination
tx.execute(
"UPDATE accounts SET balance = balance + ?1 WHERE id = ?2",
[amount, to_id as f64],
)?;
tx.commit()?;
Ok(())
}
// Create FTS5 virtual table with triggers
pub fn setup_fts(&self) -> Result<()> {
self.conn.execute_batch("
CREATE VIRTUAL TABLE IF NOT EXISTS docs_fts USING fts5(
title, content, tags, content=documents, content_rowid=id
);
CREATE TRIGGER IF NOT EXISTS docs_ai AFTER INSERT ON documents BEGIN
INSERT INTO docs_fts(rowid, title, content, tags)
VALUES (new.id, new.title, new.content, new.tags);
END;
")?;
Ok(())
}
// Search with highlighting
pub fn search_documents(&self, query: &str) -> Result<Vec<Document>> {
let mut stmt = self.conn.prepare(
"SELECT d.*, highlight(docs_fts, 1, '<mark>', '</mark>') as snippet
FROM documents d JOIN docs_fts ON d.id = docs_fts.rowid
WHERE docs_fts MATCH ?1 ORDER BY rank"
)?;
stmt.query_map([query], |row| Ok(Document { /* ... */ }))?.collect()
}
Mitigation : Update to SQLite 3.44.0+ and always use parameterized queries.
| OWASP Category | Risk | Key Controls |
|---|---|---|
| A03 - Injection | Critical | Parameterized queries, input validation |
| A04 - Insecure Design | Medium | Schema constraints, foreign keys |
| A05 - Misconfiguration | Medium | Secure PRAGMAs, file permissions (600) |
Critical Rules (see references/security-examples.md):
? positional or :name named parameters// Dynamic column selection - SAFE approach
pub fn get_user_fields(&self, user_id: i64, fields: &[&str]) -> Result<HashMap<String, String>> {
const ALLOWED: &[&str] = &["id", "name", "email", "created_at"];
let safe_fields: Vec<&str> = fields.iter()
.filter(|f| ALLOWED.contains(f)).copied().collect();
if safe_fields.is_empty() { return Err(rusqlite::Error::InvalidQuery); }
let query = format!("SELECT {} FROM users WHERE id = ?1", safe_fields.join(", "));
let mut stmt = self.conn.prepare(&query)?;
// ...
}
#[cfg(test)]
mod tests {
use super::*;
use rusqlite::Connection;
fn setup_test_db() -> Database {
let conn = Connection::open_in_memory().unwrap();
let db = Database { conn };
db.run_migrations().unwrap();
db
}
#[test]
fn test_sql_injection_prevented() {
let db = setup_test_db();
let result = db.search_users("'; DROP TABLE users; --", "active");
assert!(result.is_ok());
assert!(db.get_user_by_id(1).is_ok()); // Table still exists
}
}
# tests/test_user_repository.py
import pytest
import sqlite3
@pytest.fixture
def db():
"""In-memory SQLite for fast testing."""
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
yield conn
conn.close()
class TestUserRepository:
def test_create_user_returns_id(self, db):
repo = UserRepository(db)
repo.initialize_schema()
user_id = repo.create_user("test@example.com", "Test User")
assert user_id > 0
def test_sql_injection_prevented(self, db):
repo = UserRepository(db)
repo.initialize_schema()
malicious = "'; DROP TABLE users; --"
user_id = repo.create_user(malicious, "Hacker")
assert repo.get_by_id(user_id)["email"] == malicious
# app/repositories/user.py
class UserRepository:
def __init__(self, conn):
self.conn = conn
def initialize_schema(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email TEXT NOT NULL UNIQUE,
name TEXT NOT NULL
)""")
self.conn.commit()
def create_user(self, email: str, name: str) -> int:
cursor = self.conn.execute(
"INSERT INTO users (email, name) VALUES (?, ?)", (email, name))
self.conn.commit()
return cursor.lastrowid
def get_by_id(self, user_id: int):
return self.conn.execute(
"SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
pytest tests/test_*_repository.py -v --cov=app/repositories
# Good: Enable WAL for concurrent read/write
conn.execute("PRAGMA journal_mode = WAL")
conn.execute("PRAGMA synchronous = NORMAL")
conn.execute("PRAGMA cache_size = -64000") # 64MB
# Bad: Default DELETE mode blocks reads during writes
# Good: Single transaction for batch
conn.executemany("INSERT INTO items (name) VALUES (?)", records)
conn.commit()
# Bad: Commit per row (100x slower)
for r in records:
conn.execute("INSERT INTO items (name) VALUES (?)", (r,))
conn.commit()
# Good: Reuse connections
from queue import Queue
class ConnectionPool:
def __init__(self, db_path, size=5):
self.pool = Queue(size)
for _ in range(size):
conn = sqlite3.connect(db_path, check_same_thread=False)
conn.execute("PRAGMA journal_mode = WAL")
self.pool.put(conn)
# Bad: New connection per query
conn = sqlite3.connect(db_path) # Expensive!
# Good: Covering and partial indexes
conn.executescript("""
CREATE INDEX idx_users_email ON users(email, name);
CREATE INDEX idx_active ON items(created_at) WHERE status='active';
ANALYZE;
""")
# Bad: Full table scan on unindexed columns
# Good: Maintenance during idle time
def nightly_maintenance(conn):
conn.execute("PRAGMA optimize")
freelist = conn.execute("PRAGMA freelist_count").fetchone()[0]
if freelist > 1000:
conn.execute("VACUUM")
# Bad: VACUUM during peak usage or never
| Mistake | Wrong | Correct |
|---|---|---|
| SQL Injection | format!("...WHERE name = '{}'", input) | "...WHERE name = ?1" with params |
| No Transaction | Separate execute calls | Wrap in transaction() + commit() |
| No Foreign Keys | Default connection | PRAGMA foreign_keys = ON |
| LIKE for Search | LIKE '%term%' |
references/security-examples.md if handling user inputPRAGMA foreign_keys = ON at connectionexecutemany() for multiple insertspytest tests/test_*_repository.py -vCreate SQLite implementations that are Secure (parameterized queries), Reliable (transactions, foreign keys), and Performant (WAL mode, indexing, FTS5).
Security Reminder : NEVER concatenate user input into SQL. ALWAYS use parameterized queries.
Weekly Installs
–
Repository
GitHub Stars
29
First Seen
–
Security Audits
xdrop 文件传输脚本:Bun 环境下安全上传下载工具,支持加密分享
28,800 周安装
FTS5 MATCH 'term' |