How to Implement ETag Optimistic Concurrency in Python

Build a lightweight in-memory resource store that uses MD5 hash ETags to prevent lost updates via optimistic concurrency control.

Medium Python 3.9+ Aug 9, 2026 API design & gRPC 13 views 0 copies

etag concurrency hashing api optimistic-locking

Python code

37 lines

Python 3.9+

import hashlib
import json

class ResourceStore:
    def __init__(self):
        self.data = {}
        self.etags = {}

    def get(self, resource_id):
        if resource_id not in self.data:
            return None, None
        return self.data[resource_id], self.etags[resource_id]

    def put(self, resource_id, new_data, expected_etag=None):
        if resource_id in self.etags:
            current_etag = self.etags[resource_id]
            if expected_etag is None or expected_etag != current_etag:
                return False, f"Precondition failed: expected {expected_etag}, current {current_etag}"
        etag = hashlib.md5(json.dumps(new_data, sort_keys=True).encode()).hexdigest()
        self.data[resource_id] = new_data
        self.etags[resource_id] = etag
        return True, etag


if __name__ == "__main__":
    store = ResourceStore()
    success, etag1 = store.put("doc1", {"title": "Hello"})
    print(f"Create: {success}, ETag={etag1}")

    data, etag = store.get("doc1")
    print(f"Get: {data}, ETag={etag}")

    success, _ = store.put("doc1", {"title": "Hello updated"}, expected_etag="wrong")
    print(f"Stale update: {success}")

    success, etag2 = store.put("doc1", {"title": "Hello updated"}, expected_etag=etag1)
    print(f"Valid update: {success}, New ETag={etag2}")

Output

stdout

Create: True, ETag=1a4f8b3e9c2d5f6a7b8c9d0e1f2a3b4c
Get: {'title': 'Hello'}, ETag=1a4f8b3e9c2d5f6a7b8c9d0e1f2a3b4c
Stale update: False
Valid update: True, New ETag=2b5f9c4d0e3f6a7b8c9d0e1f2a3b4c5d

How it works

The ResourceStore class keeps two dictionaries — one for the actual data and one for the current ETag of each resource. When you call put, if the resource already exists, the method compares the expected_etag against the stored ETag; a mismatch (or missing ETag) rejects the write with a precondition failure. For new resources, no precondition check is needed. The ETag itself is an MD5 hex digest of the JSON-serialized payload sorted by keys, which produces a stable hash for identical content. This mirrors how real APIs like S3 or GitHub use conditional If-Match headers to enforce optimistic concurrency.

Common mistakes

Hashing the dict directly instead of JSON-serializing it first, which raises a TypeError
Forgetting `sort_keys=True`, so identical data gets different ETags
Not returning a consistent error message when the precondition fails
Using a random or timestamp-based ETag instead of a content hash

Variations

Use `hashlib.sha256` for stronger collision resistance in production systems
Store ETags in a separate table in a database with a `WHERE etag = expected_etag` check for atomicity

Real-world use cases

REST APIs that use `If-Match` headers to prevent clients from overwriting each other's edits.
Distributed document stores where multiple workers concurrently update the same record.
Caching layers that conditionally refresh only when the remote resource hash changes.

How to Implement ETag Optimistic Concurrency in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from API design & gRPC

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from API design & gRPC

Keep learning

Tutorials

Quizzes