Python sqlite3 Guide

import sqlite3
 
# Connect (creates file if doesn't exist)
conn = sqlite3.connect("database.db")
 
# In-memory database
conn = sqlite3.connect(":memory:")
 
# Get cursor
cursor = conn.cursor()
 
# Execute SQL
cursor.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
 
# Commit changes
conn.commit()
 
# Close connection
conn.close()

import sqlite3
 
with sqlite3.connect("database.db") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    rows = cursor.fetchall()
# 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
    )
""")
 
cursor.execute("""
    CREATE TABLE IF NOT EXISTS posts (
        id INTEGER PRIMARY KEY,
        user_id INTEGER,
        title TEXT,
        FOREIGN KEY (user_id) REFERENCES users(id)
    )
""")

# Single row (parameterized - prevents SQL injection!)
cursor.execute(
    "INSERT INTO users (name, email) VALUES (?, ?)",
    ("Alice", "alice@example.com")
)
 
# Get inserted row ID
user_id = cursor.lastrowid
 
# Multiple rows
users = [
    ("Bob", "bob@example.com"),
    ("Carol", "carol@example.com"),
]
cursor.executemany(
    "INSERT INTO users (name, email) VALUES (?, ?)",
    users
)
 
conn.commit()

# Fetch all
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
for row in rows:
    print(row)  # (1, 'Alice', 'alice@example.com', ...)
 
# Fetch one
cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
user = cursor.fetchone()
 
# Iterate directly
for row in cursor.execute("SELECT * FROM users"):
    print(row)
 
# With parameters
cursor.execute(
    "SELECT * FROM users WHERE name LIKE ?",
    ("%Ali%",)
)

cursor.execute(
    "SELECT * FROM users WHERE name = :name AND email = :email",
    {"name": "Alice", "email": "alice@example.com"}
)
 
# Or use dict directly
params = {"name": "Alice"}
cursor.execute("SELECT * FROM users WHERE name = :name", params)

def dict_factory(cursor, row):
    return {col[0]: row[idx] for idx, col in enumerate(cursor.description)}
 
conn.row_factory = dict_factory
cursor = conn.cursor()
 
cursor.execute("SELECT * FROM users")
user = cursor.fetchone()
print(user["name"])  # Access by column name
 
# Built-in Row class
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
user = cursor.fetchone()
print(user["name"])   # By name
print(user[0])        # By index

# Auto-commit mode
conn = sqlite3.connect("database.db", isolation_level=None)
 
# Manual transactions
conn = sqlite3.connect("database.db")
try:
    cursor.execute("INSERT INTO users (name) VALUES (?)", ("Test",))
    cursor.execute("UPDATE accounts SET balance = balance - 100")
    conn.commit()
except:
    conn.rollback()
    raise

cursor.execute(
    "UPDATE users SET email = ? WHERE id = ?",
    ("newemail@example.com", 1)
)
print(cursor.rowcount)  # Number of affected rows
conn.commit()

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

import sqlite3
from contextlib import contextmanager
 
class Database:
    def __init__(self, path):
        self.path = path
    
    @contextmanager
    def connection(self):
        conn = sqlite3.connect(self.path)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
        finally:
            conn.close()
    
    def execute(self, sql, params=()):
        with self.connection() as conn:
            cursor = conn.cursor()
            cursor.execute(sql, params)
            conn.commit()
            return cursor
    
    def query(self, sql, params=()):
        with self.connection() as conn:
            cursor = conn.cursor()
            cursor.execute(sql, params)
            return cursor.fetchall()
 
db = Database("app.db")
users = db.query("SELECT * FROM users WHERE active = ?", (True,))

def migrate(conn):
    cursor = conn.cursor()
    
    # Check current version
    cursor.execute("PRAGMA user_version")
    version = cursor.fetchone()[0]
    
    if version < 1:
        cursor.execute("CREATE TABLE users (...)")
        cursor.execute("PRAGMA user_version = 1")
    
    if version < 2:
        cursor.execute("ALTER TABLE users ADD COLUMN active INTEGER DEFAULT 1")
        cursor.execute("PRAGMA user_version = 2")
    
    conn.commit()

def bulk_insert(conn, table, rows):
    if not rows:
        return
    
    cols = rows[0].keys()
    placeholders = ", ".join("?" * len(cols))
    sql = f"INSERT INTO {table} ({', '.join(cols)}) VALUES ({placeholders})"
    
    cursor = conn.cursor()
    cursor.executemany(sql, [tuple(r.values()) for r in rows])
    conn.commit()

# Use transactions for bulk operations
cursor.executemany("INSERT ...", many_rows)  # Not individual inserts
 
# Use indexes
cursor.execute("CREATE INDEX idx_users_email ON users(email)")
 
# Use EXPLAIN to analyze queries
cursor.execute("EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = ?", (email,))
 
# WAL mode for concurrent reads
cursor.execute("PRAGMA journal_mode=WAL")

import sqlite3
 
# Connect
conn = sqlite3.connect("db.sqlite")
cursor = conn.cursor()
 
# Create
cursor.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
 
# Insert
cursor.execute("INSERT INTO t (name) VALUES (?)", ("value",))
cursor.executemany("INSERT INTO t (name) VALUES (?)", [("a",), ("b",)])
 
# Query
cursor.execute("SELECT * FROM t WHERE name = ?", ("value",))
row = cursor.fetchone()
rows = cursor.fetchall()
 
# Update/Delete
cursor.execute("UPDATE t SET name = ? WHERE id = ?", ("new", 1))
cursor.execute("DELETE FROM t WHERE id = ?", (1,))
 
# Commit
conn.commit()
 
# Dict results
conn.row_factory = sqlite3.Row