NeuralBlitz SQLite
What I do
I am a self-contained, serverless, zero-configuration, transactional SQL database engine. I store the entire database in a single file, making me ideal for embedded systems, mobile applications, testing environments, and small to medium web applications. Despite my simplicity, I support most of the SQL standard, including ACID transactions, views, triggers, and full-text search extensions. I am the most widely deployed database engine in the world.
When to use me
- Mobile applications (Android, iOS)
- Desktop applications requiring local storage
- Browser extensions and plugins
- Testing and development environments
- Small to medium websites with low to medium traffic
- Embedded systems and IoT devices
- Data analysis and prototyping
- Caching and session storage
- Configuration and settings storage
- Temporary databases for ETL processes
Core Concepts
- Serverless Architecture: No separate database server process; database is a single file
- Zero Configuration: No setup, installation, or administration required
- ACID Transactions: Full support for atomic, consistent, isolated, durable transactions
- Single File Storage: Entire database stored in one portable file
- SQLite3 File Format: Stable, cross-platform, and backwards-compatible file format
- WAL Mode: Write-Ahead Logging for better concurrent read/write performance
- Full-Text Search (FTS): Virtual table module for efficient full-text search
- Rowid and Primary Keys: Auto-incrementing rowid or explicit primary key selection
- PRAGMA Statements: Configuration and diagnostic statements for tuning
- Connection Limits: Recommended connection pool size of 1-2 for file-based locking
Code Examples
Basic Connection and CRUD
import sqlite3
from contextlib import contextmanager
@contextmanager
def get_connection(db_path="app.db"):
conn = sqlite3.connect(db_path, timeout=30.0)
conn.row_factory = sqlite3.Row
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def init_database():
with get_connection() as conn:
conn.executescript("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email TEXT UNIQUE NOT NULL,
name TEXT NOT NULL,
password_hash TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
SKU TEXT UNIQUE NOT NULL,
name TEXT NOT NULL,
description TEXT,
price REAL NOT NULL,
category TEXT,
stock INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_users_email ON users(email);
CREATE INDEX IF NOT EXISTS idx_products_sku ON products(SKU);
CREATE INDEX IF NOT EXISTS idx_products_category ON products(category);
""")
def create_user(email, name, password_hash):
with get_connection() as conn:
cur = conn.execute("""
INSERT INTO users (email, name, password_hash)
VALUES (?, ?, ?)
""", (email, name, password_hash))
return cur.lastrowid
def get_user_by_id(user_id):
with get_connection() as conn:
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
return dict(row) if row else None
def get_user_by_email(email):
with get_connection() as conn:
row = conn.execute("SELECT * FROM users WHERE email = ?", (email,)).fetchone()
return dict(row) if row else None
def update_user(user_id, **updates):
set_clauses = [f"{k} = ?" for k in updates.keys()]
params = list(updates.values()) + [user_id]
with get_connection() as conn:
conn.execute(f"""
UPDATE users SET {', '.join(set_clauses)}, updated_at = CURRENT_TIMESTAMP
WHERE id = ?
""", params)
def delete_user(user_id):
with get_connection() as conn:
conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
return conn.total_changes > 0
Transactions and Concurrency
from sqlite3 import IntegrityError
def transfer_funds(from_account, to_account, amount):
with get_connection() as conn:
try:
cur = conn.execute("SELECT balance FROM accounts WHERE id = ?", (from_account,))
balance = cur.fetchone()
if not balance or balance[0] < amount:
raise ValueError("Insufficient funds")
conn.execute("""
UPDATE accounts SET balance = balance - ? WHERE id = ?
""", (amount, from_account))
conn.execute("""
UPDATE accounts SET balance = balance + ? WHERE id = ?
""", (amount, to_account))
conn.execute("""
INSERT INTO transactions (from_account, to_account, amount)
VALUES (?, ?, ?)
""", (from_account, to_account, amount))
return True
except IntegrityError as e:
conn.rollback()
raise ValueError(f"Transfer failed: {e}")
def batch_create_orders(orders_data):
with get_connection() as conn:
try:
order_ids = []
for order in orders_data:
cur = conn.execute("""
INSERT INTO orders (user_id, total, status)
VALUES (?, ?, 'pending')
""", (order["user_id"], order["total"]))
order_id = cur.lastrowid
for item in order["items"]:
conn.execute("""
INSERT INTO order_items (order_id, product_id, quantity, price)
VALUES (?, ?, ?, ?)
""", (order_id, item["product_id"], item["quantity"], item["price"]))
order_ids.append(order_id)
return order_ids
except Exception as e:
conn.rollback()
raise e
def safe_concurrent_update(key, value, expected_version):
with get_connection() as conn:
cur = conn.execute("""
UPDATE settings SET value = ?, version = version + 1
WHERE key = ? AND version = ?
""", (value, key, expected_version))
if cur.rowcount == 0:
raise ValueError("Concurrent modification detected")
return True
Advanced Queries and Data Analysis
def get_sales_summary(start_date, end_date):
with get_connection() as conn:
rows = conn.execute("""
SELECT
DATE(o.created_at) as sale_date,
COUNT(*) as order_count,
SUM(o.total) as total_revenue,
AVG(o.total) as avg_order_value
FROM orders o
WHERE o.created_at BETWEEN ? AND ?
GROUP BY DATE(o.created_at)
ORDER BY sale_date
""", (start_date, end_date)).fetchall()
return [{"date": row[0], "orders": row[1], "revenue": row[2], "avg_order": row[3]} for row in rows]
def get_top_products(limit=10):
with get_connection() as conn:
rows = conn.execute("""
SELECT
p.id, p.name, p.category,
COUNT(oi.id) as total_sold,
SUM(oi.quantity) as total_quantity,
SUM(oi.quantity * oi.price) as total_revenue
FROM products p
LEFT JOIN order_items oi ON p.id = oi.product_id
LEFT JOIN orders o ON oi.order_id = o.id AND o.status != 'cancelled'
GROUP BY p.id
ORDER BY total_revenue DESC
LIMIT ?
""", (limit,)).fetchall()
return [dict(row) for row in rows]
def get_user_statistics():
with get_connection() as conn:
stats = {}
stats["total_users"] = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]
stats["users_by_month"] = conn.execute("""
SELECT strftime('%Y-%m', created_at) as month, COUNT(*) as count
FROM users
GROUP BY month
ORDER BY month DESC
""").fetchall()
stats["active_orders"] = conn.execute("""
SELECT COUNT(*) FROM orders WHERE status NOT IN ('cancelled', 'delivered')
"").fetchone()[0]
return stats
def get_running_totals():
with get_connection() as conn:
conn.execute("""
SELECT
date,
revenue,
SUM(revenue) OVER (ORDER BY date) as running_total
FROM daily_revenue
ORDER BY date
""")
def recursive_category_tree():
with get_connection() as conn:
conn.execute("""
WITH RECURSIVE category_tree AS (
SELECT id, name, parent_id, 0 as level
FROM categories WHERE parent_id IS NULL
UNION ALL
SELECT c.id, c.name, c.parent_id, ct.level + 1
FROM categories c
JOIN category_tree ct ON c.parent_id = ct.id
)
SELECT * FROM category_tree ORDER BY level, name
""")
Full-Text Search Extension
def create_fts_table():
with get_connection() as conn:
conn.executescript("""
CREATE VIRTUAL TABLE products_fts USING fts5(
name, description, category,
content='products',
content_rowid='id'
);
CREATE TRIGGER products_ai AFTER INSERT ON products BEGIN
INSERT INTO products_fts(rowid, name, description, category)
VALUES (new.id, new.name, new.description, new.category);
END;
CREATE TRIGGER products_ad AFTER DELETE ON products BEGIN
INSERT INTO products_fts(products_fts, rowid, name, description, category)
VALUES ('delete', old.id, old.name, old.description, old.category);
END;
CREATE TRIGGER products_au AFTER UPDATE ON products BEGIN
INSERT INTO products_fts(products_fts, rowid, name, description, category)
VALUES ('delete', old.id, old.name, old.description, old.category);
INSERT INTO products_fts(rowid, name, description, category)
VALUES (new.id, new.name, new.description, new.category);
END;
""")
def search_products(query, limit=50):
with get_connection() as conn:
rows = conn.execute("""
SELECT p.*, bm25(products_fts) as score
FROM products_fts
JOIN products p ON products_fts.rowid = p.id
WHERE products_fts MATCH ?
ORDER BY score
LIMIT ?
""", (query, limit)).fetchall()
return [dict(row) for row in rows]
def search_with_highlight(query):
with get_connection() as conn:
rows = conn.execute("""
SELECT p.id, p.name, p.description,
snippet(products_fts, 0, '<b>', '</b>', '...', 10) as name_snippet,
snippet(products_fts, 1, '<b>', '</b>', '...', 10) as desc_snippet
FROM products_fts
JOIN products p ON products_fts.rowid = p.id
WHERE products_fts MATCH ?
LIMIT 20
""", (query,)).fetchall()
return [dict(row) for row in rows]
Backup and Maintenance
import shutil
import os
def backup_database(source_path, backup_path):
shutil.copy2(source_path, backup_path)
return backup_path
def vacuum_database(db_path="app.db"):
with get_connection(db_path) as conn:
conn.execute("VACUUM")
def integrity_check(db_path="app.db"):
with get_connection(db_path) as conn:
rows = conn.execute("PRAGMA integrity_check").fetchall()
return all(row[0] == "ok" for row in rows)
def get_table_info(table_name):
with get_connection() as conn:
rows = conn.execute(f"PRAGMA table_info({table_name})").fetchall()
return [{"name": row[1], "type": row[2], "notnull": row[3],
"pk": row[5]} for row in rows]
def get_database_size(db_path="app.db"):
return os.path.getsize(db_path) if os.path.exists(db_path) else 0
def enable_wal_mode(db_path="app.db"):
with get_connection(db_path) as conn:
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute("PRAGMA cache_size=-64000")
conn.execute("PRAGMA temp_store=MEMORY")
def show_tables():
with get_connection() as conn:
rows = conn.execute("""
SELECT name FROM sqlite_master
WHERE type='table' AND name NOT LIKE 'sqlite_%'
ORDER BY name
""").fetchall()
return [row[0] for row in rows]
Best Practices
- Use WAL Mode for Better Concurrency: Enable WAL mode for concurrent read/write performance
- Use Parameterized Queries: Always use parameterized queries to prevent SQL injection
- Implement Connection Pooling: For web applications, use connection pooling libraries
- Enable Foreign Keys: Execute PRAGMA foreign_keys=ON to enforce referential integrity
- Use Transactions for Bulk Operations: Wrap multiple operations in transactions for performance
- Add Indexes for Query Optimization: Create indexes on frequently queried columns
- Regular Maintenance: Run VACUUM periodically to reclaim space and optimize database
- Use Appropriate Data Types: SQLite is flexible but choose appropriate types for compatibility
- Handle Timeouts for Concurrent Access: Set appropriate timeout for busy databases
- Backup Regularly: Use online backup API or file copy for reliable backups