Python sqlite3 Module: Embedded Database Basics

import sqlite3
 
# Connect (creates file if it doesn't exist)
conn = sqlite3.connect('mydb.db')
cursor = conn.cursor()
 
# In-memory database (gone when connection closes)
conn = sqlite3.connect(':memory:')

with sqlite3.connect('mydb.db') as conn:
    cursor = conn.cursor()
    # Work with database
# Auto-commits on success, rolls back on exception

cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')
conn.commit()

# BAD: SQL injection risk
cursor.execute(f"SELECT * FROM users WHERE name = '{name}'")
 
# GOOD: parameterized (? placeholder)
cursor.execute("SELECT * FROM users WHERE name = ?", (name,))
 
# GOOD: named parameters
cursor.execute(
    "SELECT * FROM users WHERE name = :name AND email = :email",
    {"name": name, "email": email}
)

# Single row
cursor.execute(
    "INSERT INTO users (name, email) VALUES (?, ?)",
    ("Alice", "alice@example.com")
)
 
# Multiple rows
users = [
    ("Bob", "bob@example.com"),
    ("Charlie", "charlie@example.com"),
]
cursor.executemany(
    "INSERT INTO users (name, email) VALUES (?, ?)",
    users
)
 
# Get inserted ID
cursor.execute("INSERT INTO users (name) VALUES (?)", ("Dave",))
print(cursor.lastrowid)
 
conn.commit()

# Fetch all
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
 
# Fetch one
cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
row = cursor.fetchone()
 
# Iterate directly
for row in cursor.execute("SELECT name, email FROM users"):
    print(row[0], row[1])

cursor.execute(
    "UPDATE users SET email = ? WHERE id = ?",
    ("new@example.com", 1)
)
print(f"Updated {cursor.rowcount} rows")
conn.commit()

cursor.execute("DELETE FROM users WHERE id = ?", (1,))
conn.commit()

conn.row_factory = sqlite3.Row
 
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
row = cursor.fetchone()
 
# Access by name
print(row["name"])
print(row["email"])
 
# Still works as tuple
print(row[0], row[1])

conn = sqlite3.connect('mydb.db')
 
try:
    cursor = conn.cursor()
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    conn.commit()
except Exception:
    conn.rollback()
    raise

with sqlite3.connect('mydb.db') as conn:
    cursor = conn.cursor()
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
# Auto-commits if no exception

cursor.execute("""
    SELECT name FROM sqlite_master 
    WHERE type='table' AND name=?
""", ('users',))
exists = cursor.fetchone() is not None

cursor.execute("""
    INSERT INTO users (id, name, email) VALUES (?, ?, ?)
    ON CONFLICT(id) DO UPDATE SET 
        name = excluded.name,
        email = excluded.email
""", (1, "Alice", "alice@example.com"))

users = [(f"user_{i}", f"user{i}@example.com") for i in range(10000)]
 
conn.execute("BEGIN")
cursor.executemany(
    "INSERT INTO users (name, email) VALUES (?, ?)",
    users
)
conn.commit()

import json
from datetime import datetime
 
# Register adapter for datetime
sqlite3.register_adapter(datetime, lambda dt: dt.isoformat())
sqlite3.register_converter("TIMESTAMP", lambda b: datetime.fromisoformat(b.decode()))
 
# Register adapter for dict/list as JSON
sqlite3.register_adapter(dict, lambda d: json.dumps(d))
sqlite3.register_adapter(list, lambda l: json.dumps(l))
sqlite3.register_converter("JSON", lambda b: json.loads(b.decode()))
 
# Enable converters
conn = sqlite3.connect('mydb.db', detect_types=sqlite3.PARSE_DECLTYPES)

# Use WAL mode for concurrent reads
conn.execute("PRAGMA journal_mode=WAL")
 
# Sync less often (faster, slightly less safe)
conn.execute("PRAGMA synchronous=NORMAL")
 
# Use transactions for bulk operations
conn.execute("BEGIN")
# ... many inserts ...
conn.commit()