PYTHON / DATABASES WITH PYTHON
sqlite3 and the DB-API
Use sqlite3 as a concrete DB-API 2.0 driver: connections, cursors, description, rowcount, the fetch methods, and the standard exception tree.
What you will learn
- Read sqlite3.apilevel, paramstyle and threadsafety to know what a driver promises
- Get column names from cursor.description and know when rowcount is -1
- Pull rows with fetchone, fetchmany, fetchall, or by iterating the cursor
- Catch sqlite3.IntegrityError vs OperationalError using the PEP 249 hierarchy
Understanding sqlite3 and the DB-API
DB-API 2.0 (PEP 249) is a contract that database drivers agree to, not a library you install. sqlite3 ships with CPython and is the easiest implementation of that contract to experiment with, because sqlite3.connect(":memory:") gives you a real SQL engine with no server, no user account and no network. The module advertises what it supports through module-level constants: sqlite3.apilevel is '2.0', sqlite3.paramstyle is 'qmark' (so placeholders are ?), and sqlite3.threadsafety says how far connections and cursors may be shared between threads. Every other driver you meet later — psycopg, mysql-connector, pyodbc — exposes the same three constants and the same connect/cursor/fetch shape.
The object you actually work with is the cursor, and the mental model is a handle onto exactly one result set at a time. execute() compiles and starts a statement; rows are then pulled lazily with fetchone(), fetchmany(n), fetchall(), or by iterating the cursor. Two attributes describe the last statement: description is a sequence of 7-tuples whose first item is the column name (sqlite3 leaves the other six as None), and rowcount is the number of rows the statement changed, or -1 when the driver cannot know. A SELECT in sqlite3 reports rowcount -1 because SQLite streams rows and has not counted them yet, and running a second execute() on the same cursor silently discards whatever was left of the first result set.
Errors are standardised too. Every sqlite3 exception descends from sqlite3.Error, then DatabaseError, then specific classes: IntegrityError for a violated constraint, OperationalError for things the engine could not do (missing table, locked database), ProgrammingError for API misuse such as wrong placeholder counts. Because those names are attributes of the driver module, code that receives a driver as a parameter can write `except driver.Error` and stay portable. sqlite3 also adds conveniences that are not part of PEP 249 — Connection.execute() as a cursor shortcut, sqlite3.Row for name-based access, executescript() — and they are worth using, as long as you know you have stepped outside the standard.
import sqlite3
print("apilevel", sqlite3.apilevel, "paramstyle", sqlite3.paramstyle)
con = sqlite3.connect(":memory:")
cur = con.cursor()
cur.execute("CREATE TABLE track(id INTEGER PRIMARY KEY, title TEXT, minutes REAL)")
cur.execute("INSERT INTO track(title, minutes) VALUES (?, ?)", ("Aja", 8.0))
print("lastrowid:", cur.lastrowid)
cur.executemany(
"INSERT INTO track(title, minutes) VALUES (?, ?)",
[("Peg", 3.9), ("Deacon Blues", 7.6)],
)
print("rowcount:", cur.rowcount)
cur.execute("SELECT id, title, minutes FROM track ORDER BY id")
print("select rowcount:", cur.rowcount)
print("columns:", [d[0] for d in cur.description])
print("first row:", cur.fetchone())
print("remaining:", cur.fetchall())
print("after exhaustion:", cur.fetchone())
con.close()sqlite3 is one implementation of the DB-API 2.0 contract, where a connection holds the session, a cursor holds a single result set, and both the fetch protocol and the exception hierarchy are the same across drivers.
Worked examples
Rows as tuples vs sqlite3.Row
Shows how setting a row_factory changes what fetch methods hand back, without changing the SQL.
import sqlite3
con = sqlite3.connect(":memory:")
con.row_factory = sqlite3.Row
cur = con.cursor()
cur.execute("CREATE TABLE city(name TEXT, pop INTEGER)")
cur.execute("INSERT INTO city VALUES ('Oslo', 709037), ('Bergen', 291940)")
for row in cur.execute("SELECT name, pop FROM city ORDER BY pop DESC"):
print(row["name"], row["pop"], tuple(row))
row = cur.execute("SELECT * FROM city ORDER BY pop DESC LIMIT 1").fetchone()
print(row.keys(), len(row), row[0])
con.close()Example explained
Line 1row_factory must be set before cursor() is called, because a new cursor copies the connection's factory at creation time.
Line 2execute() returns the cursor itself, so `for row in cur.execute(...)` works; iteration is a documented sqlite3 extension to the DB-API.
Line 3sqlite3.Row supports both row["name"] and row[0], and keys() gives the column names, so it is a superset of a tuple.
Line 4sqlite3.Row is not part of PEP 249, so this exact code will not transfer unchanged to another driver.
The exception hierarchy in practice
Distinguishes a constraint violation from an engine-level failure and prints the inheritance chain that makes generic handling possible.
import sqlite3
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE user(email TEXT PRIMARY KEY)")
con.execute("INSERT INTO user VALUES ('a@example.com')")
try:
con.execute("INSERT INTO user VALUES ('a@example.com')")
except sqlite3.IntegrityError as exc:
print("IntegrityError:", exc)
try:
con.execute("SELECT * FROM missing_table")
except sqlite3.OperationalError as exc:
print("OperationalError:", exc)
print([c.__name__ for c in sqlite3.IntegrityError.__mro__])
print(issubclass(sqlite3.OperationalError, sqlite3.Error))
con.close()Example explained
Line 1con.execute(...) is a sqlite3 shortcut that creates a throwaway cursor for you; PEP 249 only guarantees Connection.cursor().
Line 2A duplicate primary key is a data rule broken by your values, so it raises IntegrityError, which you often want to catch and report to a user.
Line 3A missing table is the engine refusing to run the statement at all, which is OperationalError, usually a bug or a migration problem.
Line 4Since both share the Error ancestor, library code can write `except sqlite3.Error` as a catch-all for anything database-related.
Streaming with fetchmany and reading description
Reads a result set in fixed-size batches and inspects the 7-tuple that describes a computed column.
import sqlite3
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE n(v INTEGER)")
con.executemany("INSERT INTO n VALUES (?)", [(i,) for i in range(1, 8)])
cur = con.cursor()
cur.execute("SELECT v, v * v AS square FROM n ORDER BY v")
print([d[0] for d in cur.description])
print(cur.description[1])
while True:
batch = cur.fetchmany(3)
if not batch:
break
print(batch)
con.close()Example explained
Line 1description gives one 7-tuple per column; sqlite3 fills only index 0 (the name) and leaves type_code, size and precision as None.
Line 2The AS alias becomes the reported column name, which is why aliasing computed columns makes result handling readable.
Line 3fetchmany(3) returns a list of at most three rows and an empty list once the result set is exhausted, which is the loop's exit condition.
Line 4Batching matters on large tables: fetchall() materialises every row in memory at once, fetchmany keeps the peak bounded.
Important notes
paramstyle differs between drivers ('qmark' for sqlite3, 'pyformat' for many others), so the API shape is portable but the placeholder characters are not.
sqlite3.threadsafety reports 1 on Python 3.10 and earlier and 3 on 3.11+, so do not hard-code an assumption about sharing connections between threads.
A ":memory:" database exists only for the life of that one connection; closing it destroys every table.
Common mistakes
Testing `if cur.rowcount == 0` after a SELECT to detect no results: sqlite3 reports -1 for SELECT, so the branch never runs. Check whether fetchone() returned None instead.
Calling execute() a second time on the same cursor and then fetching: the first result set is discarded and you get the second query's rows. Use a separate cursor per result set you still need.
Assuming `with sqlite3.connect(...) as con:` closes the connection. The context manager only ends the transaction; the connection and its file handle stay open until you call con.close().
Try it yourself
Change, predict, then run
In an in-memory database, create a table books(id INTEGER PRIMARY KEY, title TEXT), insert five rows with executemany, print the column names from cursor.description, then print the rows in batches of two using fetchmany(2).
Open the Python workspaceCheck your understanding
You run cur.execute("SELECT * FROM a"), then cur.execute("SELECT * FROM b") on the same cursor, and finally cur.fetchall(). What do you get?
- Only the rows of table b, because a cursor holds one result set and the second execute replaced the first
- The rows of a followed by the rows of b, since results accumulate on the cursor
- A ProgrammingError, because a cursor may not be executed twice
- Only the rows of a, because the first result set must be consumed before the second becomes active
Show answer
A cursor is a handle to exactly one active result set; the second execute() compiles a new statement and throws away whatever remained of the first, so fetchall() sees only b's rows. Nothing raises, which is what makes this bug quiet — re-executing is a legitimate and common operation, so if you still need the first result set you must use a second cursor.