SQLite batches, parameters and rollback

pythonapi-clientsautomationindependent-practice

Saved on this device

Full lesson · 4 learning steps Open any exercise for focused practice

Step 1 of 4 · intro card

SQLite batches, parameters and rollback

Concept and contract

Store a small batch with an explicit transaction boundary and close the connection independently of transaction exit.

A local database can retain a normalized result between program runs, but persistence must not turn a partial batch into a successful snapshot. This lesson uses a fictional SQLite file in a disposable workspace.

The database owns a primary-key and nullability boundary while Python owns connection lifetime and the batch transaction. It is not a production schema migration or an invitation to connect the lab to the ApplyDjinn working database.

Mechanism and worked example

The function wraps the connection in contextlib.closing because sqlite3’s connection context manager handles transaction exit rather than closing the connection.

The function wraps the connection in contextlib.closing because sqlite3’s connection context manager handles transaction exit rather than closing the connection. Inside the transaction, parameterized inserts update title and country for an already known identifier.

A constraint failure rolls back the batch. The later SELECT orders rows explicitly and converts SQLite tuples into plain lists for a stable JSON comparison across maintained Python runtimes.

Complete worked example

python

from contextlib import closing
import sqlite3

def store_roles(path, rows):
    with closing(sqlite3.connect(path)) as connection:
        with connection:
            connection.execute("CREATE TABLE IF NOT EXISTS roles (id TEXT PRIMARY KEY NOT NULL CHECK(length(id)>0), title TEXT NOT NULL, country TEXT NOT NULL)")
            for row in rows:
                connection.execute("INSERT INTO roles VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET title=excluded.title, country=excluded.country", row)
        return [list(row) for row in connection.execute("SELECT id, title, country FROM roles ORDER BY id")]

Reasoning: Store a small batch with an explicit transaction boundary and close the connection independently of transaction exit. Compare the stated values and failure boundary before changing the implementation.

Counterexamples and common mistakes

The counterexample commits after every row.

The counterexample commits after every row. If a later row violates a constraint, the first row is already persisted and cannot be undone by exiting the surrounding transaction.

A successful single-row fixture would miss this defect, so the lab starts from saved state and fails a later operation deliberately. Do not build SQL by interpolating caller values, and do not treat connection closure as a substitute for deciding a transaction boundary.

Counterexample: a runnable policy defect

python

from contextlib import closing
import sqlite3

def store_roles(path, rows):
    with closing(sqlite3.connect(path)) as connection:
        with connection:
            connection.execute("CREATE TABLE IF NOT EXISTS roles (id TEXT PRIMARY KEY NOT NULL CHECK(length(id)>0), title TEXT NOT NULL, country TEXT NOT NULL)")
            for row in rows:
                connection.execute("INSERT INTO roles VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET title=excluded.title, country=excluded.country", row)
                connection.commit()
        return [list(row) for row in connection.execute("SELECT id, title, country FROM roles ORDER BY id")]

Reasoning: This runnable counterexample violates the declared contract; the companion lab requires it to fail.

Independent practice: API client task

Add an optional country selection to the returned rows while keeping the entire supplied batch transactional.

Add an optional country selection to the returned rows while keeping the entire supplied batch transactional. None means return every country; an explicitly selected supported country uses a bound parameter.

Start with two countries in one batch and then select Ukraine. This modifies result scope, not storage identity or transaction semantics. The write batch must remain complete even when the returned view shows only one country.

Step 1: Filter the returned rows by an optional supported country. Step 2: Keep all supplied writes in the existing batch transaction.

Practice starter: solve before reading the solution

python

from contextlib import closing
import sqlite3

def store_roles(path, rows):
    # Remove this placeholder, then adapt the retained baseline:
    # Filter the returned rows by an optional supported country.
    raise NotImplementedError
    with closing(sqlite3.connect(path)) as connection:
        with connection:
            connection.execute("CREATE TABLE IF NOT EXISTS roles (id TEXT PRIMARY KEY NOT NULL CHECK(length(id)>0), title TEXT NOT NULL, country TEXT NOT NULL)")
            for row in rows:
                connection.execute("INSERT INTO roles VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET title=excluded.title, country=excluded.country", row)
        return [list(row) for row in connection.execute("SELECT id, title, country FROM roles ORDER BY id")]

Reasoning: Filter the returned rows by an optional supported country. Keep all supplied writes in the existing batch transaction.

Hint 1: Use a WHERE country=? value binding rather than string interpolation.
Hint 2: Keep None as the explicit all-countries selection.

Solution, tests and trade-offs

The cases verify primary-key updates, deterministic ordering and a constraint error.

The cases verify primary-key updates, deterministic ordering and a constraint error. A separate probe seeds the database, inserts one valid new row followed by an invalid one and checks that the seed is the only remaining row.

Practice additionally checks a filtered view and invalid country input. Connection and database files remain inside temporary directories. These checks do not establish concurrent multi-process import behavior or a complete recovery strategy.

Practice solution and acceptance checks

python

from contextlib import closing
import sqlite3

def store_roles(path, rows, country=None):
    if country is not None and (not isinstance(country, str) or country not in {"US", "IL", "UA", "CA"}):
        raise ValueError("supported country required")
    with closing(sqlite3.connect(path)) as connection:
        with connection:
            connection.execute("CREATE TABLE IF NOT EXISTS roles (id TEXT PRIMARY KEY NOT NULL CHECK(length(id)>0), title TEXT NOT NULL, country TEXT NOT NULL)")
            for row in rows:
                connection.execute("INSERT INTO roles VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET title=excluded.title, country=excluded.country", row)
        query = "SELECT id, title, country FROM roles"
        arguments = ()
        if country is not None:
            query += " WHERE country=?"
            arguments = (country,)
        return [list(row) for row in connection.execute(query + " ORDER BY id", arguments)]

Reasoning: Use a WHERE country=? value binding rather than string interpolation. Keep None as the explicit all-countries selection.

Interview reasoning and next step

Explain what is committed, what is returned and which identifiers define the upsert.

Explain what is committed, what is returned and which identifiers define the upsert. The identity policy here belongs to one declared fictional source; cross-source matching requires another explicit model.

Python and SQLite version defaults around transaction control deserve attention, so inspect the selected runtime rather than assuming every connection mode behaves identically. Keep the batch boundary and the separate connection-closing boundary visible in your explanation.

Optional stretch — after the core task works: Add a real two-connection concurrency test before making concurrent-writer claims.

Computer practice: https://applydjinn.com/typorithm/python-automation-lab-20261105.zip. Run verify.py using your selected Python interpreter, then adapt the retained exercise baseline and check it with --practice followed by the lesson id. The standard-library lab uses owned temporary 127.0.0.1 loopback servers, fictional pages and disposable files; it never scans production feeds or opens the working ApplyDjinn database.

Store a small batch with an explicit transaction boundary and close the connection independently of transaction exit.