SQLite Operations: Backup, Performance Tuning, and Production Deployment

# Create database with specific settings
sqlite3 app.db "PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA cache_size=-64000;"

# Check current settings
sqlite3 app.db "PRAGMA journal_mode;"
sqlite3 app.db "PRAGMA synchronous;"
sqlite3 app.db "PRAGMA cache_size;"
sqlite3 app.db "PRAGMA page_size;"

-- Journal Mode (DELETE, TRUNCATE, PERSIST, WAL)
PRAGMA journal_mode = WAL;           -- Write-Ahead Logging (recommended)

-- Synchronous (OFF, NORMAL, FULL, EXTRA)
PRAGMA synchronous = NORMAL;         -- Balance between safety and speed

-- Cache Size (negative = KB, positive = pages)
PRAGMA cache_size = -64000;          -- 64MB cache

-- Page Size (512, 1024, 2048, 4096, 8192, 16384, 32768)
PRAGMA page_size = 4096;             -- Default for most systems

-- Temp Store (DEFAULT, FILE, MEMORY)
PRAGMA temp_store = MEMORY;          -- Store temp data in memory

-- MMAP Size (bytes, 0 to disable)
PRAGMA mmap_size = 268435456;        -- 256MB memory-mapped I/O

-- Locking Mode (NORMAL, EXCLUSIVE)
PRAGMA locking_mode = NORMAL;

-- Busy Timeout (milliseconds)
PRAGMA busy_timeout = 5000;

-- Create a settings table for persistent configuration
CREATE TABLE IF NOT EXISTS _settings (
    key TEXT PRIMARY KEY,
    value TEXT
);

-- Store configuration
INSERT OR REPLACE INTO _settings (key, value) VALUES ('journal_mode', 'WAL');
INSERT OR REPLACE INTO _settings (key, value) VALUES ('cache_size', '-64000');

-- Application reads settings on startup
-- Then applies: PRAGMA journal_mode = (SELECT value FROM _settings WHERE key = 'journal_mode');

import sqlite3
import shutil
import os

def backup_database(source_db, backup_db):
    """Create hot backup of SQLite database."""
    source_conn = sqlite3.connect(source_db)
    backup_conn = sqlite3.connect(backup_db)
    
    source_conn.backup(backup_conn)
    
    backup_conn.close()
    source_conn.close()
    
    print(f"Backup created: {backup_db}")

# Usage
backup_database('production.db', f'backup_{datetime.now().strftime("%Y%m%d_%H%M%S")}.db')

# For WAL databases, also copy the WAL and SHM files
def backup_database_with_wal(db_path, backup_dir):
    """Backup database with WAL files."""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    
    # Checkpoint to consolidate WAL
    conn = sqlite3.connect(db_path)
    conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
    conn.close()
    
    # Copy main database
    shutil.copy2(db_path, os.path.join(backup_dir, f'db_{timestamp}.db'))
    
    wal_path = f"{db_path}-wal"
    if os.path.exists(wal_path):
        shutil.copy2(wal_path, os.path.join(backup_dir, f'db_{timestamp}.db-wal'))
    
    print(f"Backup completed: {timestamp}")

-- Export to SQL file
.output backup.sql
.dump
.output stdout

-- Or use .backup command in CLI
sqlite3 production.db ".backup production_backup.db"

-- Export specific tables
sqlite3 production.db "SELECT sql FROM sqlite_master WHERE type='table';" > schema.sql
sqlite3 production.db ".mode csv" ".headers on" ".output data.csv" "SELECT * FROM users;"

import sqlite3
from datetime import datetime

def incremental_backup(source_db, backup_db, last_backup_time):
    """Backup only changes since last backup."""
    source_conn = sqlite3.connect(source_db)
    backup_conn = sqlite3.connect(backup_db)
    
    # Get changes since last backup
    source_conn.row_factory = sqlite3.Row
    cursor = source_conn.cursor()
    
    # Insert changed/added rows
    cursor.execute("""
        SELECT * FROM events 
        WHERE timestamp > ?
    """, (last_backup_time,))
    
    for row in cursor.fetchall():
        # Insert into backup (use INSERT OR REPLACE for updates)
        columns = ', '.join(row.keys())
        placeholders = ', '.join(['?'] * len(row))
        backup_conn.execute(
            f"INSERT OR REPLACE INTO events ({columns}) VALUES ({placeholders})",
            tuple(row)
        )
    
    backup_conn.commit()
    source_conn.close()
    backup_conn.close()

#!/bin/bash
# backup-sqlite.sh

BACKUP_DIR="/var/backups/sqlite"
DB_PATH="/var/www/app/production.db"
RETENTION_DAYS=30

# Create backup directory
mkdir -p "$BACKUP_DIR"

# Checkpoint first (for WAL mode)
sqlite3 "$DB_PATH" "PRAGMA wal_checkpoint(TRUNCATE);"

# Create timestamped backup
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
BACKUP_FILE="$BACKUP_DIR/db_$TIMESTAMP.db"

cp "$DB_PATH" "$BACKUP_FILE"

# Also backup WAL if exists
if [ -f "${DB_PATH}-wal" ]; then
    cp "${DB_PATH}-wal" "${BACKUP_FILE}-wal"
fi

# Compress backup
gzip "$BACKUP_FILE"

# Clean old backups
find "$BACKUP_DIR" -name "db_*.db.gz" -mtime +$RETENTION_DAYS -delete

echo "Backup completed: $BACKUP_FILE.gz"

-- Analyze table for query planner
ANALYZE;

-- Check query plan
EXPLAIN QUERY PLAN 
SELECT * FROM orders WHERE customer_id = 123 AND status = 'pending';

-- Create appropriate indexes
CREATE INDEX idx_orders_customer_status ON orders(customer_id, status);
CREATE INDEX idx_orders_date ON orders(created_at);

-- Use covering indexes for frequent queries
CREATE INDEX idx_covering ON table(col1, col2, col3) WHERE col1 = 'value';

-- Partial indexes for specific queries_active_users ON users
CREATE INDEX idx(email) WHERE status = 'active';

# Using sqlite3 with persistent connections
import sqlite3
from contextlib import contextmanager

class SQLitePool:
    def __init__(self, db_path, pool_size=5):
        self.db_path = db_path
        self.connections = []
        
        # Pre-create connections
        for _ in range(pool_size):
            conn = sqlite3.connect(db_path, check_same_thread=False)
            conn.execute("PRAGMA journal_mode=WAL")
            conn.execute("PRAGMA busy_timeout=5000")
            self.connections.append(conn)
    
    @contextmanager
    def get_connection(self):
        conn = self.connections.pop()
        try:
            yield conn
        finally:
            self.connections.append(conn)

# Usage
pool = SQLitePool('app.db')
with pool.get_connection() as conn:
    conn.execute("SELECT * FROM users")

-- Increase page cache
PRAGMA cache_size = -128000;  -- 128MB

-- Enable memory-mapped I/O
PRAGMA mmap_size = 1073741824;  -- 1GB

-- Use memory for temp tables
PRAGMA temp_store = MEMORY;

-- Optimize for read vs write workload
-- For read-heavy:
PRAGMA synchronous = OFF;
PRAGMA journal_mode = MEMORY;

-- For write-heavy:
PRAGMA synchronous = NORMAL;
PRAGMA journal_mode = WAL;

import sqlite3
import time

def benchmark_batch():
    db_path = 'benchmark.db'
    conn = sqlite3.connect(db_path)
    conn.execute("CREATE TABLE IF NOT EXISTS items (id INTEGER PRIMARY KEY, data TEXT)")
    
    # Method 1: Individual inserts
    start = time.time()
    for i in range(10000):
        conn.execute("INSERT INTO items (data) VALUES (?)", (f"item_{i}",))
    conn.commit()
    print(f"Individual inserts: {time.time() - start:.2f}s")
    
    # Method 2: Executemany
    start = time.time()
    data = [(f"item_{i}",) for i in range(10000)]
    conn.executemany("INSERT INTO items (data) VALUES (?)", data)
    conn.commit()
    print(f"Executemany: {time.time() - start:.2f}s")
    
    # Method 3: Transaction with single insert (fastest)
    start = time.time()
    conn.execute("DELETE FROM items")
    conn.execute("BEGIN")
    for i in range(10000):
        conn.execute("INSERT INTO items (data) VALUES (?)", (f"item_{i}",))
    conn.execute("COMMIT")
    print(f"Transaction: {time.time() - start:.2f}s")
    
    conn.close()

benchmark_batch()

-- Enable WAL mode
PRAGMA journal_mode = WAL;

-- Checkpoint modes
PRAGMA wal_checkpoint(PASSIVE);  -- Default, non-blocking
PRAGMA wal_checkpoint(FULL);    -- Blocking, ensures complete checkpoint
PRAGMA wal_checkpoint(RESTART); -- Like FULL but holds shared lock
PRAGMA wal_checkpoint(TRUNCATE); -- Truncates WAL after checkpoint

import sqlite3
import os

def manage_wal(db_path):
    """Manage WAL file size and checkpointing."""
    conn = sqlite3.connect(db_path)
    
    # Get WAL file size
    wal_path = f"{db_path}-wal"
    if os.path.exists(wal_path):
        wal_size = os.path.getsize(wal_path) / (1024 * 1024)
        print(f"WAL size: {wal_size:.2f} MB")
    
    # Trigger checkpoint if WAL is large (> 100MB)
    if wal_size > 100:
        print("Triggering checkpoint...")
        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
    
    # Get checkpoint status
    result = conn.execute("PRAGMA wal_checkpoint").fetchone()
    print(f"Checkpoint result: busy={result[0],} written_pages={result[1]} pages={result[2]}")
    
    conn.close()

# Auto-checkpoint configuration
# In most cases, SQLite auto-checkpoints at 1000 pages by default
# Adjust with:
# PRAGMA wal_autocheckpoint = 1000;

-- Database file info
PRAGMA page_count;          -- Total pages
PRAGMA page_size;           -- Page size in bytes
PRAGMA freelist_count;      -- Free pages
PRAGMA schema_version;      -- Schema version

-- Index info
PRAGMA index_list(table_name);
PRAGMA index_info(index_name);

-- Table info
PRAGMA table_info(table_name);

-- Storage used by each table
SELECT 
    name,
    SUM(pgsize) as size_bytes
FROM dbstat
WHERE name LIKE 'sqlite_%'
GROUP BY name;

-- Table and index sizes
SELECT 
    name,
    SUM(pgsize) as bytes,
    SUM(pgsize) / 1024 as kb,
    SUM(pgsize) / (1024*1024) as mb
FROM dbstat
WHERE name NOT LIKE 'sqlite_%'
GROUP BY name
ORDER BY bytes DESC;

-- Enable extended result codes
PRAGMA extended_result_codes = ON;

-- Last insert rowid
SELECT last_insert_rowid();

-- Rows modified by last statement
SELECT changes();

-- SQLITE_STMTSTATUS_fullscan_step - count of full table scans
PRAGMA compile_options;

-- Query running time (in Python)
import sqlite3
import time

conn = sqlite3.connect('app.db')
start = time.time()
conn.execute("SELECT COUNT(*) FROM large_table")
print(f"Query time: {time.time() - start:.3f}s")

#!/bin/bash
# sqlite-health.sh

DB_PATH=$1

if [ ! -f "$DB_PATH" ]; then
    echo "Database not found: $DB_PATH"
    exit 1
fi

echo "=== SQLite Health Check ==="
echo ""

# Database size
SIZE=$(stat -f%z "$DB_PATH" 2>/dev/null || stat -c%s "$DB_PATH")
echo "Database size: $((SIZE/1024/1024)) MB"

# Page count
PAGES=$(sqlite3 "$DB_PATH" "PRAGMA page_count;")
echo "Total pages: $PAGES"

# Free pages
FREE=$(sqlite3 "$DB_PATH" "PRAGMA freelist_count;")
echo "Free pages: $FREE"

# Journal mode
MODE=$(sqlite3 "$DB_PATH" "PRAGMA journal_mode;")
echo "Journal mode: $MODE"

# WAL size if applicable
if [ -f "${DB_PATH}-wal" ]; then
    WAL_SIZE=$(stat -f%z "${DB_PATH}-wal" 2>/dev/null || stat -c%s "${DB_PATH}-wal")
    echo "WAL size: $((WAL_SIZE/1024/1024)) MB"
fi

# Schema version
VERSION=$(sqlite3 "$DB_PATH" "PRAGMA schema_version;")
echo "Schema version: $VERSION"

echo ""
echo "=== Table Sizes ==="
sqlite3 "$DB_PATH" "
    SELECT name, SUM(pgsize)/1024 as kb 
    FROM dbstat 
    WHERE name NOT LIKE 'sqlite_%' 
    GROUP BY name 
    ORDER BY kb DESC 
    LIMIT 10;
"

class SQLiteReplica:
    """Application-level read replica pattern."""
    
    def __init__(self, primary_path, replica_paths):
        self.primary_path = primary_path
        self.replica_paths = replica_paths
        self.current_replica = 0
    
    def get_read_connection(self):
        """Get connection to a replica."""
        path = self.replica_paths[self.current_replica]
        return sqlite3.connect(path, timeout=5)
    
    def get_write_connection(self):
        """Get connection to primary."""
        return sqlite3.connect(self.primary_path)
    
    def rotate_replica(self):
        """Rotate to next replica."""
        self.current_replica = (self.current_replica + 1) % len(self.replica_paths)
    
    def sync_replica(self, replica_path):
        """Sync replica from primary."""
        primary = sqlite3.connect(self.primary_path)
        replica = sqlite3.connect(replica_path)
        primary.backup(replica)
        replica.close()
        primary.close()

#!/bin/bash
# restore-sqlite.sh

BACKUP_FILE=$1
DB_PATH=$2

if [ -z "$BACKUP_FILE" ] || [ -z "$DB_PATH" ]; then
    echo "Usage: $0 <backup_file> <database_path>"
    exit 1
fi

# For WAL mode, first switch to DELETE
sqlite3 "$DB_PATH" "PRAGMA journal_mode=DELETE;"

# Stop application (ensure no writes during restore)
# ...

# Restore database
cp "$BACKUP_FILE" "$DB_PATH"

# Optionally restore WAL
if [ -f "${BACKUP_FILE}-wal" ]; then
    cp "${BACKUP_FILE}-wal" "${DB_PATH}-wal"
fi

# Switch back to WAL
sqlite3 "$DB_PATH" "PRAGMA journal_mode=WAL;"

echo "Restore completed"

# Secure database file
chmod 600 app.db
chown appuser:appgroup app.db

# For WAL mode, also secure WAL and SHM files
chmod 600 app.db-wal
chmod 600 app.db-shm

# Directory permissions (prevent deletion/movement)
chmod 755 /var/lib/app/

# Using SQLCipher for encryption (requires compilation)
# pip install sqlcipher

from sqlcipher import sqlcipher
sqlcipher.load_libsqlcipher()

conn = sqlcipher.connect('encrypted.db')
conn.execute(f"PRAGMA key = 'your-encryption-key';")
conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, data TEXT)")

# Or using System.Data.SQLite with encryption
# Or using SEE (SQLite Encryption Extension) - proprietary

# Always use parameterized queries
# Bad
cursor.execute(f"SELECT * FROM users WHERE name = '{name}'")

# Good
cursor.execute("SELECT * FROM users WHERE name = ?", (name,))

# For dynamic column names, validate against whitelist
ALLOWED_SORTS = {'name', 'created_at', 'id'}
sort_column = sort if sort in ALLOWED_SORTS else 'id'
cursor.execute(f"SELECT * FROM users ORDER BY {sort_column}")

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install SQLite (usually pre-installed in most images)
RUN apt-get update && apt-get install -y sqlite3 && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# Create data directory
RUN mkdir -p /data && chown -R appuser:appgroup /data

# Volume for persistent data
VOLUME ["/data"]

CMD ["python", "app.py"]

# docker-compose.yml
version: '3.8'

services:
  app:
    build: .
    volumes:
      - sqlite-data:/data
    environment:
      - DB_PATH=/data/app.db

volumes:
  sqlite-data:

# Single-file application deployment
import sys
import shutil
import tempfile

def deploy_embedded(db_path):
    """Deploy SQLite with application."""
    
    # Create database in user's data directory
    import os
    data_dir = os.path.expanduser('~/.myapp')
    os.makedirs(data_dir, exist_ok=True)
    
    db_path = os.path.join(data_dir, 'app.db')
    
    # Initialize database if not exists
    if not os.path.exists(db_path):
        conn = sqlite3.connect(db_path)
        conn.execute("CREATE TABLE ...")
        conn.close()
    
    return db_path

-- Production configuration
PRAGMA journal_mode = WAL;           -- Enable WAL
PRAGMA synchronous = NORMAL;          -- Balance safety/speed
PRAGMA cache_size = -64000;           -- 64MB cache
PRAGMA temp_store = MEMORY;           -- Temp in memory
PRAGMA mmap_size = 268435456;         -- 256MB mmap
PRAGMA busy_timeout = 5000;           -- 5s busy timeout
PRAGMA wal_autocheckpoint = 1000;     -- Checkpoint every 1000 pages
PRAGMA foreign_keys = ON;              -- Enable foreign keys
PRAGMA ignore_check_constraints = OFF;

import sqlite3
import time
import logging

class SQLiteMonitor:
    def __init__(self, db_path):
        self.db_path = db_path
        self.logger = logging.getLogger(__name__)
    
    def get_stats(self):
        conn = sqlite3.connect(self.db_path)
        
        stats = {
            'page_count': conn.execute("PRAGMA page_count").fetchone()[0],
            ' freelist_count': conn.execute("PRAGMA freelist_count").fetchone()[0],
            'schema_version': conn.execute("PRAGMA schema_version").fetchone()[0],
        }
        
        conn.close()
        return stats
    
    def check_health(self):
        try:
            conn = sqlite3.connect(self.db_path, timeout=1)
            conn.execute("SELECT 1")
            conn.close()
            return True
        except Exception as e:
            self.logger.error(f"Health check failed: {e}")
            return False

-- Always use WAL in production
PRAGMA journal_mode = WAL;

-- Check for slow queries
EXPLAIN QUERY PLAN SELECT * FROM large_table WHERE unindexed_column = 'value';

-- Add appropriate indexes before deployment
CREATE INDEX idx_table_column ON table(column);

# Regular checkpoint in background
import threading
import sqlite3

def checkpoint_worker(db_path, interval=3600):
    while True:
        time.sleep(interval)
        try:
            conn = sqlite3.connect(db_path)
            conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
            conn.close()
        except Exception as e:
            print(f"Checkpoint error: {e}")

# Start checkpoint thread
thread = threading.Thread(target=checkpoint_worker, args=('app.db',))
thread.daemon = True
thread.start()