Python SQLite IN Queries: Placeholders, Empty Lists, and Variable-Limit Batching
On this page
You read a list of names from CSV and want to look up their SQLite jobs. Passing a Python list to IN (?) raises a binding error; interpolating strings breaks on names containing quotes. This tutorial builds a reusable standard-library function: one placeholder per value, an explicit empty-input result, and batches sized for the current connection’s variable limit.
The beginner-to-intermediate example creates only an in-memory database and never opens an existing database. It adds a common lookup step after sorting a large CSV with SQLite, rather than repeating the import workflow.
1. Method: generate placeholders, not SQL values
A single ? in IN (?) binds one value. It does not expand a Python list. This minimal example catches and displays the error:
python3 - <<'PYCODE'
import sqlite3
from contextlib import closing
with closing(sqlite3.connect(":memory:")) as con:
con.execute("CREATE TABLE jobs(name TEXT)")
try:
con.execute("SELECT name FROM jobs WHERE name IN (?)", (["alpha", "beta"],))
except sqlite3.ProgrammingError as exc:
print(type(exc).__name__ + ": " + str(exc))
PYCODE
In the tested environment, it prints:
ProgrammingError: Error binding parameter 1: type 'list' is not supported
Instead, generate ?,?,? for the current batch and pass values as the second argument to execute(). Table and column names stay fixed; placeholders do not stand for dynamic table names. Python parameter binding
The function’s contract is explicit: empty input returns [], duplicate names are queried once, missing names produce no row, and results use Python string order rather than input order. SQLite accepts IN (), but returning early avoids a pointless query; not every database accepts an empty IN list. SQLite IN operators
2. Setup: check versions and save the script
Use Python 3.11+ with its bundled sqlite3; no pip packages are needed. Tested on Linux with Python 3.11.4 and SQLite 3.41.2. Run these commands in order in one terminal, choosing a fresh directory name for another attempt:
python3 --version
python3 -c 'import sqlite3; print(sqlite3.sqlite_version)'
mkdir sqlite-in-demo
cd sqlite-in-demo
Save the next section as query_in.py, or download the complete query script. Keep the test file in the same directory.
3. Complete implementation: reserve fixed parameters
The query has status = ? as well as the name list. Its available batch capacity is therefore connection variable limit - 1, capped by --batch-size. Build configuration and runtime settings affect SQLite’s limit: do not hard-code 999; inspect this connection with getlimit(). SQLite variable limits
"""Query a fixed demo schema with bound, limit-aware IN batches. Python 3.11+."""
import argparse
from contextlib import closing
import json
import sqlite3
import sys
def positive_int(text):
value = int(text)
if value < 1:
raise argparse.ArgumentTypeError("must be a positive integer")
return value
def find_jobs(con, names, status="ready", batch_size=500):
if batch_size < 1:
raise ValueError("batch_size must be positive")
names = list(names)
if any(not isinstance(name, str) for name in names):
raise ValueError("names must contain only strings")
unique = list(dict.fromkeys(names))
if not unique:
return []
# One parameter is reserved for status in every SELECT.
available = con.getlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER) - 1
if available < 1:
raise ValueError("variable limit leaves no room for names")
size = min(batch_size, available)
rows = []
for start in range(0, len(unique), size):
batch = unique[start:start + size]
marks = ",".join("?" for _ in batch)
sql = "SELECT name, status FROM jobs WHERE status = ? AND name IN (" + marks + ")"
with closing(con.execute(sql, [status, *batch])) as cur:
rows.extend({"name": name, "status": state} for name, state in cur)
return sorted(rows, key=lambda row: row["name"])
def demo_connection():
con = sqlite3.connect(":memory:", isolation_level=None)
try:
con.execute("CREATE TABLE jobs(name TEXT PRIMARY KEY NOT NULL, status TEXT NOT NULL)")
con.executemany("INSERT INTO jobs VALUES (?, ?)", [
("alpha", "ready"), ("beta", "done"),
("gamma", "ready"), ("o'reilly", "ready"),
])
except BaseException:
con.close()
raise
return con
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
parser.add_argument("names", nargs="*", help="exact names; no names returns []")
parser.add_argument("--status", choices=["ready", "done"], default="ready")
parser.add_argument("--batch-size", type=positive_int, default=500)
args = parser.parse_args(argv)
try:
with closing(demo_connection()) as con:
rows = find_jobs(con, args.names, args.status, args.batch_size)
except (sqlite3.Error, ValueError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 1
print(json.dumps(rows, ensure_ascii=False, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
getlimit() and the test’s setlimit() were added in Python 3.11. The latter changes only that connection and returns the previous limit. Python connection limit interfaces
Three details matter:
dict.fromkeys()deduplicates before batching so a repeated name cannot return the same row across batches. This fixed schema uses default BINARY text comparison; a case-insensitive collation needs an explicit deduplication policy.- Every name, including
o'reilly, is a bound value.markscontains only program-generated question marks and commas. closing()closes cursors and connections; the connection context manager’s commit/rollback behavior is not connection closure.
These are bounded query batches, not an unlimited-memory streaming solution: input names, deduplication data, and results remain in memory. For millions of values, consider a temporary table and JOIN, with a separate result-streaming and transaction design.
4. Run and validate
Execute:
python3 query_in.py --help
python3 query_in.py alpha "o'reilly" gamma alpha missing --batch-size 2
python3 query_in.py
python3 query_in.py alpha beta --status done
The first command shows help. The remaining three print, in order:
[{"name": "alpha", "status": "ready"}, {"name": "gamma", "status": "ready"}, {"name": "o'reilly", "status": "ready"}]
[]
[{"name": "beta", "status": "done"}]
All three queries exit 0; no matches is a valid result. Invalid CLI arguments exit 2. Query or database errors exit 1 and go to standard error. --batch-size 2 counts names only, excluding the separately reserved status parameter.
If you later update matching jobs, continue with UPSERT rather than REPLACE; do not turn this SELECT into multiple write batches by swapping the SQL string alone.
5. Automated tests: lower the limit to exercise batching
Save this as test_query_in.py, or download the test file. The test temporarily lowers the in-memory connection’s variable limit to 3. One status plus two names fit per statement, so four distinct names require two SELECTs.
"""Run with python3 -m unittest -v test_query_in.py."""
from contextlib import closing
import sqlite3
import subprocess
import sys
import unittest
from query_in import demo_connection, find_jobs
class QueryTests(unittest.TestCase):
def test_empty_skips_select(self):
with closing(demo_connection()) as con:
trace = []
con.set_trace_callback(trace.append)
self.assertEqual(find_jobs(con, []), [])
self.assertEqual(trace, [])
def test_single_quote_duplicate_and_missing(self):
with closing(demo_connection()) as con:
rows = find_jobs(con, ["o'reilly", "alpha", "alpha", "missing"], batch_size=1)
self.assertEqual([r["name"] for r in rows], ["alpha", "o'reilly"])
def test_status_filter(self):
with closing(demo_connection()) as con:
self.assertEqual(find_jobs(con, ["alpha", "beta"], "done"),
[{"name": "beta", "status": "done"}])
def test_sql_text_is_just_data(self):
with closing(demo_connection()) as con:
self.assertEqual(find_jobs(con, ["x') OR 1=1 --"]), [])
self.assertEqual(con.execute("SELECT count(*) FROM jobs").fetchone(), (4,))
def test_lowered_limit_and_reserved_parameter(self):
with closing(demo_connection()) as con:
category = sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER
previous = con.setlimit(category, 3)
trace = []
con.set_trace_callback(trace.append)
try:
rows = find_jobs(con, ["alpha", "beta", "gamma", "o'reilly"], batch_size=500)
self.assertEqual(len(rows), 3)
self.assertEqual(len(trace), 2)
finally:
con.setlimit(category, previous)
self.assertEqual(con.getlimit(category), previous)
def test_no_parameter_room(self):
with closing(demo_connection()) as con:
con.setlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, 1)
with self.assertRaisesRegex(ValueError, "no room"):
find_jobs(con, ["alpha"])
def test_invalid_input(self):
with closing(demo_connection()) as con:
with self.assertRaises(ValueError):
find_jobs(con, [1])
with self.assertRaises(ValueError):
find_jobs(con, ["alpha"], batch_size=0)
def test_cli(self):
from pathlib import Path
script = str(Path(__file__).with_name("query_in.py"))
ok = subprocess.run([sys.executable, script, "alpha"], capture_output=True, text=True)
self.assertEqual(ok.returncode, 0)
self.assertIn('"name": "alpha"', ok.stdout)
bad = subprocess.run([sys.executable, script, "--batch-size", "0"],
capture_output=True, text=True)
self.assertEqual(bad.returncode, 2)
self.assertEqual(bad.stdout, "")
if __name__ == "__main__":
unittest.main()
Run the tests, then check the invalid batch size’s exit code:
python3 -m unittest -v test_query_in.py
python3 query_in.py --batch-size 0
printf 'exit=%s\n' "$?"
Expect Ran 8 tests and OK, followed by exit=2 from the last command. Tests cover empty input without a SELECT, quotes, duplicate and missing names, status filtering, SQL-looking text as data, fixed-parameter budgeting, invalid input, and the CLI. The lowered-limit test restores the old value in finally; no system setting or database file changes.
6. Troubleshooting and boundaries
| Symptom | Cause and action |
|---|---|
type 'list' is not supported |
One placeholder received an entire list. Use one placeholder per value and a flattened parameter sequence. |
Incorrect number of bindings supplied |
Placeholder and value counts differ. Check [status, *batch] together with its SQL. |
too many SQL variables |
The real query has extra conditions. Subtract every fixed parameter from the connection limit before choosing a batch size. |
Connection has no getlimit |
The running interpreter predates Python 3.11. Check the actual command with python3 --version. |
| Results differ from input order | IN expresses membership, not ordering. This example sorts by name; input order needs a separate position mapping. |
| Data changes between batches | This in-memory demo has no concurrent writer. For a consistent snapshot on an existing database, the caller should explicitly BEGIN before querying and ROLLBACK or COMMIT afterward, with lock-wait handling. |
Do not apply this merge strategy to NOT IN: concatenating each batch’s exclusions does not implement exclusion against the entire input. NULLs, non-string names, and custom collations also need their own matching contract. For connection and lock problems on existing databases, see diagnosing SQLite database is locked.
7. Summary
Separate SQL structure from data: generate placeholders and bind every value. Handle empty input and duplicates, subtract fixed parameters from the connection limit, and verify batching with a deliberately lowered limit in memory. Before using a concurrent database, decide snapshot, ordering, and memory requirements separately; batching itself is not a transaction guarantee.
- Author: yanzhi
- Original article: https://www.bobobk.com/en/python-sqlite-in-query-batching.html
- License: This work is licensed under CC BY-NC-ND 4.0. See the license for attribution, noncommercial use and no-derivatives terms.
Related articles
- Python SQLite UPSERT vs REPLACE: Keep Related Records When Updating
- Fix Python SQLite Foreign Keys: Connection Checks, Batch Rollback, and Orphan Audits
- Fix Python SQLite database is locked: Timeouts and Short Transactions
- Sort Large CSV Files by Integer Column with Python and SQLite
- Python Timeout Leaves Workers Running? Stop a Linux Process Group