Flask: rollback and honest transaction outcomes

pythonflaskindependent-practice

Saved on this device

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

Step 1 of 4 · intro card

Flask: rollback and honest transaction outcomes

Concept and contract

Flush pending writes, inject a controlled failure and verify the surviving state.

A transaction groups related database changes into a success or rollback boundary. Flushing sends pending statements to the database within that transaction, while committing completes it.

These are different events. A successful flush does not mean a later validation or workflow failure can be ignored, and a response should not claim durable success before the required transaction outcome is known.

Mechanism and worked example

The helper creates one owned database, stages a row and flushes so the write really reaches the transaction.

The helper creates one owned database, stages a row and flushes so the write really reaches the transaction. A controlled exception then either triggers rollback or allows commit.

A selected read afterward reveals the surviving rows. This order matters: raising before any SQL executes would be a weaker demonstration of rollback than the actual flushed-write scenario.

Complete worked example

python

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import select, String
from sqlalchemy.orm import Mapped, mapped_column

def make_store():
    app = Flask(__name__)
    app.config.update(TESTING=True, SQLALCHEMY_DATABASE_URI='sqlite:///:memory:')
    db = SQLAlchemy(app)
    class Role(db.Model):
        id: Mapped[int] = mapped_column(primary_key=True)
        title: Mapped[str] = mapped_column(String(20), nullable=False)
    with app.app_context():
        db.create_all()
    return app, db, Role

def transaction_result(fail):
    app, db, Role = make_store()
    with app.app_context():
        try:
            db.session.add(Role(title='API'))
            db.session.flush()
            if fail:
                raise ValueError('controlled failure')
            db.session.commit()
        except ValueError:
            db.session.rollback()
        return list(db.session.scalars(select(Role.title).order_by(Role.id)))

Reasoning: The helper creates one owned database, stages a row and flushes so the write really reaches the transaction. A controlled exception then either triggers rollback or allows commit.

Counterexamples and common mistakes

The counterexample commits inside the error branch, preserving a row after the operation reported failure..

The counterexample commits inside the error branch, preserving a row after the operation reported failure.

Swallowing an exception without resetting session state can cause additional problems after database errors. Catch only the failures you intend to handle, rollback at the correct boundary and re-raise or translate unexpected faults according to the application’s error policy.

Counterexample: a runnable framework mistake

python

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import select, String
from sqlalchemy.orm import Mapped, mapped_column

def make_store():
    app = Flask(__name__)
    app.config.update(TESTING=True, SQLALCHEMY_DATABASE_URI='sqlite:///:memory:')
    db = SQLAlchemy(app)
    class Role(db.Model):
        id: Mapped[int] = mapped_column(primary_key=True)
        title: Mapped[str] = mapped_column(String(20), nullable=False)
    with app.app_context():
        db.create_all()
    return app, db, Role

def transaction_result(fail):
    app, db, Role = make_store()
    with app.app_context():
        try:
            db.session.add(Role(title='API'))
            db.session.flush()
            if fail:
                raise ValueError('controlled failure')
            db.session.commit()
        except ValueError:
            db.session.commit()
        return list(db.session.scalars(select(Role.title).order_by(Role.id)))

Reasoning: The counterexample commits inside the error branch, preserving a row after the operation reported failure.

Independent practice: one bounded framework change

Add a second staged Role in the same transaction before the controlled failure.

Add a second staged Role in the same transaction before the controlled failure. A successful operation should keep both rows in order; a failed one should keep neither.

Preserve the flush before failure so the rollback test remains meaningful. This task changes the size of one atomic group rather than introducing retries, nested savepoints or background delivery.

Step 1: Stage a second row inside the same atomic operation. Step 2: Add QA before the existing flush.

Practice starter: solve before reading the solution

python

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import select, String
from sqlalchemy.orm import Mapped, mapped_column

def make_store():
    app = Flask(__name__)
    app.config.update(TESTING=True, SQLALCHEMY_DATABASE_URI='sqlite:///:memory:')
    db = SQLAlchemy(app)
    class Role(db.Model):
        id: Mapped[int] = mapped_column(primary_key=True)
        title: Mapped[str] = mapped_column(String(20), nullable=False)
    with app.app_context():
        db.create_all()
    return app, db, Role

def transaction_result(fail):
    # Remove this placeholder, then adapt the retained baseline:
    # Stage a second row inside the same atomic operation.
    raise NotImplementedError
    app, db, Role = make_store()
    with app.app_context():
        try:
            db.session.add(Role(title='API'))
            db.session.flush()
            if fail:
                raise ValueError('controlled failure')
            db.session.commit()
        except ValueError:
            db.session.rollback()
        return list(db.session.scalars(select(Role.title).order_by(Role.id)))

Reasoning: Stage a second row inside the same atomic operation. Add QA before the existing flush.

Hint 1: Keep one commit on success.
Hint 2: Keep rollback on the controlled failure.

Solution, tests and trade-offs

Cases alternate successful and failed transactions using a fresh factory-created database each time.

Cases alternate successful and failed transactions using a fresh factory-created database each time. Practice checks both titles after success and an empty collection after failure.

Inspect selected database state instead of a local success flag. SQLite’s owned transaction provides useful learning evidence, but other database engines and concurrent operations still require their own integration and isolation tests.

Practice solution and acceptance checks

python

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import select, String
from sqlalchemy.orm import Mapped, mapped_column

def make_store():
    app = Flask(__name__)
    app.config.update(TESTING=True, SQLALCHEMY_DATABASE_URI='sqlite:///:memory:')
    db = SQLAlchemy(app)
    class Role(db.Model):
        id: Mapped[int] = mapped_column(primary_key=True)
        title: Mapped[str] = mapped_column(String(20), nullable=False)
    with app.app_context():
        db.create_all()
    return app, db, Role

def transaction_result(fail):
    app, db, Role = make_store()
    with app.app_context():
        try:
            db.session.add(Role(title='API'))
            db.session.add(Role(title='QA'))
            db.session.flush()
            if fail:
                raise ValueError('controlled failure')
            db.session.commit()
        except ValueError:
            db.session.rollback()
        return list(db.session.scalars(select(Role.title).order_by(Role.id)))

Reasoning: Cases alternate successful and failed transactions using a fresh factory-created database each time. Practice checks both titles after success and an empty collection after failure.

Interview reasoning and next step

Production workflows may include email, queue messages or other effects that cannot be rolled back by a database transaction..

Production workflows may include email, queue messages or other effects that cannot be rolled back by a database transaction.

Publish such effects through an explicit after-commit or outbox design rather than placing an irreversible send between database writes. Keep that extension separate from this lesson’s local atomicity proof, and document what a caller can safely retry without creating duplicate effects.

Optional stretch — after the core task works: Design an outbox record committed with the business rows.

Computer practice: https://applydjinn.com/typorithm/python-frameworks-lab-20261105.zip. Install the pinned dependencies in an isolated environment, run verify.py --track flask, then edit the retained exercise baseline and check it with --practice fl-transactions. The lab uses owned in-memory databases, local framework clients and tiny CPU tensors. It never reads the working ApplyDjinn database, contacts production feeds or downloads models.

Flush pending writes, inject a controlled failure and verify the surviving state.