Introduction
SQLite’s versatility makes it suitable for a wide range of applications beyond simple embedded use cases. From mobile applications to IoT devices, from caching layers to analytical workloads, SQLite powers critical functionality in countless applications worldwide.
This article explores real-world SQLite use cases, implementation patterns, and best practices for production deployments.
Mobile Application Development
iOS Applications
import SQLite
// Swift example using SQLite.swift
class DataManager {
private var db: Connection?
func setup() throws {
let path = NSSearchPathForDirectoriesInDomains(
.documentDirectory, .userDomainMask, true
).first!
db = try Connection("\(path)/app.sqlite3")
// Create tables
try db?.run(users.create(ifNotExists: true) { t in
t.column(id, primaryKey: .autoincrement)
t.column(name)
t.column(email, unique: true)
t.column(createdAt, defaultValue: Date())
})
try db?.run(notes.create(ifNotExists: true) { t in
t.column(id, primaryKey: .autoincrement)
t.column(userId, references: users, id)
t.column(title)
t.column(content)
t.column(updatedAt)
})
}
// CRUD operations
func createUser(name: String, email: String) throws -> Int64 {
let insert = users.insert(
name <- name,
email <- email,
createdAt <- Date()
)
return try db!.run(insert)
}
func getUserNotes(userId: Int64) throws -> [Note] {
let query = notes.filter(self.userId == userId)
return try db!.prepare(query).map { row in
Note(
id: row[id],
title: row[title],
content: row[content]
)
}
}
}
Android Applications
// Kotlin with Room (SQLite abstraction)
@Entity(tableName = "users")
data class User(
@PrimaryKey(autoGenerate = true) val id: Long = 0,
val name: String,
val email: String,
@ColumnInfo(defaultValue = "CURRENT_TIMESTAMP")
val createdAt: String
)
@Entity(
tableName = "notes",
foreignKeys = [
ForeignKey(
entity = User::class,
parentColumns = ["id"],
childColumns = ["userId"],
onDelete = ForeignKey.CASCADE
)
]
)
data class Note(
@PrimaryKey(autoGenerate = true) val id: Long = 0,
val userId: Long,
val title: String,
val content: String,
val updatedAt: Long = System.currentTimeMillis()
)
@Dao
interface NoteDao {
@Query("SELECT * FROM notes WHERE userId = :userId ORDER BY updatedAt DESC")
fun getNotesByUser(userId: Long): Flow<List<Note>>
@Insert(onConflict = OnConflictStrategy.REPLACE)
suspend fun insert(note: Note): Long
@Query("DELETE FROM notes WHERE id = :noteId")
suspend fun delete(noteId: Long)
}
@Database(entities = [User::class, Note::class], version = 1)
abstract class AppDatabase : RoomDatabase() {
abstract fun noteDao(): NoteDao
}
Cross-Platform with React Native
// React Native with expo-sqlite
import * as SQLite from 'expo-sqlite';
const db = SQLite.openDatabaseSync('app.db');
// Initialize database
export function initDatabase() {
db.execSync(`
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS tasks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
title TEXT NOT NULL,
completed INTEGER DEFAULT 0,
due_date TEXT,
FOREIGN KEY(user_id) REFERENCES users(id)
);
CREATE INDEX IF NOT EXISTS idx_tasks_user ON tasks(user_id);
`);
}
// CRUD operations
export function createTask(userId, title, dueDate) {
return db.runAsync(
'INSERT INTO tasks (user_id, title, due_date) VALUES (?, ?, ?)',
[userId, title, dueDate]
);
}
export function getTasks(userId) {
return db.getAllAsync(
'SELECT * FROM tasks WHERE user_id = ? ORDER BY completed, due_date',
[userId]
);
}
export function toggleTaskComplete(taskId) {
return db.runAsync(
'UPDATE tasks SET completed = NOT completed WHERE id = ?',
[taskId]
);
}
IoT and Embedded Systems
Sensor Data Storage
import sqlite3
import time
from datetime import datetime
class SensorDatabase:
"""Lightweight database for IoT sensor data."""
def __init__(self, db_path, max_rows=10000):
self.db_path = db_path
self.max_rows = max_rows
self._init_db()
def _init_db(self):
"""Initialize database with minimal footprint."""
conn = sqlite3.connect(self.db_path)
# Minimal PRAGMA for embedded
conn.execute("PRAGMA journal_mode = DELETE")
conn.execute("PRAGMA synchronous = OFF")
conn.execute("PRAGMA cache_size = -500") # 500KB
conn.execute("PRAGMA temp_store = MEMORY")
# Simple schema
conn.execute("""
CREATE TABLE IF NOT EXISTS readings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sensor_id TEXT NOT NULL,
value REAL NOT NULL,
timestamp INTEGER NOT NULL,
unit TEXT
)
""")
# Index for common queries
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_readings_sensor_time
ON readings(sensor_id, timestamp DESC)
""")
conn.commit()
conn.close()
def record_reading(self, sensor_id, value, unit=None):
"""Record a sensor reading."""
conn = sqlite3.connect(self.db_path)
conn.execute(
"INSERT INTO readings (sensor_id, value, timestamp, unit) VALUES (?, ?, ?, ?)",
(sensor_id, value, int(time.time()), unit)
)
conn.commit()
conn.close()
# Cleanup old data
self._cleanup()
def _cleanup(self):
"""Remove old readings to manage size."""
conn = sqlite3.connect(self.db_path)
# Keep only last max_rows
conn.execute("""
DELETE FROM readings
WHERE id NOT IN (
SELECT id FROM readings
ORDER BY timestamp DESC
LIMIT ?
)
""", (self.max_rows,))
conn.commit()
conn.close()
def get_recent(self, sensor_id, limit=100):
"""Get recent readings."""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.execute("""
SELECT * FROM readings
WHERE sensor_id = ?
ORDER BY timestamp DESC
LIMIT ?
""", (sensor_id, limit))
results = [dict(row) for row in cursor.fetchall()]
conn.close()
return results
# Usage
db = SensorDatabase('/data/sensors.db')
# Record sensor data
db.record_reading('temp_sensor_1', 23.5, 'celsius')
db.record_reading('humidity_sensor_1', 65.2, 'percent')
Data Collection with Offline Support
import sqlite3
import requests
import threading
class OfflineCollector:
"""Collect data offline, sync when connected."""
def __init__(self, db_path, api_endpoint):
self.db_path = db_path
self.api_endpoint = api_endpoint
self._init_db()
def _init_db(self):
"""Initialize pending data table."""
conn = sqlite3.connect(self.db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS pending_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
endpoint TEXT NOT NULL,
payload TEXT NOT NULL,
created_at INTEGER NOT NULL,
retry_count INTEGER DEFAULT 0
)
""")
conn.commit()
conn.close()
def collect(self, endpoint, payload):
"""Store data for later sync."""
import json
conn = sqlite3.connect(self.db_path)
conn.execute(
"INSERT INTO pending_data (endpoint, payload, created_at) VALUES (?, ?, ?)",
(endpoint, json.dumps(payload), int(time.time()))
)
conn.commit()
conn.close()
def sync(self):
"""Sync pending data to server."""
import json
conn = sqlite3.connect(self.db_path)
pending = conn.execute(
"SELECT * FROM pending_data ORDER BY created_at LIMIT 100"
).fetchall()
for row in pending:
try:
response = requests.post(
f"{self.api_endpoint}{row[1]}",
json=json.loads(row[2]),
timeout=10
)
if response.ok:
conn.execute("DELETE FROM pending_data WHERE id = ?", (row[0],))
else:
conn.execute(
"UPDATE pending_data SET retry_count = retry_count + 1 WHERE id = ?",
(row[0],)
)
except Exception as e:
print(f"Sync error: {e}")
conn.commit()
conn.close()
def start_background_sync(self, interval=60):
"""Start automatic background sync."""
def sync_worker():
while True:
time.sleep(interval)
self.sync()
thread = threading.Thread(target=sync_worker, daemon=True)
thread.start()
Caching Layer
Application Cache
import sqlite3
import json
import hashlib
import time
from functools import wraps
class Cache:
"""SQLite-based cache with TTL."""
def __init__(self, db_path, default_ttl=3600):
self.db_path = db_path
self.default_ttl = default_ttl
self._init_db()
def _init_db(self):
"""Initialize cache table."""
conn = sqlite3.connect(self.db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS cache (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
created_at INTEGER NOT NULL,
expires_at INTEGER NOT NULL,
metadata TEXT
)
""")
conn.execute("PRAGMA journal_mode = WAL")
conn.commit()
conn.close()
def get(self, key):
"""Get cached value if not expired."""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT * FROM cache WHERE key = ? AND expires_at > ?",
(key, int(time.time()))
).fetchone()
conn.close()
if row:
return json.loads(row['value'])
return None
def set(self, key, value, ttl=None):
"""Set cache value with TTL."""
ttl = ttl or self.default_ttl
now = int(time.time())
conn = sqlite3.connect(self.db_path)
conn.execute(
"""
INSERT OR REPLACE INTO cache (key, value, created_at, expires_at)
VALUES (?, ?, ?, ?)
""",
(key, json.dumps(value), now, now + ttl)
)
conn.commit()
conn.close()
def delete(self, key):
"""Delete cache entry."""
conn = sqlite3.connect(self.db_path)
conn.execute("DELETE FROM cache WHERE key = ?", (key,))
conn.commit()
conn.close()
def cleanup(self):
"""Remove expired entries."""
conn = sqlite3.connect(self.db_path)
conn.execute(
"DELETE FROM cache WHERE expires_at <= ?",
(int(time.time()),)
)
conn.commit()
conn.close()
# Decorator for caching function results
def cached(ttl=300):
"""Decorator to cache function results."""
_cache = Cache('function_cache.db', ttl)
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# Generate cache key
key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
key = hashlib.md5(key.encode()).hexdigest()
# Check cache
result = _cache.get(key)
if result is not None:
return result
# Execute and cache
result = func(*args, **kwargs)
_cache.set(key, result, ttl)
return result
return wrapper
return decorator
# Usage
@cached(ttl=600)
def expensive_query(param):
# Simulate expensive operation
time.sleep(2)
return f"Result for {param}"
Web Application Backend
Session Storage
import sqlite3
import secrets
import time
import json
class SessionManager:
"""SQLite-based session management."""
def __init__(self, db_path, session_ttl=86400):
self.db_path = db_path
self.session_ttl = session_ttl
self._init_db()
def _init_db(self):
"""Initialize session table."""
conn = sqlite3.connect(self.db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS sessions (
session_id TEXT PRIMARY KEY,
user_id INTEGER NOT NULL,
data TEXT NOT NULL,
created_at INTEGER NOT NULL,
expires_at INTEGER NOT NULL,
ip_address TEXT,
user_agent TEXT
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_sessions_expires
ON sessions(expires_at)
""")
conn.execute("PRAGMA journal_mode = WAL")
conn.commit()
conn.close()
def create(self, user_id, ip=None, user_agent=None, data=None):
"""Create new session."""
session_id = secrets.token_urlsafe(32)
now = int(time.time())
conn = sqlite3.connect(self.db_path)
conn.execute(
"""
INSERT INTO sessions (session_id, user_id, data, created_at, expires_at, ip_address, user_agent)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(session_id, user_id, json.dumps(data or {}), now, now + self.session_ttl, ip, user_agent)
)
conn.commit()
conn.close()
return session_id
def get(self, session_id):
"""Get session data."""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT * FROM sessions WHERE session_id = ? AND expires_at > ?",
(session_id, int(time.time()))
).fetchone()
conn.close()
if row:
return {
'session_id': row['session_id'],
'user_id': row['user_id'],
'data': json.loads(row['data']),
'created_at': row['created_at'],
'expires_at': row['expires_at']
}
return None
def update(self, session_id, data):
"""Update session data."""
conn = sqlite3.connect(self.db_path)
conn.execute(
"UPDATE sessions SET data = ?, expires_at = ? WHERE session_id = ?",
(json.dumps(data), int(time.time()) + self.session_ttl, session_id)
)
conn.commit()
conn.close()
def destroy(self, session_id):
"""Delete session."""
conn = sqlite3.connect(self.db_path)
conn.execute("DELETE FROM sessions WHERE session_id = ?", (session_id,))
conn.commit()
conn.close()
def cleanup(self):
"""Remove expired sessions."""
conn = sqlite3.connect(self.db_path)
conn.execute(
"DELETE FROM sessions WHERE expires_at <= ?",
(int(time.time()),)
)
conn.commit()
conn.close()
# Flask integration
from flask import Flask, request, session
app = Flask(__name__)
sessions = SessionManager('sessions.db')
@app.before_request
def load_session():
session_id = request.cookies.get('session_id')
if session_id:
user_session = sessions.get(session_id)
if user_session:
session['user_id'] = user_session['user_id']
session['data'] = user_session['data']
@app.route('/login', methods=['POST'])
def login():
user_id = authenticate(request.form)
session_id = sessions.create(user_id)
response = make_response(redirect('/dashboard'))
response.set_cookie('session_id', session_id)
return response
Content Management
import sqlite3
import markdown
class ContentManager:
"""Simple CMS with SQLite."""
def __init__(self, db_path):
self.db_path = db_path
self._init_db()
def _init_db(self):
"""Initialize content tables."""
conn = sqlite3.connect(self.db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS pages (
slug TEXT PRIMARY KEY,
title TEXT NOT NULL,
content TEXT NOT NULL,
template TEXT DEFAULT 'default',
published INTEGER DEFAULT 0,
created_at TEXT DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS redirects (
from_path TEXT PRIMARY KEY,
to_path TEXT NOT NULL,
status_code INTEGER DEFAULT 301
)
""")
conn.commit()
conn.close()
def get_page(self, slug):
"""Get published page."""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT * FROM pages WHERE slug = ? AND published = 1",
(slug,)
).fetchone()
conn.close()
if row:
return {
'title': row['title'],
'content': markdown.markdown(row['content']),
'template': row['template']
}
return None
def create_page(self, slug, title, content, published=False):
"""Create or update page."""
conn = sqlite3.connect(self.db_path)
conn.execute("""
INSERT OR REPLACE INTO pages (slug, title, content, published, updated_at)
VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
""", (slug, title, content, 1 if published else 0))
conn.commit()
conn.close()
Analytics and Reporting
Event Tracking
import sqlite3
import json
import time
from datetime import datetime, timedelta
class Analytics:
"""SQLite-based analytics for events."""
def __init__(self, db_path):
self.db_path = db_path
self._init_db()
def _init_db(self):
"""Initialize analytics tables."""
conn = sqlite3.connect(self.db_path)
# Events table
conn.execute("""
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
event_type TEXT NOT NULL,
user_id TEXT,
session_id TEXT,
properties TEXT,
timestamp INTEGER NOT NULL
)
""")
# Index for common queries
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_events_type_time
ON events(event_type, timestamp)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_events_user_time
ON events(user_id, timestamp)
""")
conn.commit()
conn.close()
def track(self, event_type, user_id=None, session_id=None, properties=None):
"""Track an event."""
conn = sqlite3.connect(self.db_path)
conn.execute(
"""
INSERT INTO events (event_type, user_id, session_id, properties, timestamp)
VALUES (?, ?, ?, ?, ?)
""",
(event_type, user_id, session_id, json.dumps(properties or {}), int(time.time()))
)
conn.commit()
conn.close()
def get_event_count(self, event_type, start_time, end_time):
"""Count events in time range."""
conn = sqlite3.connect(self.db_path)
count = conn.execute("""
SELECT COUNT(*) FROM events
WHERE event_type = ? AND timestamp BETWEEN ? AND ?
""", (event_type, start_time, end_time)).fetchone()[0]
conn.close()
return count
def get_unique_users(self, start_time, end_time):
"""Count unique users in time range."""
conn = sqlite3.connect(self.db_path)
count = conn.execute("""
SELECT COUNT(DISTINCT user_id) FROM events
WHERE user_id IS NOT NULL
AND timestamp BETWEEN ? AND ?
""", (start_time, end_time)).fetchone()[0]
conn.close()
return count
def get_top_events(self, limit=10):
"""Get most common events."""
conn = sqlite3.connect(self.db_path)
results = conn.execute("""
SELECT event_type, COUNT(*) as count
FROM events
GROUP BY event_type
ORDER BY count DESC
LIMIT ?
""", (limit,)).fetchall()
conn.close()
return results
def get_daily_summary(self, days=7):
"""Get daily event summary."""
conn = sqlite3.connect(self.db_path)
end_time = int(time.time())
start_time = int((datetime.now() - timedelta(days=days)).timestamp())
results = conn.execute("""
SELECT
date(timestamp, 'unixepoch') as date,
COUNT(*) as events,
COUNT(DISTINCT user_id) as unique_users
FROM events
WHERE timestamp BETWEEN ? AND ?
GROUP BY date(timestamp, 'unixepoch')
ORDER BY date
""", (start_time, end_time)).fetchall()
conn.close()
return results
# Usage
analytics = Analytics('analytics.db')
# Track events
analytics.track('page_view', user_id='user_123', session_id='sess_abc', properties={'page': '/home'})
analytics.track('button_click', user_id='user_123', properties={'button': 'signup'})
analytics.track('purchase', user_id='user_456', properties={'amount': 99.99})
# Get analytics
print(analytics.get_top_events())
print(analytics.get_daily_summary(7))
Desktop Application Data
Configuration Storage
import sqlite3
import json
class ConfigStore:
"""Application configuration with SQLite."""
def __init__(self, db_path):
self.db_path = db_path
self._init_db()
def _init_db(self):
"""Initialize config tables."""
conn = sqlite3.connect(self.db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS config (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
type TEXT DEFAULT 'string',
updated_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS presets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
config TEXT NOT NULL,
is_default INTEGER DEFAULT 0
)
""")
conn.commit()
conn.close()
def get(self, key, default=None):
"""Get config value."""
conn = sqlite3.connect(self.db_path)
row = conn.execute(
"SELECT value, type FROM config WHERE key = ?",
(key,)
).fetchone()
conn.close()
if row:
value, type_ = row
if type_ == 'json':
return json.loads(value)
elif type_ == 'int':
return int(value)
elif type_ == 'float':
return float(value)
elif type_ == 'bool':
return value == 'true'
return value
return default
def set(self, key, value, type_='string'):
"""Set config value."""
if type_ == 'json':
value = json.dumps(value)
elif type_ == 'bool':
value = 'true' if value else 'false'
else:
value = str(value)
conn = sqlite3.connect(self.db_path)
conn.execute(
"""
INSERT OR REPLACE INTO config (key, value, type, updated_at)
VALUES (?, ?, ?, CURRENT_TIMESTAMP)
""",
(key, value, type_)
)
conn.commit()
conn.close()
def get_all(self):
"""Get all config as dict."""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
rows = conn.execute("SELECT key, value, type FROM config").fetchall()
conn.close()
result = {}
for row in rows:
result[row['key']] = self.get(row['key'])
return result
# Usage
config = ConfigStore('app_config.db')
# Set preferences
config.set('theme', 'dark', 'string')
config.set('font_size', 14, 'int')
config.set('notifications', True, 'bool')
config.set('window_bounds', {'x': 100, 'y': 100, 'width': 800, 'height': 600}, 'json')
# Get preferences
theme = config.get('theme', 'light')
font_size = config.get('font_size', 12)
Best Practices Summary
When to Use SQLite
- Mobile applications (iOS, Android)
- IoT/embedded systems
- Single-user desktop applications
- Caching layers
- Development and testing environments
- Small-to-medium analytical workloads
- Edge computing scenarios
When NOT to Use SQLite
- High-concurrency write workloads (use PostgreSQL/MySQL)
- Distributed systems requiring replication
- Very large datasets (>100GB with heavy writes)
- Complex stored procedures (use PostgreSQL)
- Real-time analytics at scale (use ClickHouse/StarRocks)
Configuration for Production
-- Recommended production settings
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA cache_size = -128000;
PRAGMA foreign_keys = ON;
PRAGMA busy_timeout = 5000;
PRAGMA temp_store = MEMORY;
-- For analytical workloads:
PRAGMA cache_size = -512000;
PRAGMA mmap_size = 1073741824;
Resources
Conclusion
SQLite’s versatility makes it an excellent choice for a wide range of applications beyond traditional database use. From mobile apps to IoT devices, from caching to analytics, SQLite provides reliable, zero-configuration data storage that scales well for many use cases.
The key is understanding when SQLite is the right tool and applying appropriate patterns and configurations for your specific needs.
Comments