A sample app to show how to use optimistic concurrency / optimistic locking in Flask and SQLAlchemy

optimistic-locking.py

"""
A sample app to show how to use optimistic concurrency / optimistic locking in Flask and SQLAlchemy

The trick is to store your entity in session between requests
You need to use Flask-Session, so can serialize your entities properly

Usage:
flask --app optimistic-locking.py run
navigate to http://127.0.0.1:5000/edit/1

to see the error, run the following in a new terminal
AFTER you load the page but BEFORE you hit Save:

sqlite3 instance/app.db "update foo set version=version+1;"

Then hit Save and see the error

requirements.txt:

flask==2.2.3
wtforms==3.0.1
flask-session==0.4.0
flask-sqlalchemy==3.0.3
"""

from flask import Flask, render_template_string, request, session
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm.exc import StaleDataError
from wtforms import Form, StringField, SubmitField

from flask_session import Session

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
app.config["SESSION_TYPE"] = "filesystem"

Session(app)
db = SQLAlchemy(app)


@app.route("/edit/<int:id>", methods=["GET", "POST"])
def edit(id: int):
    if request.method == "GET":
        entity = db.get_or_404(Foo, id)
        form = FooForm(obj=entity)  # load the form w entity attributes
        session["entity"] = entity  # store the entity for retrieval on POST
    else:
        entity = session["entity"]  # get the entity
        form = FooForm(request.form)  # populate form from request
        if form.validate():
            form.populate_obj(entity)  # populate entity from form
            try:
                db.session.merge(entity)
                db.session.commit()
                # TODO: session.pop("entity",None) then redirect to your "success" page
            except StaleDataError:  # if the version # changed between GET and POST will get this error
                form.form_errors = ["Optimistic Lock error. Try reloading"]

    return render_template_string(FORM, form=form)


FORM = """
<form method="post">
{{form.form_errors if form.form_errors}}
{{form.name}}
{{form.save}}
</form>
"""


class Foo(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    version = db.Column(db.Integer, nullable=False)
    __mapper_args__ = {"version_id_col": version}


class FooForm(Form):
    name = StringField("name")
    save = SubmitField("Save")


@app.before_first_request
def before_first_request():
    db.drop_all()
    db.create_all()
    foo1 = Foo(name="foo1", version=1)
    db.session.add(foo1)
    db.session.commit()