SQLite with Python

Python ships with SQLite support built in. The sqlite3 module is part of the standard library, so there's nothing to install.

Quick start

Three lines to get a working database:

import sqlite3

conn = sqlite3.connect("products.db")
cursor = conn.cursor()
cursor.execute("SELECT sqlite_version()")
print(cursor.fetchone())  # ('3.45.0',)

Connecting to a database

sqlite3.connect() opens an existing database file or creates one if it doesn't exist:

import sqlite3

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

To work with a temporary in-memory database, pass ":memory:" instead of a file path:

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

In-memory databases are fast and isolated. They disappear when the connection closes, which makes them useful for tests.

Creating tables

Use cursor.execute() to run DDL statements, then call conn.commit() to persist the change:

import sqlite3

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

cursor.execute("""
    CREATE TABLE IF NOT EXISTS products (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        price REAL NOT NULL,
        stock INTEGER DEFAULT 0
    )
""")

conn.commit()

CREATE TABLE IF NOT EXISTS prevents an error if the table already exists. Always commit after DDL and DML statements when not using a context manager.

Inserting rows

Insert a single row by passing a tuple of values as the second argument to execute():

cursor.execute(
    "INSERT INTO products (name, price, stock) VALUES (?, ?, ?)",
    ("Wireless Keyboard", 49.99, 120)
)
conn.commit()

To insert multiple rows at once, use executemany() with a list of tuples:

new_products = [
    ("USB-C Hub", 29.99, 85),
    ("Webcam HD", 79.99, 40),
    ("Monitor Stand", 34.99, 200),
]

cursor.executemany(
    "INSERT INTO products (name, price, stock) VALUES (?, ?, ?)",
    new_products
)
conn.commit()

executemany() is more efficient than calling execute() in a loop because it batches the operations in a single transaction.

Parameterized queries

The ? placeholders in the examples above are parameterized queries. Always use them when incorporating user input or external data into a query.

String formatting a query like this is dangerous:

# Never do this
name = input("Product name: ")
cursor.execute(f"SELECT * FROM products WHERE name = '{name}'")

A user could enter '; DROP TABLE products; -- and destroy your data. With parameterized queries, the value is treated as data, not SQL:

# Safe
name = input("Product name: ")
cursor.execute("SELECT * FROM products WHERE name = ?", (name,))

Note the trailing comma inside the tuple. Python requires it to distinguish a single-element tuple from a parenthesized expression.

Querying data

After executing a SELECT, fetch results with fetchone(), fetchall(), or fetchmany():

# Fetch a single row
cursor.execute("SELECT * FROM products WHERE id = ?", (1,))
row = cursor.fetchone()
print(row)  # (1, 'Wireless Keyboard', 49.99, 120)

# Fetch all rows
cursor.execute("SELECT name, price FROM products ORDER BY price")
rows = cursor.fetchall()
for name, price in rows:
    print(f"{name}: ${price:.2f}")

# Fetch in batches
cursor.execute("SELECT * FROM products")
while batch := cursor.fetchmany(50):
    process(batch)

You can also iterate over a cursor directly, which avoids loading all rows into memory at once:

cursor.execute("SELECT * FROM products WHERE stock > 0")
for row in cursor:
    print(row)

Using context managers

The with statement handles commits and rollbacks automatically. If the block completes without error, the connection commits. If an exception is raised, it rolls back:

import sqlite3

with sqlite3.connect("store.db") as conn:
    cursor = conn.cursor()
    cursor.execute(
        "UPDATE products SET stock = stock - 1 WHERE id = ?", (3,)
    )
# Committed automatically here

Note that the context manager does not close the connection. Call conn.close() explicitly when you're done, or use a nested with block for fully automatic cleanup.

Row factories: dict-like access

By default, rows are plain tuples. Assign sqlite3.Row as the row factory to get named column access:

conn.row_factory = sqlite3.Row

cursor = conn.cursor()
cursor.execute("SELECT * FROM products WHERE id = ?", (1,))
row = cursor.fetchone()

print(row["name"])   # Wireless Keyboard
print(row["price"])  # 49.99

sqlite3.Row supports both index access (row[0]) and name access (row["name"]), and it works with dict(row) to convert to a plain dictionary.

Transactions

SQLite wraps statements in transactions automatically, but the exact behavior depends on how you use the connection.

By default, sqlite3 operates in deferred transaction mode. DML statements (INSERT, UPDATE, DELETE) are held until you call conn.commit(). To discard pending changes, call conn.rollback():

try:
    cursor.execute("INSERT INTO orders (product_id, qty) VALUES (?, ?)", (1, 2))
    cursor.execute("UPDATE products SET stock = stock - 2 WHERE id = ?", (1,))
    conn.commit()
except sqlite3.Error:
    conn.rollback()
    raise

To disable implicit transaction management and run in autocommit mode, set isolation_level to None:

conn = sqlite3.connect("store.db", isolation_level=None)

In autocommit mode, each statement commits immediately. Use this carefully, as there is no rollback.

Error handling

The sqlite3 module raises specific exceptions for different failure conditions:

import sqlite3

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

try:
    # Violates NOT NULL constraint
    cursor.execute("INSERT INTO products (name, price) VALUES (?, ?)", (None, 19.99))
    conn.commit()
except sqlite3.IntegrityError as e:
    print(f"Constraint violation: {e}")
    conn.rollback()
except sqlite3.OperationalError as e:
    print(f"Database error: {e}")
    conn.rollback()

Common exceptions:

• sqlite3.IntegrityError: constraint violations (NOT NULL, UNIQUE, foreign key)
• sqlite3.OperationalError: table doesn't exist, locked database, syntax errors
• sqlite3.ProgrammingError: misuse of the API (wrong number of parameters, closed cursor)

All of these are subclasses of sqlite3.DatabaseError, which is itself a subclass of the base sqlite3.Error.

Working with pandas

If you have pandas installed, you can move data between DataFrames and SQLite with two functions.

Read a query result into a DataFrame:

import pandas as pd
import sqlite3

conn = sqlite3.connect("store.db")
df = pd.read_sql_query("SELECT * FROM products WHERE stock > 0", conn)
print(df.head())

Write a DataFrame back to a table:

df["discount_price"] = df["price"] * 0.9
df.to_sql("discounted_products", conn, if_exists="replace", index=False)

if_exists accepts "fail" (default), "replace", or "append".

In-memory databases for testing

In-memory databases are a clean way to test database logic without touching the filesystem:

import sqlite3
import unittest

def get_product_count(conn):
    cursor = conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM products")
    return cursor.fetchone()[0]

class TestProductCount(unittest.TestCase):
    def setUp(self):
        self.conn = sqlite3.connect(":memory:")
        cursor = self.conn.cursor()
        cursor.execute("""
            CREATE TABLE products (
                id INTEGER PRIMARY KEY,
                name TEXT,
                price REAL
            )
        """)
        cursor.executemany(
            "INSERT INTO products (name, price) VALUES (?, ?)",
            [("Keyboard", 49.99), ("Mouse", 29.99)]
        )
        self.conn.commit()

    def tearDown(self):
        self.conn.close()

    def test_count(self):
        self.assertEqual(get_product_count(self.conn), 2)

Each test gets a fresh, isolated database with no cleanup required between runs.

Common pitfalls

Forgetting to commit. Changes made with execute() are not saved until you call conn.commit(). If you close the connection without committing, the changes are lost. Use context managers to avoid this.

Thread safety. By default, a sqlite3 connection can only be used from the thread that created it. If you need to share a database across threads, either create separate connections per thread or pass check_same_thread=False when connecting and manage your own locking.

Not closing connections. Open connections hold file locks. In long-running applications, close connections explicitly or use context managers to release them when done.

conn = sqlite3.connect("store.db")
try:
    # do work
finally:
    conn.close()