python 50 lines · 7 steps

Optimistic locking in a Flask PATCH endpoint

A Flask route updates a document only if its version matches, returning 409 when a concurrent edit wins the race.

Explained by highlit
Flask Intermediate optimistic-locking concurrency rest-api sqlalchemy input-validation
1from flask import Blueprint, request, jsonify, abort
2from sqlalchemy import update
3 
4from .models import db, Document
5 
6bp = Blueprint("documents", __name__, url_prefix="/documents")
7 
8 
9@bp.route("/<int:doc_id>", methods=["PATCH"])
10def update_document(doc_id):
11 payload = request.get_json(silent=True) or {}
12 
13 try:
14 expected_version = int(payload["version"])
15 except (KeyError, TypeError, ValueError):
16 abort(400, description="A numeric 'version' field is required.")
17 
18 changes = {
19 field: payload[field]
20 for field in ("title", "body", "status")
21 if field in payload
22 }
23 if not changes:
24 abort(400, description="No updatable fields supplied.")
25 
26 document = db.session.get(Document, doc_id)
27 if document is None:
28 abort(404)
29 
30 stmt = (
31 update(Document)
32 .where(Document.id == doc_id, Document.version == expected_version)
33 .values(**changes, version=Document.version + 1)
34 )
35 result = db.session.execute(stmt)
36 
37 if result.rowcount == 0:
38 db.session.rollback()
39 return (
40 jsonify(
41 error="conflict",
42 message="The document was modified by someone else.",
43 current_version=document.version,
44 ),
45 409,
46 )
47 
48 db.session.commit()
49 db.session.refresh(document)
50 return jsonify(document.to_dict()), 200
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A version column in the WHERE clause lets a single UPDATE detect concurrent modifications without explicit locks.
  2. 2Checking rowcount tells you whether the guarded update actually matched, distinguishing success from a stale-version conflict.
  3. 3Returning 409 with the current version gives clients enough context to retry against fresh data.

Related explainers

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
go 
func (w *Watcher) resetDebounce(d time.Duration) {
	if !w.timer.Stop() {
		select {
		case <-w.timer.C:

Debouncing a stream of events in Go

debounce timers channels
Advanced 7 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps
rust 
use std::collections::VecDeque;
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
 

Building a counting semaphore in Rust

concurrency synchronization condition-variable
Advanced 9 steps

Share this explainer

Here's the card — post it anywhere.

Optimistic locking in a Flask PATCH endpoint — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code