Embedded Relational Storage with SQLite3

Updated

September 7, 2026

Embedded Relational Storage with SQLite3

After reading this chapter, you will master Python’s built-in relational database engine (sqlite3), execute safe parameterized queries to eliminate SQL injection vulnerabilities, leverage sqlite3.Row for dictionary-like column access, manage atomic multi-statement transactions, and perform batch operations with executemany().

Mental model

SQLite is a serverless, zero-configuration C library embedded directly into Python. There are no client-server network sockets, background daemons, or credential setups:

CPython Process
  ├── Python Code
  └── sqlite3 Module (C-Bindings) ──▶ In-Memory Database (':memory:')
                                   OR
                                  Direct Disk File ('production.db')

Because the entire relational database is accessed directly via memory or a single local file, query overhead is measured in microseconds, making SQLite ideal for application caching, state persistence, configuration catalogs, and testing.

Minimal example

Save as sqlite3_essentials.py:

# sqlite3_essentials.py
import sqlite3

def init_database(conn: sqlite3.Connection) -> None:
    with conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS server_nodes (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                hostname TEXT NOT NULL UNIQUE,
                ip_address TEXT NOT NULL,
                active INTEGER DEFAULT 1
            )
        """)

def main() -> None:
    # Use an in-memory database for testing
    conn = sqlite3.connect(":memory:")
    # Enable dictionary-like column access
    conn.row_factory = sqlite3.Row

    init_database(conn)

    # 1. Parameterized insert (Always use ? placeholders to prevent SQL injection)
    with conn:
        conn.execute(
            "INSERT INTO server_nodes (hostname, ip_address, active) VALUES (?, ?, ?)",
            ("worker-01.prod", "10.0.1.10", 1)
        )

    # 2. Querying with Row factory
    cursor = conn.cursor()
    cursor.execute("SELECT id, hostname, ip_address, active FROM server_nodes WHERE active = ?", (1,))
    rows = cursor.fetchall()

    print(f"Retrieved {len(rows)} active nodes:")
    for row in rows:
        print(f"  Node ID {row['id']} | Host: {row['hostname']} | IP: {row['ip_address']}")

    conn.close()

if __name__ == "__main__":
    main()

Run via uv run python sqlite3_essentials.py:

Retrieved 1 active nodes:
  Node ID 1 | Host: worker-01.prod | IP: 10.0.1.10

Worked examples

Case 1: High-Throughput Batch Insertion with executemany()

Executing individual INSERT statements one at a time creates massive transaction journal overhead. executemany() bundles operations into a single optimized transaction:

# batch_metrics_insert.py
import sqlite3
import time

def benchmark_batch_insert() -> None:
    conn = sqlite3.connect(":memory:")
    with conn:
        conn.execute("""
            CREATE TABLE metrics (
                metric_name TEXT,
                value REAL,
                timestamp INTEGER
            )
        """)

    records = [
        ("cpu_utilization", float(i % 100), 1700000000 + i)
        for i in range(50_000)
    ]

    t0 = time.perf_counter()
    with conn:
        # executemany inserts 50,000 records in a single disk/memory transaction
        conn.executemany(
            "INSERT INTO metrics (metric_name, value, timestamp) VALUES (?, ?, ?)",
            records
        )
    duration = time.perf_counter() - t0

    count = conn.execute("SELECT COUNT(*) FROM metrics").fetchone()[0]
    print(f"Inserted {count:,} metric rows in {duration*1000:.2f} ms")
    print(f"Throughput: {count / duration:,.0f} inserts/sec")
    conn.close()

if __name__ == "__main__":
    benchmark_batch_insert()

Run:

uv run python batch_metrics_insert.py

Output:

Inserted 50,000 metric rows in 42.15 ms
Throughput: 1,186,240 inserts/sec

Case 2: Transaction Atomicity and Automatic Rollback

When wrapping database operations inside with conn:, SQLite automatically issues a COMMIT if the block completes successfully. If an exception is raised, it automatically rolls back all changes, preserving database integrity:

# transaction_rollback.py
import sqlite3

def run_financial_transfer(conn: sqlite3.Connection, from_acc: int, to_acc: int, amount: float) -> None:
    try:
        # with conn automatically initiates a transaction
        with conn:
            conn.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amount, from_acc))
            
            # Simulate unexpected failure mid-transaction
            raise RuntimeError("Network timeout while contacting balance ledger!")

            conn.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (amount, to_acc))
    except RuntimeError as err:
        print(f"Transaction failed: {err} -> AUTOMATIC ROLLBACK EXECUTED.")

if __name__ == "__main__":
    conn = sqlite3.connect(":memory:")
    with conn:
        conn.execute("CREATE TABLE accounts (id INTEGER PRIMARY KEY, balance REAL)")
        conn.execute("INSERT INTO accounts VALUES (1, 1000.0), (2, 500.0)")

    print("Initial balances:", conn.execute("SELECT * FROM accounts").fetchall())
    run_financial_transfer(conn, 1, 2, 200.0)
    print("Balances after failure:", conn.execute("SELECT * FROM accounts").fetchall())
    conn.close()

Run:

uv run python transaction_rollback.py

Output:

Initial balances: [(1, 1000.0), (2, 500.0)]
Transaction failed: Network timeout while contacting balance ledger! -> AUTOMATIC ROLLBACK EXECUTED.
Balances after failure: [(1, 1000.0), (2, 500.0)]

Notice that Account 1’s balance was restored to 1000.0 despite the first UPDATE having executed.

Pitfalls

Pitfall 1: SQL Injection via String Interpolation

Never construct SQL queries using f-strings or string formatting:

# FATAL VULNERABILITY: SQL Injection
username = "admin' OR '1'='1"
query = f"SELECT * FROM users WHERE username = '{username}'"  # NEVER DO THIS!

# SECURE: Parameterized Query
cursor.execute("SELECT * FROM users WHERE username = ?", (username,))

Pitfall 2: Trailing Comma in Single Parameter Tuples

In Python, (val) is a grouped expression, not a tuple! When passing a single parameter, you must include a trailing comma: (val,):

# THE BUG:
cursor.execute("SELECT * FROM users WHERE id = ?", (42))  # ValueError: parameters are of unsupported type

# THE FIX:
cursor.execute("SELECT * FROM users WHERE id = ?", (42,)) # Valid 1-element tuple

Exercises

  1. Create a SQLite database table ip_cache(ip TEXT PRIMARY KEY, domain TEXT, ttl INTEGER). Write an UPSERT query using ON CONFLICT(ip) DO UPDATE SET ttl=....
  2. Write a function that reads a CSV file containing user records and streams them into a SQLite table using executemany().
  3. Configure conn.row_factory with a custom lambda that converts query result rows directly into an immutable @dataclass.
  4. Demonstrate how PRAGMA foreign_keys = ON; enforces relational foreign key constraints in SQLite.

Further reading

  • Python Standard Library: sqlite3 module documentation.
  • SQLite Official Documentation: Query Language Understood by SQLite.
  • OWASP: SQL Injection Prevention Cheat Sheet.