Python SQLite Part 1: connect, execute, fetchall

import sqlite3. conn = sqlite3.connect("file.db"). cursor.execute(sql, params) runs queries. Always use ? placeholders, never f-strings — that's how SQL injection happens. conn.commit() saves changes; conn.close() releases the file.

SQLite is a database in a file. No server, no setup, just import sqlite3 and start querying. Built into Python's standard library.

Connect and create

import sqlite3

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

cursor.execute("""
  CREATE TABLE IF NOT EXISTS contacts (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    email TEXT NOT NULL,
    phone TEXT
  )
""")
conn.commit()

sqlite3.connect("file.db") opens (or creates) a database file. cursor is what you use to execute SQL. conn.commit() flushes changes to disk.

CREATE TABLE IF NOT EXISTS is idempotent — safe to run on every startup.

Insert with parameters

cursor.execute(
  "INSERT INTO contacts (name, email, phone) VALUES (?, ?, ?)",
  ("Alice Smith", "alice@acme.com", "555-0101")
)
conn.commit()
print(cursor.lastrowid)    # auto-generated ID

The ? placeholders are filled by the second argument (a tuple). Never use f-strings or %-formatting for user input:

# BAD — SQL injection
cursor.execute(f"INSERT INTO contacts VALUES ('{name}', '{email}')")
# Imagine name = "'; DROP TABLE contacts; --"

# GOOD — parameterized
cursor.execute("INSERT INTO contacts (name, email) VALUES (?, ?)", (name, email))

The library escapes values safely. This is non-negotiable.

Bulk insert: executemany

contacts = [
  ("Bob Jones", "bob@globex.com", "555-0102"),
  ("Charlie Brown", "charlie@initech.com", "555-0103"),
  ("Diana Prince", "diana@hooli.com", None),
]

cursor.executemany(
  "INSERT INTO contacts (name, email, phone) VALUES (?, ?, ?)",
  contacts
)
conn.commit()
print(cursor.rowcount)    # number inserted

executemany runs the same query for each tuple in the list. Much faster than a loop of individual execute calls — fewer round trips to the database engine.

Querying: fetchall, fetchone

cursor.execute("SELECT * FROM contacts")
rows = cursor.fetchall()
for row in rows:
  print(row)
# (1, 'Alice Smith', 'alice@acme.com', '555-0101')
# (2, 'Bob Jones', 'bob@globex.com', '555-0102')

fetchall() reads all results into a list of tuples. fetchone() returns the next single row.

For huge results, iterate the cursor directly:

cursor.execute("SELECT * FROM contacts")
for row in cursor:
  print(row)

This streams — doesn't load all rows into memory.

Row factory: dict-like access

conn.row_factory = sqlite3.Row
cursor = conn.cursor()

cursor.execute("SELECT * FROM contacts")
for row in cursor:
  print(row["name"], row["email"])

By default, rows are tuples — row[0], row[1]. Set conn.row_factory = sqlite3.Row to get name-indexable rows. Much more readable.

You can also write a custom factory:

def dict_factory(cursor, row):
  return {col[0]: row[i] for i, col in enumerate(cursor.description)}

conn.row_factory = dict_factory

Now rows come back as plain dicts.

WHERE, ORDER BY, LIMIT

cursor.execute("SELECT name, price FROM products WHERE price > ?", (50,))
expensive = cursor.fetchall()

cursor.execute("SELECT name, price FROM products ORDER BY price DESC LIMIT 3")
top_three = cursor.fetchall()

The , after 50 is required — (50,) is a 1-tuple, (50) is just 50 in parens.

UPDATE and DELETE

cursor.execute(
  "UPDATE products SET price = ? WHERE name = ?",
  (69.99, "Headphones")
)
conn.commit()
print(cursor.rowcount)    # how many rows were affected

cursor.execute("DELETE FROM products WHERE in_stock = 0")
conn.commit()
print(cursor.rowcount)

UPDATE and DELETE return rowcount via cursor.rowcount. Useful for confirming the intended rows were touched.

Aggregates: COUNT, SUM, AVG, MIN, MAX

cursor.execute("""
  SELECT COUNT(*) as total,
         AVG(price) as avg_price,
         MIN(price) as cheapest,
         MAX(price) as priciest
  FROM products
""")
stats = cursor.fetchone()
print(stats["total"], stats["avg_price"])

Aliases (AS total) become column names accessible via sqlite3.Row.

GROUP BY

cursor.execute("""
  SELECT category, COUNT(*) as count, AVG(price) as avg_price
  FROM products
  GROUP BY category
""")
for row in cursor:
  print(f"{row['category']}: {row['count']} items, avg ${row['avg_price']:.2f}")

Standard SQL — GROUP BY aggregates rows by the group column.

Connection as a context manager

with sqlite3.connect("contacts.db") as conn:
  conn.row_factory = sqlite3.Row
  cursor = conn.cursor()
  cursor.execute("SELECT * FROM contacts")
  rows = cursor.fetchall()
# Auto-commits on success, rollback on exception. Doesn't auto-close.

The with context auto-commits if the block succeeded, rolls back if it raised. Note: it does NOT close the connection. Wrap in try/finally or contextlib.closing(...) for that:

import contextlib

with contextlib.closing(sqlite3.connect("file.db")) as conn:
  with conn:        # auto-commit/rollback
    cursor = conn.cursor()
    ...
# closed and committed

Datatypes

SQLite has dynamic typing — column types are advisory:

Python types map automatically: None ↔ NULL, bool → INTEGER 0/1, etc.

For dates and datetimes, store as ISO strings or Unix timestamps. SQLite has functions like date() and strftime() to work with them.

Memory database

conn = sqlite3.connect(":memory:")

:memory: is an in-memory database — fast, ephemeral. Useful for tests or temporary processing.

Schema migrations

For real projects, you'll need to evolve schema over time. Options:

• Hand-rolled CREATE TABLE IF NOT EXISTS and ALTER TABLE with version tracking.
• Use a migration tool: Alembic (works with SQLite via SQLAlchemy), yoyo-migrations, golang-migrate.

For small projects, hand-rolled is fine. For bigger projects, the tools save grief.

A complete CRUD example

import sqlite3
import contextlib

DB = "products.db"

def init():
  with contextlib.closing(sqlite3.connect(DB)) as conn:
    conn.execute("""
      CREATE TABLE IF NOT EXISTS products (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        price REAL NOT NULL
      )
    """)
    conn.commit()

def add_product(name, price):
  with contextlib.closing(sqlite3.connect(DB)) as conn:
    cur = conn.execute(
      "INSERT INTO products (name, price) VALUES (?, ?)",
      (name, price)
    )
    conn.commit()
    return cur.lastrowid

def get_product(product_id):
  with contextlib.closing(sqlite3.connect(DB)) as conn:
    conn.row_factory = sqlite3.Row
    row = conn.execute(
      "SELECT * FROM products WHERE id = ?", (product_id,)
    ).fetchone()
    return dict(row) if row else None

def update_price(product_id, new_price):
  with contextlib.closing(sqlite3.connect(DB)) as conn:
    cur = conn.execute(
      "UPDATE products SET price = ? WHERE id = ?",
      (new_price, product_id)
    )
    conn.commit()
    return cur.rowcount

def delete_product(product_id):
  with contextlib.closing(sqlite3.connect(DB)) as conn:
    cur = conn.execute(
      "DELETE FROM products WHERE id = ?", (product_id,)
    )
    conn.commit()
    return cur.rowcount

init()
pid = add_product("Widget", 9.99)
print(get_product(pid))
update_price(pid, 12.99)
delete_product(pid)

Each function opens, does, commits, closes. Fine for a script. For a long-running app, share one connection (or a connection pool).

Common stumbles

SQL injection via f-strings. f"WHERE name = '{name}'" — disaster. Always ? placeholders.

Forgetting conn.commit(). Writes are buffered until commit; closing without commit loses them. (Unless you set isolation_level=None for autocommit.)

Forgetting the trailing comma. cursor.execute(sql, ("Alice")) — passes a string, not a 1-tuple. Use ("Alice",).

Reusing a closed cursor. After conn.close(), the cursor is invalid.

Threads sharing connections. SQLite connections aren't thread-safe by default. Each thread should have its own connection, or set check_same_thread=False carefully.

Big WAL files. SQLite's write-ahead log can grow large. For long-running apps, periodic PRAGMA wal_checkpoint; keeps it in check.

Not handling unique constraint failures. Inserting a duplicate raises sqlite3.IntegrityError. Catch and handle.

What's next

Lesson 35: SQLite Part 2. Joins, transactions, indexes, performance, the bigger picture.

Recap

sqlite3.connect("file.db") + cursor.execute(sql, params) runs SQL. Always parameterize with ? — no f-strings. fetchall returns all rows; fetchone returns one. Set conn.row_factory = sqlite3.Row for dict-like rows. executemany for bulk inserts. conn.commit() to save; conn.close() to release. Use with conn: for auto-commit on success.

Next lesson: SQLite Part 2.