Database Tutorial #17: SQLite — When and How to Use It

import sqlite3

conn = sqlite3.connect("myapp.db")
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")  # faster, still safe
conn.execute("PRAGMA cache_size=2500")     # ~10MB page cache (2500 × 4KB pages)
conn.execute("PRAGMA foreign_keys=ON")    # enforce foreign keys

import sqlite3
from contextlib import contextmanager

DB_PATH = "myapp.db"

def get_connection():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # access columns by name
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=ON")
    return conn

@contextmanager
def get_db():
    conn = get_connection()
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# Create tables
def init_db():
    with get_db() as db:
        db.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id        INTEGER PRIMARY KEY AUTOINCREMENT,
                email     TEXT NOT NULL UNIQUE,
                name      TEXT NOT NULL,
                created_at TEXT NOT NULL DEFAULT (datetime('now'))
            )
        """)
        db.execute("""
            CREATE TABLE IF NOT EXISTS posts (
                id         INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id    INTEGER NOT NULL REFERENCES users(id),
                title      TEXT NOT NULL,
                body       TEXT NOT NULL,
                created_at TEXT NOT NULL DEFAULT (datetime('now'))
            )
        """)

# CRUD
def create_user(email: str, name: str) -> int:
    with get_db() as db:
        cursor = db.execute(
            "INSERT INTO users (email, name) VALUES (?, ?)",
            (email, name)
        )
        return cursor.lastrowid

def get_user(user_id: int) -> dict | None:
    with get_db() as db:
        row = db.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
        return dict(row) if row else None

def get_users_with_posts():
    with get_db() as db:
        rows = db.execute("""
            SELECT u.id, u.name, COUNT(p.id) as post_count
            FROM users u
            LEFT JOIN posts p ON p.user_id = u.id
            GROUP BY u.id
            ORDER BY post_count DESC
        """).fetchall()
        return [dict(r) for r in rows]

npm install better-sqlite3
npm install -D @types/better-sqlite3

import Database from "better-sqlite3";

const db = new Database("myapp.db");

// Configure for production use
db.pragma("journal_mode = WAL");
db.pragma("foreign_keys = ON");
db.pragma("synchronous = NORMAL");

// Create tables
db.exec(`
  CREATE TABLE IF NOT EXISTS users (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    email      TEXT NOT NULL UNIQUE,
    name       TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
  );
  CREATE TABLE IF NOT EXISTS posts (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id    INTEGER NOT NULL REFERENCES users(id),
    title      TEXT NOT NULL,
    body       TEXT NOT NULL
  );
`);

// Prepared statements (safe from SQL injection, faster for repeated queries)
const insertUser = db.prepare("INSERT INTO users (email, name) VALUES (?, ?) RETURNING *");
const getUserById = db.prepare("SELECT * FROM users WHERE id = ?");
const listUsers = db.prepare("SELECT * FROM users ORDER BY id DESC LIMIT ?");

// Use
const user = insertUser.get("alex@example.com", "Alex Johnson");
const found = getUserById.get(1);
const users = listUsers.all(10);

// Transaction
const transferPost = db.transaction((postId: number, newUserId: number) => {
  db.prepare("UPDATE posts SET user_id = ? WHERE id = ?").run(newUserId, postId);
});

transferPost(1, 2);

# litestream.yml
dbs:
  - path: /app/myapp.db
    replicas:
      - url: s3://my-bucket/myapp.db

npm install @libsql/client

import { createClient } from "@libsql/client";

const db = createClient({
  url: "libsql://my-db.turso.io",
  authToken: process.env.TURSO_AUTH_TOKEN,
});

const { rows } = await db.execute("SELECT * FROM users WHERE id = ?", [1]);