⚡ Advanced Insert Optimization — Bulk Operations & Transactions

Master performance-critical insert patterns for production systems.

🎯 Bulk Insert Performance

Single inserts are slow. Batching is orders of magnitude faster.

import sqlite3
import time

conn = sqlite3.connect("app.db")
cursor = conn.cursor()

data = [(f"User{i}", i) for i in range(10000)]

# Method 1: Individual inserts (SLOW)
start = time.time()
for name, age in data:
    cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", (name, age))
conn.commit()
print(f"Individual: {time.time() - start:.2f}s")  # ~5 seconds

# Method 2: executemany (FAST)
cursor.execute("DELETE FROM users")  # Clear for comparison
start = time.time()
cursor.executemany("INSERT INTO users (name, age) VALUES (?, ?)", data)
conn.commit()
print(f"executemany: {time.time() - start:.2f}s")  # ~0.05 seconds = 100x faster!

# Method 3: Transactions (FASTEST for individual inserts)
cursor.execute("DELETE FROM users")
start = time.time()
cursor.execute("BEGIN TRANSACTION")
for name, age in data:
    cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", (name, age))
cursor.execute("COMMIT")
print(f"With transaction: {time.time() - start:.2f}s")  # ~0.5 seconds

🔄 Transaction Control for Reliability

Transactions ensure all-or-nothing operations:

# Transfer money between accounts
cursor.execute("PRAGMA foreign_keys = ON")

try:
    cursor.execute("BEGIN")

    # Deduct from account A
    cursor.execute("""
        UPDATE accounts SET balance = balance - 100
        WHERE id = 1
    """)

    # Add to account B
    cursor.execute("""
        UPDATE accounts SET balance = balance + 100
        WHERE id = 2
    """)

    cursor.execute("COMMIT")
except Exception as e:
    cursor.execute("ROLLBACK")
    print(f"Transaction failed: {e}")

📊 Batch Operations with PRAGMA Optimization

import sqlite3

conn = sqlite3.connect("app.db")

# Disable unnecessary safety features for bulk inserts
original_sync = conn.execute("PRAGMA synchronous").fetchone()[0]
conn.execute("PRAGMA synchronous = OFF")  # Skip disk sync
conn.execute("PRAGMA cache_size = 50000")  # Large cache
conn.execute("PRAGMA temp_store = MEMORY")  # Use RAM for temp tables
conn.execute("PRAGMA query_only = OFF")

try:
    cursor = conn.cursor()
    data = [(f"User{i}", i) for i in range(100000)]

    cursor.execute("BEGIN")
    cursor.executemany("INSERT INTO users VALUES (NULL, ?, ?)", data)
    cursor.execute("COMMIT")
finally:
    # Restore original settings
    conn.execute(f"PRAGMA synchronous = {original_sync}")

🔗 Multi-Table Inserts with Foreign Keys

# Insert with cascading relationships
def insert_order_with_items(cursor, customer_id, items):
    """Insert order and items atomically"""
    cursor.execute("BEGIN")

    try:
        # Insert order
        cursor.execute("""
            INSERT INTO orders (customer_id, order_date)
            VALUES (?, CURRENT_TIMESTAMP)
        """, (customer_id,))

        order_id = cursor.lastrowid

        # Insert items (all succeed or all fail)
        for product_id, quantity in items:
            cursor.execute("""
                INSERT INTO order_items (order_id, product_id, quantity)
                VALUES (?, ?, ?)
            """, (order_id, product_id, quantity))

        cursor.execute("COMMIT")
        return order_id

    except Exception as e:
        cursor.execute("ROLLBACK")
        raise Exception(f"Order insert failed: {e}")

# Usage
cursor = conn.cursor()
items = [(1, 2), (3, 1), (5, 4)]  # product_id, quantity
order_id = insert_order_with_items(cursor, customer_id=10, items=items)

⚠️ Conflict Resolution Strategies

# INSERT OR REPLACE - Update if exists
cursor.execute("""
    INSERT OR REPLACE INTO users (id, name, email)
    VALUES (1, 'Alice', 'alice@example.com')
""")

# INSERT OR IGNORE - Skip if constraint violation
cursor.execute("""
    INSERT OR IGNORE INTO users (email, name)
    VALUES ('bob@example.com', 'Bob')
""")

# INSERT OR FAIL - Raise error (default)
cursor.execute("""
    INSERT OR FAIL INTO users (email, name)
    VALUES ('carol@example.com', 'Carol')
""")

# Custom conflict handling with ON CONFLICT
cursor.execute("""
    CREATE TABLE users (
        id INTEGER PRIMARY KEY,
        email TEXT UNIQUE,
        name TEXT NOT NULL,
        version INTEGER
    )
""")

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

📈 Streaming Large Datasets

import sqlite3
from contextlib import closing

def batch_insert_from_csv(db_path, csv_file, batch_size=1000):
    """Insert data from CSV in chunks"""
    import csv

    with closing(sqlite3.connect(db_path)) as conn:
        cursor = conn.cursor()
        cursor.execute("PRAGMA synchronous = OFF")

        batch = []
        with open(csv_file, 'r') as f:
            reader = csv.DictReader(f)
            for row in reader:
                batch.append((row['name'], row['age'], row['email']))

                if len(batch) >= batch_size:
                    cursor.execute("BEGIN")
                    cursor.executemany(
                        "INSERT INTO users VALUES (NULL, ?, ?, ?)", batch
                    )
                    cursor.execute("COMMIT")
                    batch = []

        # Insert remaining
        if batch:
            cursor.execute("BEGIN")
            cursor.executemany(
                "INSERT INTO users VALUES (NULL, ?, ?, ?)", batch
            )
            cursor.execute("COMMIT")

🔍 Savepoints for Partial Rollback

cursor.execute("BEGIN")

# Insert first batch
cursor.executemany("INSERT INTO users VALUES (NULL, ?, ?)", batch1)
cursor.execute("SAVEPOINT after_batch1")

# Insert second batch
try:
    cursor.executemany("INSERT INTO users VALUES (NULL, ?, ?)", batch2)
    cursor.execute("COMMIT")
except Exception:
    # Rollback only batch2, keep batch1
    cursor.execute("ROLLBACK TO after_batch1")
    cursor.execute("COMMIT")  # Commit batch1

💡 Upsert Patterns (Update or Insert)

# Update existing, insert if not exists
def upsert_user(cursor, email, name, age):
    cursor.execute("""
        INSERT INTO users (email, name, age)
        VALUES (?, ?, ?)
        ON CONFLICT(email) DO UPDATE SET
            name = excluded.name,
            age = excluded.age
    """, (email, name, age))

# Usage
cursor.execute("PRAGMA foreign_keys = ON")
upsert_user(cursor, "alice@example.com", "Alice Updated", 31)

🔑 Key Takeaways

• ✅ Use executemany() for batch inserts (100x faster)

• ✅ Wrap related operations in transactions

• ✅ PRAGMA synchronous = OFF for non-critical data

• ✅ ON CONFLICT for upsert patterns

• ✅ Use SAVEPOINT for partial rollback

• ✅ Disable foreign keys temporarily for bulk imports (if safe)

• ✅ Monitor memory with PRAGMA cache_size

• ✅ Batch large CSV imports in chunks

Ready to practice? Challenges | Quiz