APPLYDJINN
← SQLAlchemy and Alembic: reliable database workProgress & review ↗

Typorithm · easy

Engine, connections and bound SQL parameters

pythonsqlalchemyalembicdatabasesindependent-practice
Saved on this device
Full lesson · 4 learning steps Open any exercise for focused practice
1

Step 1 of 4 · intro card

Engine, connections and bound SQL parameters

0%

Concept and contract

Own a small database transaction and keep SQL structure separate from input values.

SQLAlchemy is a SQL toolkit as well as an ORM. Begin with Core so that a database connection, a statement and a transaction remain visible before mapped objects add another layer. An Engine owns connection configuration and a pool; it is not itself an open business transaction.

This course uses SQLAlchemy 2.1.8, Alembic 1.20.1 and Python 3.13 in a pinned local lab. Every example owns an in-memory database or a temporary file. The SQLite driver uses autocommit=False so transaction boundaries also cover its DDL and reads.

Mechanism and worked example

The complete example creates a tiny roles table, inserts the supplied scores inside engine.begin(), and reads through another connection.

The complete example creates a tiny roles table, inserts the supplied scores inside engine.begin(), and reads through another connection. Successful exit from begin commits that block; an exception rolls it back. Opening a connection with engine.connect() does not mean that successful exit commits pending writes.

The minimum travels in a parameter dictionary. The SQL text contains :minimum rather than a formatted number or a quoted user string. For scores [4, 2, 4, 1] and minimum 2, the independently expected ordered result is [2, 4, 4]. Duplicate observations remain duplicate rows.

Complete worked example

python

from sqlalchemy import create_engine, text

def inspect_example(scores, minimum):
    engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
    try:
        with engine.begin() as connection:
            connection.execute(text("CREATE TABLE roles (score INTEGER NOT NULL)"))
            if scores:
                connection.execute(text("INSERT INTO roles VALUES (:score)"),
                                   [{"score": value} for value in scores])
        with engine.connect() as connection:
            statement = text("SELECT score FROM roles WHERE score >= :minimum ORDER BY score")
            return list(connection.scalars(statement, {"minimum": minimum}))
    finally:
        engine.dispose()
Reasoning
The complete example creates a tiny roles table, inserts the supplied scores inside engine.begin(), and reads through another connection. Successful exit from begin commits that block; an exception rolls it back. Opening a connection with engine.connect() does not mean that successful exit commits pending writes. The minimum travels in a parameter dictionary. The SQL text contains :minimum rather than a formatted number or a quoted user string. For scores [4, 2, 4, 1] and minimum 2, the independently expected ordered result is [2, 4, 4]. Duplicate observations remain duplicate rows.
Download the SQLAlchemy and Alembic computer lab ↗

Focused decision: context is in the complete example

python

with engine.begin() as connection:
    connection.execute(statement, {"minimum": minimum})
Database decision
begin owns this fixture’s write transaction. The statement and parameter dictionary remain separate; the connection block is not an implicit promise to commit.
Download the SQLAlchemy and Alembic computer lab ↗

Counterexamples and common mistakes

The controlled bug reverses the boundary to score <= minimum.

The controlled bug reverses the boundary to score <= minimum. It still executes valid SQL and returns plausible integers, but it selects the wrong population. A database call succeeding is weaker evidence than checking the intended rows and their order.

Do not interpolate input into SQL text, confuse a connection with a transaction owner, or remove repeated scores merely because they look alike. Parameter binding handles values, not arbitrary identifiers or sort directions. Such structural choices need an explicit allowlist and SQL construction, rather than an invented parameter placeholder for a table name.

Controlled bug: observe the rejected database contract

python

from sqlalchemy import create_engine, text

def inspect_example(scores, minimum):
    engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
    try:
        with engine.begin() as connection:
            connection.execute(text("CREATE TABLE roles (score INTEGER NOT NULL)"))
            if scores:
                connection.execute(text("INSERT INTO roles VALUES (:score)"),
                                   [{"score": value} for value in scores])
        with engine.connect() as connection:
            statement = text("SELECT score FROM roles WHERE score <= :minimum ORDER BY score")
            return list(connection.scalars(statement, {"minimum": minimum}))
    finally:
        engine.dispose()
Reasoning
The controlled bug reverses the boundary to score <= minimum. It still executes valid SQL and returns plausible integers, but it selects the wrong population. A database call succeeding is weaker evidence than checking the intended rows and their order. Do not interpolate input into SQL text, confuse a connection with a transaction owner, or remove repeated scores merely because they look alike. Parameter binding handles values, not arbitrary identifiers or sort directions. Such structural choices need an explicit allowlist and SQL construction, rather than an invented parameter placeholder for a table name.
Download the SQLAlchemy and Alembic computer lab ↗

Independent practice

Start by typing the transaction block and parameter dictionary, then predict which supplied values meet the inclusive minimum.

Start by typing the transaction block and parameter dictionary, then predict which supplied values meet the inclusive minimum. For the independent task change that one boundary to exclusive: a score equal to the minimum must disappear. Keep the repeated values above the minimum and the ascending order.

Use the existing complete program as the exercise baseline. Do not add ORM classes, new packages or a different database while solving this first task. Test an empty list and a boundary containing zero so that the task is about a precise predicate rather than one happy-path sample.

Step 1: Change the lower-bound predicate so a score equal to the minimum is excluded. Step 2: Keep ordering and duplicate observations unchanged.

Practice starter: solve before reading the solution

python

from sqlalchemy import create_engine, text

def inspect_example(scores, minimum):
    engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
    try:
        with engine.begin() as connection:
            connection.execute(text("CREATE TABLE roles (score INTEGER NOT NULL)"))
            if scores:
                connection.execute(text("INSERT INTO roles VALUES (:score)"),
                                   [{"score": value} for value in scores])
        with engine.connect() as connection:
            statement = text("SELECT score FROM roles WHERE score >= :minimum ORDER BY score")
            return list(connection.scalars(statement, {"minimum": minimum}))
    finally:
        engine.dispose()

# Task: Change the lower-bound predicate so a score equal to the minimum is excluded.
Reasoning
Change the lower-bound predicate so a score equal to the minimum is excluded.
Download the SQLAlchemy and Alembic computer lab ↗
  • Hint 1: Find the one comparison in the WHERE clause.
  • Hint 2: Use an input that contains the exact minimum.

Solution, tests and trade-offs

The solution changes >= to > while retaining the bound value.

The solution changes >= to > while retaining the bound value. Its acceptance cases include a score exactly on the boundary, repeated higher scores and an empty input. The baseline is intentionally accepted for the old contract and rejected for the new one, so an unchanged starter cannot count as completed practice.

The lab checks the actual returned scalar values and leaves the caller-owned list unchanged. It also checks disposal of the owned Engine. Creating a new Engine in this fixture isolates each run; a real application normally reuses its configured Engine and gives shorter lifetimes to connections and transactions.

Practice solution and acceptance checks

python

from sqlalchemy import create_engine, text

def inspect_example(scores, minimum):
    engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
    try:
        with engine.begin() as connection:
            connection.execute(text("CREATE TABLE roles (score INTEGER NOT NULL)"))
            if scores:
                connection.execute(text("INSERT INTO roles VALUES (:score)"),
                                   [{"score": value} for value in scores])
        with engine.connect() as connection:
            statement = text("SELECT score FROM roles WHERE score > :minimum ORDER BY score")
            return list(connection.scalars(statement, {"minimum": minimum}))
    finally:
        engine.dispose()
Reasoning
The solution changes >= to > while retaining the bound value. Its acceptance cases include a score exactly on the boundary, repeated higher scores and an empty input. The baseline is intentionally accepted for the old contract and rejected for the new one, so an unchanged starter cannot count as completed practice. The lab checks the actual returned scalar values and leaves the caller-owned list unchanged. It also checks disposal of the owned Engine. Creating a new Engine in this fixture isolates each run; a real application normally reuses its configured Engine and gives shorter lifetimes to connections and transactions.
Download the SQLAlchemy and Alembic computer lab ↗

Interview reasoning and next step

Explain the difference between an Engine, a checked-out connection and an active transaction without substituting their names for one another.

Explain the difference between an Engine, a checked-out connection and an active transaction without substituting their names for one another. Describe what commits on a successful block, what rolls back on an exception, and why bound values do not become SQL syntax.

SQLAlchemy 2.1 requires a maintained Python starting at 3.11; this lab deliberately selects 3.13 for the SQLite autocommit interface. That is a lab requirement rather than a claim that every 2.1 application needs 3.13. Next, map a table explicitly and compare Python annotations with database constraints.

Optional stretch — after the core task works: Add a negative minimum and predict the selected population.

Computer practice: https://applydjinn.com/typorithm/sqlalchemy-lab-20261107.zip. Install the pinned requirements in an isolated Python 3.13 environment and run verify.py. Edit the working exercise baseline, then run --practice sa-engine. The lab uses disposable SQLite databases and actual SQLAlchemy/Alembic calls; it does not connect to application accounts, production databases or paid services. The app includes prepared results and explained questions without an editor or Python runtime.

Own a small database transaction and keep SQL structure separate from input values.