NeuralBlitz SQLite
What I Do
I provide guidance on SQLite, the lightweight serverless database engine. I help with schema design, query optimization, transactions, full-text search extensions, JSON support, and best practices for embedded and mobile applications.
When to Use Me
- Mobile applications (iOS, Android)
- Embedded systems and IoT devices
- Development and testing environments
- Desktop applications with local storage
- Small to medium web applications
- Caching and temporary data storage
- CLI tools and scripts
- Microservices with simple data needs
Core Concepts
- Serverless: No separate database process
- Single-File: Entire database in one file
- ACID: Atomic, Consistent, Isolated, Durable
- Transactions: Full transaction support
- SQLite3: Latest SQLite version (Python module)
- Connection: File-based connections
- Prepared Statements: Parameterized queries
- Row Factories: Custom result formatting
- Extensions: FTS5, JSON1, RTree
- WAL Mode: Write-Ahead Logging for concurrency
Code Examples
Basic Connection and CRUD
import sqlite3
from typing import Optional, List, Dict
def get_connection(db_path: str = "app.db") -> sqlite3.Connection:
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
return conn
def create_tables(conn: sqlite3.Connection) -> None:
conn.executescript("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email TEXT UNIQUE NOT NULL,
name TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS posts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
title TEXT NOT NULL,
content TEXT,
published_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id)
);
""")
def insert_user(conn: sqlite3.Connection, email: str, name: str) -> int:
cursor = conn.execute(
"INSERT INTO users (email, name) VALUES (?, ?)",
(email, name)
)
conn.commit()
return cursor.lastrowid
def get_user(conn: sqlite3.Connection, user_id: int) -> Optional[sqlite3.Row]:
cursor = conn.execute(
"SELECT * FROM users WHERE id = ?",
(user_id,)
)
return cursor.fetchone()
Transactions and Error Handling
import sqlite3
from sqlite3 import Error
def transfer_credits(
conn: sqlite3.Connection,
from_user: int,
to_user: int,
amount: float
) -> bool:
try:
cursor = conn.cursor()
cursor.execute("BEGIN TRANSACTION")
cursor.execute(
"SELECT credits FROM users WHERE id = ? FOR UPDATE",
(from_user,)
)
result = cursor.fetchone()
if not result or result[0] < amount:
cursor.execute("ROLLBACK")
return False
cursor.execute(
"UPDATE users SET credits = credits - ? WHERE id = ?",
(amount, from_user)
)
cursor.execute(
"UPDATE users SET credits = credits + ? WHERE id = ?",
(amount, to_user)
)
cursor.execute("COMMIT")
return True
except Error as e:
conn.rollback()
raise e
Full-Text Search with FTS5
import sqlite3
def create_fts_table(conn: sqlite3.Connection) -> None:
conn.executescript("""
CREATE VIRTUAL TABLE documents USING fts5(
title, content, tokenize='porter unicode61'
);
""")
def search_documents(conn: sqlite3.Connection, query: str) -> list:
cursor = conn.execute(
"""
SELECT rowid, title, snippet(documents, 2, '<b>', '</b>', '...', 10)
FROM documents
WHERE documents MATCH ?
ORDER BY bm25(documents)
LIMIT 20
""",
(query,)
)
return [{"id": row[0], "title": row[1], "snippet": row[2]} for row in cursor]
def add_document(conn: sqlite3.Connection, title: str, content: str) -> int:
cursor = conn.execute(
"INSERT INTO documents (title, content) VALUES (?, ?)",
(title, content)
)
conn.commit()
return cursor.lastrowid
JSON Support (JSON1 Extension)
import sqlite3
import json
def store_json_data(conn: sqlite3.Connection, user_id: int, data: dict) -> None:
conn.execute(
"""
INSERT INTO user_profiles (user_id, data)
VALUES (?, ?)
ON CONFLICT(user_id) DO UPDATE SET data = ?
""",
(user_id, json.dumps(data), json.dumps(data))
)
conn.commit()
def query_by_json_field(
conn: sqlite3.Connection,
field_path: str,
value: str
) -> list:
cursor = conn.execute(
"""
SELECT user_id, data FROM user_profiles
WHERE json_extract(data, ?) = ?
""",
(field_path, value)
)
return [{"user_id": row[0], "data": json.loads(row[1])} for row in cursor]
Best Practices
- Use context managers for connections
- Always use parameterized queries (prevent SQL injection)
- Enable foreign keys with PRAGMA foreign_keys = ON
- Use WAL mode for better concurrency: PRAGMA journal_mode=WAL
- Create appropriate indexes on queried columns
- Use appropriate data types (INTEGER for IDs, TEXT for strings)
- Implement proper backup strategies (sqlite3 backup API)
- Use FTS5 for full-text search requirements
- Handle database locking gracefully (retry on busy)
- Vacuum periodically to reclaim space
Common Patterns
Connection Pool (for threaded apps):
import sqlite3
import threading
from queue import Queue
class SQLitePool:
def __init__(self, db_path: str, size: int = 5):
self.db_path = db_path
self.pool = Queue(size)
for _ in range(size):
conn = sqlite3.connect(db_path, check_same_thread=False)
conn.row_factory = sqlite3.Row
self.pool.put(conn)
def get_connection(self):
return self.pool.get(timeout=5)
def return_connection(self, conn):
self.pool.put(conn)
Atomic Database Copy (Hot Backup):
import sqlite3
def backup_database(source: str, dest: str) -> None:
source_conn = sqlite3.connect(source)
dest_conn = sqlite3.connect(dest)
source_conn.backup(dest_conn)
dest_conn.close()
source_conn.close()
Upsert Pattern:
INSERT INTO stats (key, value) VALUES ('page_views', 1)
ON CONFLICT(key) DO UPDATE SET value = stats.value + 1;