Order status state machine with allowed transitions and select_for_update

Backend Live Coding November 13, 2026 python

The transitions are data (a dict), not if-else scattered across views, so they are visible at a glance and can be tested exhaustively (for every pair of statuses I check allowed/forbidden).

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
# ======================================================================
# Q: Write a state machine for an order's status with allowed transitions.
#
# The transitions are data (a dict), not if-else scattered across views, so they
# are visible at a glance and can be tested exhaustively (for every pair of
# statuses I check allowed/forbidden). select_for_update prevents two concurrent
# transitions (e.g. a payment webhook and a cancellation by the customer). The
# status changes only from this function: no order.status = ... anywhere else.
# There is django-fsm as an alternative, but for six states I prefer explicit
# code.
# ======================================================================

from enum import StrEnum
from django.db import transaction
from django.utils import timezone

class Status(StrEnum):
    DRAFT = 'draft'
    PLACED = 'placed'
    PAID = 'paid'
    SHIPPED = 'shipped'
    DELIVERED = 'delivered'
    CANCELLED = 'cancelled'

TRANSITIONS: dict[Status, set[Status]] = {
    Status.DRAFT:     {Status.PLACED, Status.CANCELLED},
    Status.PLACED:    {Status.PAID, Status.CANCELLED},
    Status.PAID:      {Status.SHIPPED, Status.CANCELLED},    # cancel => refund
    Status.SHIPPED:   {Status.DELIVERED},
    Status.DELIVERED: set(),
    Status.CANCELLED: set(),
}

class InvalidTransition(Exception):
    pass

@transaction.atomic
def transition(order_id: int, target: Status, actor: User) -> Order:
    order = Order.objects.select_for_update().get(pk=order_id)   # serialization
    current = Status(order.status)
    if target not in TRANSITIONS[current]:
        raise InvalidTransition(f'{current} -> {target}')
    order.status = target
    order.save(update_fields=['status', 'updated_at'])
    OrderEvent.objects.create(order=order, from_status=current, to_status=target,
                              actor=actor, at=timezone.now())       # audit trail
    if target is Status.CANCELLED and current is Status.PAID:
        transaction.on_commit(lambda: refund_payment.delay(order.id))
    if target is Status.SHIPPED:
        transaction.on_commit(lambda: send_tracking_email.delay(order.id))
    return order