DEV Community: Yasir Jafri

Sorting in Python

Yasir Jafri — Sun, 28 Jun 2026 10:18:01 +0000

Sorting is a fundamental operation in computer science and programming. Whether organizing data for analysis, building efficient algorithms, or enhancing application performance, sorting plays a critical role. Python provides robust tools for sorting and managing sorted data, making it a go-to language for developers. In this article, we’ll explore sorting in Python, covering everything from basics to advanced techniques.

1. Basic Sorting

Python offers two primary methods for sorting collections:

list.sort(): This method sorts a list in place and modifies the original list.
sorted(): This function returns a new sorted list without modifying the original.

Both methods use Timsort, a hybrid sorting algorithm derived from merge sort and insertion sort, ensuring efficiency for real-world data.

Examples:

# Using list.sort()
numbers = [5, 2, 9, 1]
numbers.sort()
print(numbers)  # Output: [1, 2, 5, 9]

# Using sorted()
words = ["apple", "orange", "banana"]
sorted_words = sorted(words)
print(sorted_words)  # Output: ['apple', 'banana', 'orange']

Customization:

Key parameter: Sort elements based on custom logic.

# Sort by length
data = ["pear", "banana", "apple"]
print(sorted(data, key=len))  # Output: ['pear', 'apple', 'banana']

Reverse parameter: Sort in descending order.

print(sorted(numbers, reverse=True))  # Output: [9, 5, 2, 1]

2. Time Complexity of Sorting

Python’s Timsort algorithm has the following complexities:

Best case: O(n) for nearly sorted data.
Average case: O(n log n).
Worst case: O(n log n).

The algorithm’s efficiency stems from its ability to exploit runs (ordered subsequences) within the data and optimize merging operations.

3. Stable Sorting

A sorting algorithm is stable if it preserves the relative order of equal elements. Python’s sort() and sorted() are stable by design, which is useful in scenarios like multi-key sorting.

Example:

students = [("Alice", 90), ("Bob", 90), ("Eve", 85)]
# Sort by score, then by name
sorted_students = sorted(students, key=lambda x: (x[1], x[0]))
print(sorted_students)
# Output: [('Eve', 85), ('Alice', 90), ('Bob', 90)]

4. Sorting Data Structures

`sortedcontainers` Module:

SortedList, SortedDict, and SortedSet maintain data in sorted order dynamically.
Efficient for insertions, deletions, and lookups.

from sortedcontainers import SortedList
sl = SortedList([5, 1, 3])
sl.add(4)
print(sl)  # Output: [1, 3, 4, 5]

`heapq` Module:

Implements a min-heap for priority queues.
Useful for maintaining partial order efficiently.

import heapq
nums = [5, 2, 9, 1]
heapq.heapify(nums)
print(nums)  # Output: [1, 2, 9, 5]

5. Taking a Look at the `bisect` Module

The bisect module provides tools for binary search and maintaining order in sorted lists:

bisect.insort(): Insert while maintaining order.
bisect.bisect_left() and bisect.bisect_right(): Find positions for insertion.

Example:

import bisect
nums = [1, 3, 4, 10]
bisect.insort(nums, 5)
print(nums)  # Output: [1, 3, 4, 5, 10]

6. Sorting with Multiprocessing

Sorting large datasets can benefit from parallel processing. Python’s multiprocessing module can be used to distribute sorting workloads across multiple processors.

Example:

from multiprocessing import Pool

def sort_chunk(chunk):
    return sorted(chunk)

data = [5, 9, 1, 7, 3, 2, 8, 4]
chunks = [data[:4], data[4:]]

with Pool() as pool:
    sorted_chunks = pool.map(sort_chunk, chunks)

# Merge sorted chunks
result = sorted(sum(sorted_chunks, []))
print(result)  # Output: [1, 2, 3, 4, 5, 7, 8, 9]

7. Generator-Based Sorting

For memory-efficient sorting, generators can process data lazily. Use heapq.merge() to sort multiple sorted iterables without loading them entirely into memory.

Example:

import heapq

data1 = iter([1, 4, 7])
data2 = iter([2, 5, 8])

merged = heapq.merge(data1, data2)
print(list(merged))  # Output: [1, 2, 4, 5, 7, 8]

8. External Sorting

For datasets too large to fit into memory, external sorting divides data into manageable chunks, sorts each chunk, and merges them.

Example:

Split large file into smaller sorted chunks.
Use heapq.merge() for final merging.

9. Use Cases

Real-World Applications:

Event Scheduling: Sorting events by timestamps.
Leaderboards: Dynamic ranking systems.
Financial Analysis: Sorting stock data for trend analysis.

10. Conclusion

Sorting is more than an academic exercise; it’s a cornerstone of efficient programming. Python provides powerful tools and libraries to handle sorting for various scenarios, from in-memory operations to large-scale data processing. Understanding these techniques can elevate your problem-solving skills and optimize your applications.

Tap, Tap, Charged Twice: The Idempotency-Key Pattern

Yasir Jafri — Sun, 28 Jun 2026 10:05:45 +0000

A friend once messaged me, mildly furious, with a screenshot from his bank: the same $500 payment, twice, two seconds apart. He'd tapped "Pay," the app spun for a moment, nothing happened, so he tapped again. Except the first tap had gone through - the response just never made it back to his phone. The server happily charged him a second time.

Nobody wrote a bug for this. Every individual piece worked exactly as designed. The payment endpoint did its job, twice, because from its point of view it received two perfectly valid requests. The problem lives in the space between the client and the server, and that space is where a huge category of production incidents hides.

This article is about the pattern that fixes it: idempotency keys. We'll build up from the failure, write a real implementation in Python (FastAPI + SQLAlchemy + Postgres), and - like its sibling, the transactional outbox - we'll spend most of our time on the four hard questions that the tidy "just store the key" tutorials skip.

The villain: retries are not optional

Here's the uncomfortable truth that makes idempotency necessary: on a network, you cannot tell the difference between "my request failed" and "my request succeeded but the response got lost." From the client's side, both look identical - you sent something, you waited, you got nothing back.

So clients retry. They should retry - it's the only sane response to a timeout. Mobile apps retry on flaky connections. Load balancers and API gateways retry on 502s. Message consumers redeliver. SDKs have automatic retry built-in. Users double-tap impatient buttons. Retries are a permanent fact of distributed life, and the only honest delivery guarantee the network gives you is at-least-once.

Which means duplicate requests aren't an edge case you can validate away. They are guaranteed to arrive. The server's job is not to prevent them - it can't - but to absorb them: to make a repeated request harmless. An operation you can safely apply many times with the same result as applying it once is called idempotent. GET, PUT, and DELETE are idempotent by their HTTP definitions. POST - "create a payment," "place an order" - is the dangerous one, and it's exactly the one that moves money.

The idea: a client-supplied key that means "this is the same operation"

The fix is wonderfully simple in concept. The client generates a unique key for each logical operation - one key per "I want to pay $500 for order #42," reused across every retry of that same intent - and sends it as an HTTP header:

POST /payments HTTP/1.1
Idempotency-Key: 5f5c1b0e-0b8a-4c7e-9d6a-2c1f3e4a5b6c
Content-Type: application/json

{ "order_id": "42", "amount_paise": 50000 }

The server makes a promise in return: for a given key, I will execute the operation at most once. If I see the same key again, I will replay the response I already produced - without doing the work a second time.

This isn't a niche trick. Stripe has done it for years, and it's being standardized by the IETF as the Idempotency-Key HTTP header field (a Standards-Track draft, currently -07). If you've used a payments API, you've used this pattern from the outside.

The happy path has two halves - the first request does the work; the second replays the answer:

Easy, right? It is, until you try to make it correct. The simplest implementation is a check-then-act:

# DON'T ship this - it's a race waiting to happen.
async def handle(key, request):
    existing = await store.get(key)
    if existing:
        return existing.response           # replay
    response = await do_the_work(request)
    await store.save(key, response)        # record
    return response

This is broken in at least three ways, and each one is a question worth answering properly.

1. What exactly do you store and where do you intercept?

When you "replay the response," what is the response? Not the return value of your handler function - the actual HTTP response: its status code, a selected set of headers, and the body, byte for byte. A replay that returns 200 when the original returned 201, or drops the Location header, is a subtly broken replay that will confuse clients.

So you must capture and persist something like this:

class StoredResponse(TypedDict):
    status_code: int
    headers: list[tuple[str, str]]   # an allow-list, not everything
    body_b64: str                    # base64 of the raw bytes

A crucial detail: don't replay every header. Store an allow-list (content-type, location, etag, …). Headers like Date, Server, or anything connection-specific should be regenerated fresh, not served from a week-old cache. Replaying them is how you ship weird bugs.

This requirement also answers where the logic lives, and it's a question people get wrong. There are three candidate layers:

A decorator on the handler - but a decorator sees the function's return value, not the serialized HTTP response. It can't faithfully capture the status and headers FastAPI is about to generate.
A dependency (Depends(...)) - runs before the handler returns, so it can detect a replay and short-circuit, but it can't cleanly emit a full stored response without forcing your handlers into an unnatural shape.
ASGI middleware - wraps the entire request/response lifecycle and owns the final response object end to end. This is the only layer that can both intercept early (to replay) and capture late (to store).

So the right home is middleware, with an optional per-route marker to opt routes in:

from starlette.middleware.base import BaseHTTPMiddleware


class IdempotencyMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, store, methods=("POST", "PATCH"), ttl=86_400):
        super().__init__(app)
        self.store = store
        self.methods = set(methods)
        self.ttl = ttl

    async def dispatch(self, request, call_next):
        if request.method not in self.methods:
            return await call_next(request)

        key = request.headers.get("Idempotency-Key")
        if key is None:
            return await call_next(request)   # opt-in: no key, no magic

        body = await request.body()
        fp = fingerprint(request.method, request.url.path, body)

        outcome = await self.store.claim(key, fp, self.ttl)
        if outcome.kind == "conflict":
            return JSONResponse({"error": "idempotency key reuse"}, 409)
        if outcome.kind == "replay":
            return replay(outcome.stored)

        # outcome.kind == "fresh": we own this key - execute once.
        response = await call_next(request)
        await self.store.complete(key, capture(response))
        return response

(There's a real-world wrinkle: reading request.body() in middleware consumes the stream, so you must make it re-readable downstream. It's a few lines of plumbing I'm eliding here to keep the shape clear - but if you implement this, that's the bit that'll bite you first.)

2. What if someone reuses a key with a different request?

Keys are client-generated, and clients have bugs. Sooner or later a buggy client will send key k1 for a $500 payment and then reuse k1 for a $900 payment. If you blindly replay, you'll return the $500 receipt for a $900 charge - or worse, silently swallow the second payment. Both are terrible.

The defense is a fingerprint: a hash of the request's identity - method, path, and canonicalized body. You store it alongside the key, and on every hit you compare:

import hashlib


def fingerprint(method: str, path: str, body: bytes) -> str:
    h = hashlib.sha256()
    h.update(method.encode())
    h.update(b"\x00")
    h.update(path.encode())
    h.update(b"\x00")
    h.update(body)
    return h.hexdigest()

The rule becomes:

Same key, same fingerprint, completed → replay the stored response. (A genuine retry.)
Same key, different fingerprint → 409 Conflict. The client reused a key for a different operation; that's a bug on their side and you refuse to guess.
Same key, same fingerprint, still in flight → wait (see Question 3).

That 409 is you being a good API citizen: you'd rather loudly reject an ambiguous request than quietly do the wrong thing with someone's money. (In practice you canonicalize the body before hashing - sort JSON keys, normalize whitespace - so that semantically identical payloads don't produce different fingerprints over trivial formatting differences.)

3. What happens when duplicates arrive at the same time?

This is the question that separates a toy from a tool, and it's the one the check-then-act code fails hardest. Picture not a slow retry seconds later, but fifty identical requests landing in the same instant - a stampede from an aggressive client SDK, or a gateway that fanned out a retry. Every one of them runs store.get(key), every one finds nothing recorded yet (because none has finished), and every one proceeds to execute. Fifty charges.

The race is structural: "check if it exists" and "record that it exists" are two steps, and concurrency loves a gap between two steps. To close it you need a single-flight lock: a per-key lock that guarantees exactly one request executes while the rest wait and then replay.

On Postgres, the cleanest tool for this is an advisory lock keyed on the idempotency key. You take it at the start, hold it until the transaction commits, and the database serializes everyone competing for that key:

from sqlalchemy import text


async def claim(session, key: str, fp: str, ttl: int) -> Outcome:
    # Serialize everyone using this key until we commit/rollback.
    await session.execute(
        text("SELECT pg_advisory_xact_lock(hashtext(:k))"), {"k": key}
    )

    row = await session.get(IdempotencyKey, key)
    if row is None:
        session.add(IdempotencyKey(key=key, fingerprint=fp,
                                   status="in_flight"))
        return Outcome(kind="fresh")

    if row.fingerprint != fp:
        return Outcome(kind="conflict")          # → 409
    if row.status == "completed":
        return Outcome(kind="replay", stored=row.response)

    # in_flight, but WE now hold the lock - the previous owner is gone.
    # Its work was never recorded, so this request retries it safely.
    return Outcome(kind="fresh")

Why an advisory lock and not a row lock? Because on the very first request there's no row to lock yet - and reaching for a row lock creates an insert race precisely where you can least afford one. An advisory lock lets you serialize on a key that may not exist yet. The honest trade-off: holding pg_advisory_xact_lock for the whole handler means holding a transaction open for the whole handler, which you don't want for slow operations. For those, a short-lived lock plus an in_flight state column is the variation - more bookkeeping, shorter locks.

Redis is the popular alternative for speed: SET key value NX PX <ttl> is an atomic "acquire lock if absent," and a small Lua check-and-delete releases it. It's faster, but read Question 4 before you reach for it on a payments path.

4. What if the request holding the lock crashes?

Here's the failure that turns a clever idea into a 3am outage: Request A acquires the lock, starts executing, and the process dies mid-flight - OOM, deploy, pod eviction. The lock is held. Every retry of that key now blocks forever on a lock owned by a process that no longer exists. The key is wedged, and the customer can never complete their payment.

A correct implementation makes a dead lock-holder release its lock automatically. How depends on the backend:

Postgres advisory locks are owned by the database session. If the connection drops - which is exactly what happens when your process dies - Postgres releases the lock automatically. There's no timeout to tune and no reaper job to write; crash safety is a property of the mechanism. The key simply becomes claimable again, and the next retry re-executes (the original work was never recorded, so re-executing is correct).
Redis SET NX PX relies on the PX expiry: a dead holder's lock evaporates after the TTL. This is where Redis gets subtle. If your handler runs longer than the lock TTL while still alive, the lock can expire out from under a living request, and a second request can start - a double execution. The classic distributed-lock caveat. You mitigate by setting the TTL safely above your handler timeout, but you cannot make it airtight without fencing tokens.

So my honest guidance, the kind I'd give in a design review: use Postgres for payment-critical idempotency - its lock gives you a strong, crash-safe single-flight guarantee. Use Redis when you want speed and can tolerate best-effort single-flight (it's still a massive improvement over nothing). Don't let "Redis is faster" talk you into weaker correctness on the path that moves money.

Putting it together: the state machine

Step back and the whole thing is a small, legible state machine per key:

And the storage is a single table:

from datetime import datetime
from sqlalchemy import DateTime, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column


class IdempotencyKey(Base):
    __tablename__ = "idempotency_key"

    key: Mapped[str] = mapped_column(String, primary_key=True)
    fingerprint: Mapped[str] = mapped_column(String)
    status: Mapped[str] = mapped_column(String)  # in_flight | completed
    response: Mapped[dict | None] = mapped_column(JSONB, nullable=True)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))

That expires_at matters more than it looks. Idempotency keys are not forever - you keep them for a TTL (24 hours is a common choice; Stripe holds them around 24h) and then let them expire so the table doesn't grow without bound and keys can eventually be reused. A periodic sweep deletes rows past expires_at. The TTL is a deliberate product decision: long enough to cover realistic retries, short enough to bound storage.

What idempotency keys do not give you

Every honest pattern comes with a boundary, and naming it is what separates engineering from cargo-culting.

Idempotency keys make a specific request safe to repeat. They do not make your business logic idempotent in general. If two different keys both create a payment for the same order, idempotency keys won't stop you - that's a job for a unique constraint on (order_id) in your domain, a different layer of defense. Keys protect against retries of the same intent, not against distinct requests with the same effect.

They also don't help idempotent-by-definition methods. A GET doesn't need a key; a well-designed PUT (full replacement) is already idempotent. Reserve the machinery for the genuinely unsafe verbs - POST, sometimes PATCH - where re-execution causes harm. Putting an idempotency layer in front of everything is overhead with no payoff.

And one more piece of honesty, echoing the outbox pattern's "exactly-once is a lie": idempotency keys give you at-most-once execution per key, plus faithful replay - which, combined with client retries, feels like exactly-once from the outside. That's the strongest truthful promise. The duplicate request still arrives; you've just made it harmless. That's the whole game in this corner of distributed systems - you can't stop the duplicate, so you make it not matter.

The one-paragraph version

Clients will retry, because the network can't tell a lost response from a failed request, so duplicate requests are guaranteed, not hypothetical. Let clients send an Idempotency-Key header, and promise to execute each key at most once and replay the stored response thereafter. Store the real HTTP response (status, allow-listed headers, body) so replays are faithful. Fingerprint the request and return 409 if a key is reused for a different body. Use a per-key single-flight lock - a Postgres advisory lock for the strong, crash-safe guarantee - so a stampede of simultaneous duplicates produces exactly one execution. Expire keys on a TTL. Do that, and the bug that charged my friend $1000 for a $500 payment simply stops being possible.

This is the inbound cousin of the transactional outbox pattern - together they cover both directions of "do this exactly once" in a payments system. I write about backend reliability and the unglamorous distributed-systems details that only surface in production.

The Transactional Outbox Pattern: Solving the Dual-Write Problem

Yasir Jafri — Sun, 28 Jun 2026 09:33:25 +0000

There's a particular kind of bug I've learned to be afraid of. It doesn't throw an exception. It doesn't show up in your tests. It passes code review, ships on a Friday, and works perfectly for three weeks. Then one night a customer pays for an order, gets charged, and never receives a confirmation. Support can't reproduce it. The logs look clean. The order is right there in the database.

What happened is that your service did two things that looked like one thing:

async def create_order(payload):
    await db.save(order)  # 1. write to your database
    await broker.publish("order.created", evt)  # 2. notify

These two lines are the most innocent-looking distributed systems bug I know. This article is about why they're broken, why none of the "obvious" fixes actually work, and how the transactional outbox pattern fixes them properly, including the parts of the pattern that the tidy architecture diagrams quietly leave out.

The dual-write problem

Step back and look at what those two lines actually do. They write to two different systems, your database and your message broker, and there is no transaction spanning both. The database doesn't know the broker exists. The broker doesn't know the database exists. You are hoping both succeed, and hope is not a delivery guarantee.

Here's the failure that bites people:

The order is safely in your database. The event was never published. Every downstream system: the email service, the analytics pipeline, the ledger, the warehouse, etc have no idea this order exists. Nothing errored. Nothing retried. The event simply evaporated.

And crashing isn't the only way to lose. Walk through the combinations:

DB commits, then crash before publish → order exists, event lost (the case above).
Publish "succeeds" but the ack times out → you retry and publish a duplicate.
You flip the order - publish first, then save → crash in between, and now you've announced an order that doesn't exist.
Publish succeeds, then the DB transaction rolls back → you've told the world about something that never happened.

There is no ordering of two independent writes that survives a crash in the gap between them. That gap is the whole problem, and it has a name: the dual-write problem.

Why the obvious fixes don't work

When engineers first hit this, they reach for one of three fixes. Each one is worth killing explicitly, because each feels like it should work.

"I'll just reorder them / wrap them in try/except." Reordering moves the gap; it doesn't close it. A try/except around the publish lets you detect a failure, but what then? You can't un-commit the database row - it's already committed. You're left logging "publish failed" and... what, hoping someone reads the log? You've turned a silent failure into a slightly louder one.

"I'll use a two-phase commit (2PC) / distributed transaction." In theory, a distributed transaction coordinator could make the DB write and the broker publish atomic. In practice: most modern brokers (Kafka, SQS, Redis Streams) don't support XA transactions with your database, 2PC is operationally miserable, it couples the availability of two systems together, and it scales badly. The industry walked away from distributed transactions for good reasons. This is not the road back.

"I'll add a retry queue / background job that republishes." Now you're getting warmer - you've intuited that the event needs to be durable somewhere so it can be retried. But where do you put it durably? If you put it in a separate queue, you're back to a dual write (save to DB and save to queue). The insight you're circling is: it needs to be durable in the same place as the order.

That last instinct, taken to its conclusion, is the outbox pattern.

The core idea: turn two writes into one

The transactional outbox pattern rests on a single move: stop publishing to the broker inside your request. Instead, write the event as a row in your own database, in the same transaction as the business data.

Your database already gives you atomicity for writes to itself. Two INSERTs in one transaction either both commit or both roll back - that's the one guarantee you can actually lean on. So we use it:

Two parts:

The write side records the event atomically with the business change. No broker call in the request path. If the transaction commits, the event is guaranteed to exist. If it rolls back, so does the event. The dual write is gone.
A separate relay worker reads unpublished events from the outbox table and forwards them to the broker, marking each as published once the broker confirms. If the relay crashes, it just re-reads the table on restart. Nothing is lost, because the events live in durable, transactional storage.

The outbox table is, in effect, a crash-proof queue that happens to live inside your transactional database. That's the entire trick. Everything else is detail - but the details are where this pattern earns its keep, so let's build it.

The write side, in code

First, the table. Nothing exotic, it's a log of things that need to be published:

from datetime import datetime
from uuid import UUID, uuid4

from sqlalchemy import DateTime, Integer, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class OutboxEvent(Base):
    __tablename__ = "outbox_event"

    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    aggregate_id: Mapped[str] = mapped_column(String, index=True)
    topic: Mapped[str] = mapped_column(String)
    payload: Mapped[dict] = mapped_column(JSONB)

    status: Mapped[str] = mapped_column(String, default="pending")
    attempts: Mapped[int] = mapped_column(Integer, default=0)
    max_attempts: Mapped[int] = mapped_column(Integer, default=8)

    available_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    published_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    last_error: Mapped[str | None] = mapped_column(String, nullable=True)

Hold on to aggregate_id, attempts, and available_at - each one earns its place later.

Now the write itself. The non-negotiable rule: the event row must be inserted in the same transaction as the business row. In SQLAlchemy, that means both adds happen inside one session.begin() block:

from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()


@app.post("/orders")
async def create_order(
    req: CreateOrderRequest,
    session: AsyncSession = Depends(get_session),
):
    order = Order(id=uuid4(), total_paise=req.total_paise)

    async with session.begin():  # one transaction...
        session.add(order)  # ...the business row...
        session.add(
            OutboxEvent(  # ...and the event, together.
                aggregate_id=str(order.id),
                topic="orders.created",
                payload={
                    "order_id": str(order.id),
                    "total_paise": req.total_paise,
                },
            )
        )
    # commit happens here - order and event, atomically or not at all

    return {"id": str(order.id)}

That's the heart of the pattern, and it's almost anticlimactically simple. There's no broker in this handler at all. The request is fast (no network call to Kafka/SQS) and it cannot half-succeed. If the server is struck by lightning right after the commit, the order and its event are both safely on disk.

A small but real discipline: resist the temptation to "just publish here too, for low latency." The moment you add a broker call back into this transaction, you've reintroduced the dual write. Latency is the relay's job now.

The relay: a small worker with a hard job

The relay is a separate, long-running process. Conceptually its loop is trivial:

The naive implementation writes itself:

async def run_once(session_factory, publisher):
    async with session_factory() as session:
        async with session.begin():
            events = (
                await session.execute(
                    select(OutboxEvent)
                    .where(OutboxEvent.status == "pending")
                    .limit(100)
                )
            ).scalars().all()

            for event in events:
                await publisher.publish(event)
                event.status = "published"

This works on your laptop. It will hurt you in production, and the reasons it hurts you are exactly the four questions that separate "I read a blog post" from "I've run this." Let's take them one at a time.

1. How do you run more than one relay?

You will want more than one relay process - for availability (if the only relay dies, events stop flowing) and for throughput (one process may not keep up). So picture two relays running that naive loop at the same instant. Both execute SELECT ... WHERE status = 'pending' LIMIT 100. Both get the same 100 rows. Both publish them. Every consumer receives every event twice.

A plain SELECT takes no locks, so the database has no idea these two relays are fighting over the same work.

The fix is a feature Postgres has had since v9.5 and that I genuinely love: SELECT ... FOR UPDATE SKIP LOCKED.

from sqlalchemy import func, select


async def claim_batch(session, batch_size: int = 100):
    stmt = (
        select(OutboxEvent)
        .where(OutboxEvent.status == "pending")
        .where(OutboxEvent.available_at <= func.now())
        .order_by(OutboxEvent.aggregate_id, OutboxEvent.created_at)
        .limit(batch_size)
        .with_for_update(skip_locked=True)   # the magic
    )
    return (await session.execute(stmt)).scalars().all()

FOR UPDATE locks the rows you select so no other transaction can grab them. But on its own, FOR UPDATE makes the second relay block - it waits for the first relay's transaction to finish, and your two processes serialize into one. You paid for concurrency and got a queue.

SKIP LOCKED is the difference. It tells Postgres: "lock the rows I select, and skip any rows that are already locked by someone else - give me the next available ones instead." Now:

Relay A claims and locks rows 1–100.
Relay B runs the same query, sees 1–100 are locked, skips them, and claims 101–200.
Each publishes its own disjoint batch and commits, releasing its locks.

No row is ever claimed twice, and the relays run fully in parallel. This one clause turns the outbox table into a safe concurrent work queue - it's the same mechanism that Postgres-backed job queues use under the hood. If you take one piece of SQL away from this article, make it this one.

2. How do you keep events in order?

Order #42 emits three events over its life: order.created, order.paid, order.shipped. A consumer that sees shipped before created is in a corrupted state. So events for the same entity must arrive in the order they were produced.

The key phrase is for the same entity. Order #42's events have no ordering relationship with order #99's events - they're independent. Forcing a single global order across all events would serialize your entire system for no reason. So the guarantee we actually want is per-aggregate ordering: events sharing an aggregate_id are delivered in production order; everything else is free to be parallel.

That's why the claim query above has ORDER BY aggregate_id, created_at. But here's the subtlety that trips people up, and it's where Questions 1 and 2 collide: SKIP LOCKED happily lets Relay A grab order.created for #42 while Relay B grabs order.paid for #42 in a different batch. Now two relays are racing to publish the same aggregate's events, and B might win - shipping paid before created. The very concurrency we wanted in Question 1 just broke the ordering we wanted in Question 2.

The reconciliation: make the aggregate, not the row, the unit of concurrency. A single relay must own all currently-pending events for a given aggregate_id at once. In practice you either claim by aggregate (lock every pending row for a set of aggregates together and publish each aggregate's events in sequence) or you partition aggregates across relays by hashing aggregate_id, so a given aggregate is only ever handled by one relay. Different aggregates still flow in parallel - that's your throughput - but a single aggregate is always strictly serial.

I call this out because it's the most important design decision in the whole pattern, and it's the one most tutorials skip. "SKIP LOCKED gives you concurrency; per-aggregate ordering constrains it; you reconcile the two by claiming aggregates rather than rows." Say that in a design review and people will know you've actually run this thing.

(If your domain genuinely doesn't need ordering - say, fire-and-forget notifications and then skip all of this and claim rows freely. Ordering is a cost; only pay it when you need it.)

3. What about events that keep failing?

Publishes fail. The trick is realizing there are two kinds of failure and they need opposite treatment.

Transient failures: the broker is briefly down, a network blip, a rate limit. These succeed if you try again later. But "try again" needs manners: retrying instantly and forever just hammers a broker that's already struggling. So you back off exponentially - wait 1s, then 2s, 4s, 8s and you express that in the data, not with sleep():

from datetime import datetime, timedelta, timezone


def next_attempt_at(attempts: int) -> datetime:
    delay = min(2 ** attempts, 300)  # cap at 5 minutes
    return datetime.now(timezone.utc) + timedelta(seconds=delay)

On a failure you bump attempts and set available_at = next_attempt_at(attempts). Remember the claim query filters available_at <= now() - so a failed row simply becomes invisible until its cooldown passes. Backoff falls out of the schema for free. No timers, no background sleeps, nothing to coordinate across relays.

Permanent failures (poison messages): the event is malformed, points at a topic that doesn't exist, or trips a bug. No number of retries will ever help. And this one is genuinely dangerous, because of Question 2: if you enforce per-aggregate ordering, you cannot publish later events for that aggregate until the stuck one succeeds. So one poison event doesn't just fail itself - it freezes that customer's entire event stream. Forever. This is head-of-line blocking, and it's the kind of thing that takes down a feature at 3am.

The fix is a dead-letter table plus a max_attempts cap. After N attempts, you stop retrying, move the row out of the hot path into outbox_dead_letter (with the failure reason and a timestamp), and let the queue behind it flow again:

async def run_once(session_factory, publisher, batch_size: int = 100):
    async with session_factory() as session:
        async with session.begin():
            events = await claim_batch(session, batch_size)

            for event in events:
                try:
                    await publisher.publish(event)
                    event.status = "published"
                    event.published_at = datetime.now(timezone.utc)
                except Exception as exc:  # noqa: BLE001
                    event.attempts += 1
                    event.last_error = str(exc)[:500]
                    if event.attempts >= event.max_attempts:
                        event.status = "dead"  # → dead-letter
                    else:
                        event.available_at = next_attempt_at(event.attempts)
        # COMMIT here releases the row locks held by SKIP LOCKED

Anything landing in the dead-letter table should page someone - it means a real bug needs a human. The point of the DLQ isn't to hide failures; it's to contain them so a single bad event can't hold the rest hostage.

(One trade-off baked into the loop above: we hold the row locks for the whole publish call. Simple and correct. If your broker is slow and you want shorter locks, the variation is to flip rows to a publishing state and commit quickly before publishing - slightly more bookkeeping, shorter lock windows. Start simple; optimize when a benchmark tells you to.)

4. What can you honestly promise? (Spoiler: not exactly-once)

This is the part I care about most, because it's where engineers are tempted to lie - sometimes to others, often to themselves.

Look closely at the success path: the relay publishes to the broker, the broker acks, then the relay writes status = 'published' and commits. Now crash the relay in between - after the ack, before the commit:

The event reached the broker. But the row never got marked published, so on restart the relay finds it still pending and sends it again. A duplicate.

Here's the uncomfortable truth: you cannot close this window. Closing it would require the broker-publish and the DB-update to be one atomic transaction across two systems - which is the dual-write problem all over again, the exact thing this whole pattern exists to avoid. It's turtles all the way down. There is no atomic "publish-and-record" spanning a broker and a database.

So you don't get to eliminate the failure; you only get to choose it:

Mark published before publishing → crash in the window and the row says "done" but nothing went out → lost event. This is at-most-once guarantee. For anything that matters, losing events is unacceptable.
Mark published after the ack → crash in the window and you re-send → duplicate. This is at-least-once guarantee. You never lose; you might double up.

The outbox pattern chooses at-least-once guarantee, because duplicates can be made harmless and lost events can't be recovered. Every event carries a stable id. The consumer deduplicates on it:

# On the consumer side - make reprocessing a no-op.
async def handle(event, session):
    async with session.begin():
        result = await session.execute(
            insert(ProcessedEvent)
            .values(event_id=event.id)
            .on_conflict_do_nothing()  # already seen? skip.
        )
        if result.rowcount == 0:
            return  # duplicate - ignore
        await do_the_actual_work(event)

A one-row INSERT ... ON CONFLICT DO NOTHING (or your broker's native dedup, like SQS FIFO's MessageDeduplicationId) turns a duplicate delivery into a no-op. The observable behavior becomes "processed once." That combination - at-least-once delivery from the relay, deduplication on the consumer - is the strongest honest guarantee there is. The proper name for it is effectively-once, and you should be suspicious of anyone who promises you the other thing. If a system claims true exactly-once delivery across a database, a relay, and a broker, either it's wrong or it's hiding a distributed transaction nobody wanted.

Write "exactly-once" in your docs and you'll be debugging a duplicate charge while re-reading your own marketing. Write "at-least-once, dedupe on event id" and you'll sleep better.

When not to reach for this

A pattern you can't argue against is a pattern you don't understand, so: the outbox is not free. It adds a table, a worker to operate and monitor, a dead-letter queue to watch, and end-to-end latency bounded by your poll interval. If you're building a small monolith where the "downstream consumer" is another function in the same process and the same transaction, you don't need any of this - just call the function. If you can tolerate occasionally losing an event (some analytics, some best-effort notifications), the bookkeeping may not be worth it.

Reach for the outbox when an event leaving your service matters - money moved, an order placed, a contract signed - and a lost or duplicated event has real consequences. That's exactly when those two innocent lines from the top of this article will eventually betray you, and exactly when it's worth the table and the worker to make sure they don't.

The one-paragraph version

If you only remember one thing: never write to your database and your broker as two separate steps. Write the event into an outbox table in the same transaction as your data, and let a separate relay forward it. Use SELECT ... FOR UPDATE SKIP LOCKED so multiple relays can share the work without stepping on each other. Order by aggregate_id and claim per-aggregate when ordering matters. Back off and dead-letter the events that won't go through, so one poison message can't block the rest. And tell the truth about your delivery guarantee: at-least-once, with consumers deduplicating on a stable event id. Do that, and the bug that loses a customer's confirmation email at 2am simply stops being possible.

If this was useful, I write about backend reliability, payments, and the unglamorous distributed-systems details that only show up in production. The idempotency-key pattern - the inbound cousin of the outbox - is next.

Sorted Data Structures in Python

Yasir Jafri — Fri, 27 Dec 2024 17:15:31 +0000

Sorted data structures play a critical role in optimizing search, insertion, and deletion operations while maintaining order. Python provides a variety of tools and libraries to work with such structures, offering efficient solutions for numerous real-world problems. We'll cover the following ones:

Heaps.
Sorted lists.
Sorted dictionaries.
Sorted sets.

`heapq` Module

For a robust implementation of a heap data structure (specifically a min-heap), Python's standard library provides built-in support. The heapq module provides a heap-based priority queue implementation. It uses a binary heap to maintain partial order, making it ideal for scenarios requiring repeated access to the smallest (or largest) element.

Example:

import heapq

heap = [3, 1, 4]
heapq.heapify(heap)
heapq.heappush(heap, 2)
print(heap)  # Output: [1, 2, 4, 3]

smallest = heapq.heappop(heap)
print(smallest)  # Output: 1

Refer to the official documentation for a comprehensive list of available operations and additional examples.

`sortedcontainers` Module

The sortedcontainers module provides dynamic sorted data structures that adjust automatically as elements are added or removed. This library is highly efficient and easy to use.

`SortedList`:

Maintains a sorted list with dynamic ordering.

from sortedcontainers import SortedList

sl = SortedList([3, 1, 4])
sl.add(2)
print(sl)  # Output: [1, 2, 3, 4]

It also accepts a key parameter, similar to the one used in the sorted() function.

from sortedcontainers import SortedList
from operator import neg

sl = SortedList([3, 1, 4], key=neg)
print(sl)  # Output: [4, 3, 1]

Note: SortedList supports almost all the methods of mutable sequences except a few which are not supported and will raise not-implemented error.

`SortedDict`:

A dictionary with keys maintained in sorted order. The design of sorted dict is simple: sorted dict inherits from dict to store items and maintains a sorted list of keys.

Sorted dict keys must be hashable and comparable. The hash and total ordering of keys must not change while they are stored in the sorted dict.

from sortedcontainers import SortedDict

sd = SortedDict({"b": 2, "a": 1})
sd["c"] = 3
print(sd)  # Output: {'a': 1, 'b': 2, 'c': 3}

`SortedSet`:

A set that ensures its elements are sorted.

from sortedcontainers import SortedSet

ss = SortedSet([3, 1, 1, 4])
ss.add(2)
print(ss)  # Output: SortedSet([1, 2, 3, 4])

As with SortedList, SortedSet also accepts a key parameter which can be used in the same way.

Trade-offs of Sorted Data Structures

While sorted data structures offer significant advantages, they come with trade-offs:

Insertion/Deletion Overhead: Maintaining order during these operations may increase computational cost compared to unsorted structures.
Memory Overhead: Some implementations may use additional memory for indexing or maintaining order.

Conclusion

Sorted data structures are indispensable tools for optimizing applications requiring dynamic order maintenance. Although developers should be easily able to implement these data structures, it's nice to have these robust implementations readily available which can be uses right-off the bat without having a nightmare about a corner-case in a service that is deployed in production. Python’s built-in libraries and third-party modules like sortedcontainers provide versatile and efficient solutions for a wide array of problems. By understanding their strengths and trade-offs, you can select the right tools to build performant and scalable applications.

Concurrency Patterns: Balking Pattern

Yasir Jafri — Tue, 24 Dec 2024 18:52:52 +0000

Introduction

The Balking Design Pattern is a behavioral design pattern used to manage state-dependent actions in a system. It ensures that operations are executed only when the system is in an appropriate state. If the required precondition is not met, the operation is aborted or the system "balks". For those like me, who don't know what Balking is, this is what google has to say about it: "hesitate or be unwilling to accept an idea or undertaking". This pattern is particularly useful in multithreaded environments or systems where invalid actions could cause conflicts or errors.

Balking pattern is also considered more of an anti-pattern than a design pattern by some people in the community. If an object cannot support its API, it should either limit the API so that the offending call is not available, or so that the call can be made without limitation. This is an old pattern which seems to have arisen when JVMs were slower and synchronization wasn't as well understood and implemented as it is today. Regardless it is worth discussing and whether to use it or not is upto the developers.

The Balking Pattern relies on three fundamental concepts

Guard Condition: A condition that must be satisfied for an operation to proceed.
State-Dependent Actions: Operations that depend on the current state of the system.
Thread Safety: The pattern often uses locks or other synchronization mechanisms to ensure safety in concurrent environments.

Let's understand these with an example:

A printing system demonstrates the Balking Pattern:

Scenario: A printer can only process one print request at a time. Even though multiple processes can place the print request.
Guard Condition: The printing must not be actively "printing" to handle a new print request.
Behavior: If the printer is busy, the system balks and does not proceed with the new print requests.

Note: Yeah, we can handle this using a queue, but let's assume for now we don't know that such an elegant data structure exists.

import threading
import time

class Printer:
    def __init__(self):
        self.state = "idle"
        self.lock = threading.Lock()

    def start_printing(self, job_id):
        print(f"Attempting to start Print Job {job_id}...")

        with self.lock:  # Ensure thread safety
            if self.state == "printing":
                print(f"Balking: Print Job {job_id} cannot start. Printer is busy.")
                return
            self.state = "printing"

        # Simulate the printing process
        print(f"Print Job {job_id} started.")
        time.sleep(3)
        print(f"Print Job {job_id} completed.")

        with self.lock:
            self.printing = "idle"

# Multiple threads attempting to start print jobs
printer = Printer()

threads = [
    threading.Thread(target=printer.start_printing, args=(1,)),
    threading.Thread(target=printer.start_printing, args=(2,))
]

for t in threads:
    t.start()

for t in threads:
    t.join()

Looking at the code we can see that if we send a print request start_printing to the printer and the printer is busy it will check it's current state self.state and if the state is "printing", it will return without doing anything. Otherwise, it will take up that request and adjust its state accordingly.

When to Use the Balking Pattern

Multithreaded Systems: To prevent race conditions or invalid operations.
State-Dependent Workflows: When actions are permissible only in certain states.
Resource Management: To guard against improper use of shared resources. Objects that use this pattern are generally only in a state that is prone to balking temporarily but for an unknown amount of time. If objects are to remain in a state which is prone to balking for a known, finite period of time, then the guarded suspension pattern may be preferred.

Advantages of the Balking Pattern

Prevents Invalid Operations: Guards ensure operations occur only under valid conditions.
Thread Safety: Particularly useful in multithreaded systems.
Simplifies Logic: Encapsulates state-dependent actions into a clear, reusable pattern.

Disadvantages

Limited Applicability: Most useful when actions are binary (allowed or not allowed).
Potential Overhead: Guard checks and synchronization mechanisms can introduce performance costs.

Conclusion

The Balking Design Pattern provides an effective way to manage state-dependent actions and prevent invalid operations in software systems. By introducing clear guard conditions and ensuring thread safety, it enhances the reliability and maintainability of the system. Whether it's preventing multiple trips in a cab booking system or managing concurrent print jobs, the Balking Pattern offers a structured approach to avoid conflicts and maintain operational integrity. Ultimately, the choice to use the Balking Pattern depends on the specific requirements of your application and its concurrency needs.

References

Concurrency Patterns: Active Object

Yasir Jafri — Sun, 22 Dec 2024 18:59:09 +0000

Introduction

The Active Object Pattern is a concurrency design pattern that decouples method execution from method invocation. The primary goal of this pattern is to introduce asynchronous behavior by executing operations in a separate thread, while providing a synchronous interface to the client. This is achieved using a combination of message passing, request queues, and scheduling mechanisms.

Key Components

Proxy: Represents the public interface to the client. In even simpler terms, this is what the client is going to interact to. It translates method calls into requests for the active object.
Scheduler: Manages the request queue and determines the order of request execution.
Servant: Contains the actual implementation of the methods being invoked. This is where actual computation logic goes.
Activation Queue: Stores the requests from the proxy until the scheduler processes them.
Future/Callback: A placeholder for the result of an asynchronous computation.

Workflow

A client invokes a method on the proxy.
The proxy creates a request and places it in the activation queue.
The scheduler picks up the request and forwards it to the servant for execution.
The result is returned to the client via a future object.

Use Cases

Real-time systems requiring predictable execution patterns.
GUI applications to keep the main thread responsive.
Distributed systems for handling asynchronous requests.

Implementation

Let's say we need to do a computation, maybe a API call, a database query, etc. I am not going to implement any exception handling because I am too lazy.

def compute(x, y):
    time.sleep(2)  # Some time taking task
    return x + y

Without Active Object Pattern

Below is an example of how we might handle concurrent requests without using the Active Object Pattern.

import threading
import time


def main():
    # Start threads directly
    results = {}

    def worker(task_id, x, y):
        results[task_id] = compute(x, y)

    print("Submitting tasks...")
    thread1 = threading.Thread(target=worker, args=(1, 5, 10))
    thread2 = threading.Thread(target=worker, args=(2, 15, 20))

    thread1.start()
    thread2.start()

    print("Doing other work...")

    thread1.join()
    thread2.join()

    # Retrieve results
    print("Result 1:", results[1])
    print("Result 2:", results[2])


if __name__ == "__main__":
    main()

Drawbacks of the Above Approach

Thread Management: Direct management of threads increases complexity, especially as the number of tasks grows.
Lack of Abstraction: The client is responsible for managing the lifecycle of threads, coupling task management with business logic.
Scalability Issues: Without a proper queue or scheduling mechanism, there’s no control over task execution order.
Limited Responsiveness: The client has to wait for threads to join before accessing results.

Implementation using Active Object Pattern

Below is a Python implementation of the Active Object Pattern using threading and queues for doing the same thing as above. We'll walk through each part one by one:

MethodRequest: Encapsulates the method, arguments, and a Future to store the result.

class MethodRequest:
    def __init__(self, method, args, kwargs, future):
        self.method = method
        self.args = args
        self.kwargs = kwargs
        self.future = future

    def execute(self):
        try:
            result = self.method(*self.args, **self.kwargs)
            self.future.set_result(result)
        except Exception as e:
            self.future.set_exception(e)

Scheduler: Continuously processes requests from the activation_queue in a separate thread.

import threading
import queue


class Scheduler(threading.Thread):
    def __init__(self):
        super().__init__()
        self.activation_queue = queue.Queue()
        self._stop_event = threading.Event()

    def enqueue(self, request):
        self.activation_queue.put(request)

    def run(self):
        while not self._stop_event.is_set():
            try:
                request = self.activation_queue.get(timeout=0.1)
                request.execute()
            except queue.Empty:
                continue

    def stop(self):
        self._stop_event.set()
        self.join()

Servant: Implements the actual logic (e.g., the compute method).

import time


class Servant:
    def compute(self, x, y):
        time.sleep(2)
        return x + y

Proxy: Translates method calls into requests and returns a Future for the result.

from concurrent.futures import Future


class Proxy:
    def __init__(self, servant, scheduler):
        self.servant = servant
        self.scheduler = scheduler

    def compute(self, x, y):
        future = Future()
        request = MethodRequest(self.servant.compute, (x, y), {}, future)
        self.scheduler.enqueue(request)
        return future

Client: Submits tasks asynchronously and retrieves results when needed.

def main():
    # Initialize components
    scheduler = Scheduler()
    scheduler.start()

    servant = Servant()
    proxy = Proxy(servant, scheduler)

    # Client makes an asynchronous call
    print("Submitting tasks...")
    future1 = proxy.compute(5, 10)
    future2 = proxy.compute(15, 20)

    # Perform other tasks while computation is ongoing
    print("Doing other work...")

    # Retrieve results
    print("Result 1:", future1.result())
    print("Result 2:", future2.result())

    # Shutdown scheduler
    scheduler.stop()

if __name__ == "__main__":
    main()

Advantages

Decoupled Interface: Clients can invoke methods without worrying about the execution details.
Responsiveness: Asynchronous execution ensures that the client remains responsive.
Scalability: Supports multiple concurrent requests.

Disadvantages

Complexity: Increases architectural complexity.
Overhead: Requires additional resources for managing threads and queues.
Latency: Asynchronous processing may introduce additional latency.

Conclusion

The Active Object Pattern is a powerful tool for managing asynchronous operations in multi-threaded environments. By separating method invocation from execution, it ensures better responsiveness, scalability, and a cleaner codebase. While it comes with some complexity and potential performance overhead, its benefits make it an excellent choice for scenarios requiring high concurrency and predictable execution. However, its use depends on the specific problem at hand. As with most patterns and algorithms, there is no one-size-fits-all solution.

References

Wikipedia - Active Object

DEV Community: Yasir Jafri

Sorting in Python

1. Basic Sorting

Examples:

Customization:

2. Time Complexity of Sorting

3. Stable Sorting

Example:

4. Sorting Data Structures

sortedcontainers Module:

heapq Module:

5. Taking a Look at the bisect Module

Example:

6. Sorting with Multiprocessing

Example:

7. Generator-Based Sorting

Example:

8. External Sorting

Example:

9. Use Cases

Real-World Applications:

10. Conclusion

Tap, Tap, Charged Twice: The Idempotency-Key Pattern

The villain: retries are not optional

The idea: a client-supplied key that means "this is the same operation"

1. What exactly do you store and where do you intercept?

2. What if someone reuses a key with a different request?

3. What happens when duplicates arrive at the same time?

4. What if the request holding the lock crashes?

Putting it together: the state machine

What idempotency keys do not give you

The one-paragraph version

The Transactional Outbox Pattern: Solving the Dual-Write Problem

The dual-write problem

Why the obvious fixes don't work

The core idea: turn two writes into one

The write side, in code

The relay: a small worker with a hard job

1. How do you run more than one relay?

2. How do you keep events in order?

3. What about events that keep failing?

4. What can you honestly promise? (Spoiler: not exactly-once)

When not to reach for this

The one-paragraph version

Sorted Data Structures in Python

heapq Module

Example:

sortedcontainers Module

SortedList:

SortedDict:

SortedSet:

Trade-offs of Sorted Data Structures

Conclusion

Concurrency Patterns: Balking Pattern

Introduction

The Balking Pattern relies on three fundamental concepts

When to Use the Balking Pattern

Advantages of the Balking Pattern

Disadvantages

Conclusion

References

Concurrency Patterns: Active Object

Introduction

Key Components

Workflow

Use Cases

Implementation

Without Active Object Pattern

Drawbacks of the Above Approach

Implementation using Active Object Pattern

Advantages

Disadvantages

Conclusion

References

`sortedcontainers` Module:

`heapq` Module:

5. Taking a Look at the `bisect` Module

`heapq` Module

`sortedcontainers` Module

`SortedList`:

`SortedDict`:

`SortedSet`: