Автомат на бумаге ничего не гарантирует: пока он не перенесён в код так, чтобы недопустимые переходы стали невозможны, статус заказа меняет кто угодно и куда угодно. В Python для этого четыре способа, от трёх строк до библиотеки, и у каждого своё место. Отдельный вопрос, одинаковый для всех четырёх, — где хранить состояние и как не затереть его при двух одновременных запросах.
StrEnum и match
Самый прямой способ: состояния и события это перечисления, а логика перехода это одна функция с match, которая по текущему состоянию и событию решает, куда двигаться дальше. StrEnum даёт значение, которое в базе и в логах читается как есть, а проверка типов не даст передать событие туда, где ждут состояние.
from enum import StrEnum
class State(StrEnum):
NEW = "NEW"
PAID = "PAID"
SHIPPED = "SHIPPED"
DELIVERED = "DELIVERED"
CANCELLED = "CANCELLED"
class Event(StrEnum):
PAY = "PAY"
SHIP = "SHIP"
DELIVER = "DELIVER"
CANCEL = "CANCEL"
class IllegalTransition(Exception):
def __init__(self, current: State, event: Event) -> None:
super().__init__(f"переход из {current} по событию {event} запрещён")
self.current = current
self.event = event
def next_state(current: State, event: Event) -> State:
match current, event:
case State.NEW, Event.PAY:
return State.PAID
case State.NEW, Event.CANCEL:
return State.CANCELLED
case State.PAID, Event.SHIP:
return State.SHIPPED
case State.PAID, Event.CANCEL:
return State.CANCELLED
case State.SHIPPED, Event.DELIVER:
return State.DELIVERED
raise IllegalTransition(current, event)
Запрещённый переход это не assert и не ValueError со строкой, а свой класс исключения: обработчик HTTP узнаёт его по типу и отвечает 409 с кодом, а остальные ошибки остаются пятисоткой — как устроен такой перевод, разобрано в статье про единый обработчик ошибок.
Плюсы. Ничего не нужно подключать, всё видно в одном месте, читается сверху вниз. Для маленького автомата это идеальный вариант.
Про помощь инструментов надо сказать честно. Интерпретатор полноту match не проверяет: добавите шестое состояние, и код запустится как ни в чём не бывало, а новое состояние тихо провалится мимо всех веток к IllegalTransition. Эту дыру частично закрывает mypy: если match исчерпывающий и после него стоит assert_never(current), то новое значение перечисления даст ошибку типов; pyright в строгом режиме делает то же. Без проверки типов в сборке полнота переходов держится только на тестах, о которых ниже.
Минусы. Как только состояний и событий становится много, match разрастается в длинную лесенку, в которой легко ошибиться и трудно увидеть картину целиком. Логика перехода размазана по коду, а не описана как данные: новую связку «состояние плюс событие» приходится вписывать руками в нужную ветку. Пока переходов десяток, терпимо, дальше становится больно.
Таблица переходов
Следующий шаг: вынести правила из кода в данные. Автомат это по сути таблица: для пары «состояние плюс событие» указано новое состояние. В Python такая таблица это словарь с ключом-кортежем:
TRANSITIONS: dict[tuple[State, Event], State] = {
(State.NEW, Event.PAY): State.PAID,
(State.NEW, Event.CANCEL): State.CANCELLED,
(State.PAID, Event.SHIP): State.SHIPPED,
(State.PAID, Event.CANCEL): State.CANCELLED,
(State.SHIPPED, Event.DELIVER): State.DELIVERED,
}
def next_state(current: State, event: Event) -> State:
try:
return TRANSITIONS[(current, event)]
except KeyError:
raise IllegalTransition(current, event) from None
Теперь весь автомат виден одним взглядом, это буквально список разрешённых переходов. Чтобы изменить правила, не нужно трогать логику: добавили или убрали строку в словаре, и поведение поменялось.
Но у правил-данных есть та же слабость, что у match: забытая пара «состояние плюс событие» никак себя не проявит, она просто окажется запрещённой, и никто этого не заметит. Поэтому таблицу покрывают тестом на полноту. Перечисление само знает все свои значения, и список рядом с константами, как в языках без enum, вести не нужно:
from itertools import product
FORBIDDEN: set[tuple[State, Event]] = {
(State.NEW, Event.SHIP),
(State.NEW, Event.DELIVER),
}
def test_every_pair_is_decided():
undecided = [pair for pair in product(State, Event) if pair not in TRANSITIONS and pair not in FORBIDDEN]
assert not undecided, f"не решено, что делать: {undecided}"
Добавили новое состояние в State, и тест сразу покраснеет на четырёх непокрытых парах и заставит про каждую подумать. Без такого теста «правила в данных» теряют переходы так же молча, как match.
Когда подходит. Средний по размеру автомат, где переходов много, но каждый из них это просто «смена состояния», без сложной сопутствующей логики. Если же для каждого перехода нужно выполнять разное поведение (отправить письмо, списать деньги, проверить условие), таблица «состояние в состояние» этого не выражает, и тут напрашивается следующий приём.
Guard: где живёт условие перехода
Условие перехода это то, что отличает учебный автомат от рабочего: «отменить можно, но только пока не отгружено», «оплатить можно, только если сумма совпадает». В трёх самописных вариантах guard живёт в трёх разных местах, и это как раз та разница, ради которой между ними выбирают.
В match guard это условие прямо в ветке — у case есть if:
case State.NEW, Event.PAY if order.paid != order.total:
raise PartialPayment(order.paid, order.total)
case State.NEW, Event.PAY:
return State.PAID
Просто и читаемо, пока ветвей мало. Минус в том, что условие и переход смешаны: глядя на код, нельзя отдельно ответить «какие вообще есть правила».
В таблице переходов guard становится полем записи, и вот это самое ценное свойство таблицы, потому что правила остаются данными. Отказ это своё исключение с кодом: по нему обработчик отвечает нужным code в Problem Details, а тест проверяет pytest.raises.
from collections.abc import Callable
from dataclasses import dataclass
class PartialPayment(Exception):
code = "PARTIAL_PAYMENT"
@dataclass(frozen=True)
class Rule:
to: State
guard: Callable[["Order"], None] = lambda order: None
def full_payment(order: "Order") -> None:
if order.paid != order.total:
raise PartialPayment()
RULES: dict[tuple[State, Event], Rule] = {
(State.NEW, Event.PAY): Rule(State.PAID, guard=full_payment),
(State.NEW, Event.CANCEL): Rule(State.CANCELLED),
(State.PAID, Event.SHIP): Rule(State.SHIPPED),
(State.PAID, Event.CANCEL): Rule(State.CANCELLED),
(State.SHIPPED, Event.DELIVER): Rule(State.DELIVERED),
}
class Order:
def apply(self, event: Event) -> None:
rule = RULES.get((self.status, event))
if rule is None:
raise IllegalTransition(self.status, event)
rule.guard(self)
self.status = rule.to
Что это даёт: весь набор правил читается одним взглядом, у каждого отказа есть свой код, и по этой же таблице автомат проверяется тестом (см. ниже) и рисуется схемой. Цена: условие теперь функция, а не код в контексте, и внутрь неё нельзя затащить лишние зависимости, что, впрочем, скорее плюс.
В состоянии как классе guard это часть метода того состояния, которое его проверяет: PaidState.ship() сам решает, можно ли. Это самый естественный вариант, когда правил много и они разные для каждого состояния.
Практическое правило: guard, который зависит только от самого объекта, живёт в автомате; guard, которому нужны внешние данные, нет. «Не отгружать, пока не оплачено» в автомате. «Не отгружать, если товара нет на складе» это проверка до вызова перехода, потому что для неё нужен запрос в другой контекст, а автомат не должен ходить по сети — тем более что метод перехода синхронный, и await внутри него не написать.
Состояние как класс
Если в каждом состоянии много собственного поведения, разумно сделать состояние отдельным классом. Это классический паттерн «Состояние» из каталога GoF: есть базовый класс с методами-событиями, и по одному классу на каждое состояние. Каждый класс сам знает, как отвечать на события: и куда переходить, и что при этом делать.
Базовый класс отвергает любое событие, а состояние наследует его и переопределяет только то, что умеет:
class OrderState:
name: State
def pay(self) -> "OrderState":
raise IllegalTransition(self.name, Event.PAY)
def ship(self) -> "OrderState":
raise IllegalTransition(self.name, Event.SHIP)
def cancel(self) -> "OrderState":
raise IllegalTransition(self.name, Event.CANCEL)
class NewState(OrderState):
name = State.NEW
def pay(self) -> OrderState:
return PaidState()
def cancel(self) -> OrderState:
return CancelledState()
class PaidState(OrderState):
name = State.PAID
def ship(self) -> OrderState:
return ShippedState()
def cancel(self) -> OrderState:
return CancelledState()
Каждый класс отвечает только за одно состояние: видно, какие события оно принимает, а какие отвергает, и вся логика этого состояния собрана рядом. Добавить состояние значит добавить класс, не трогая остальные. Плата за это много мелких классов и рассеянность общей картины: чтобы понять весь автомат целиком, нужно открыть все реализации сразу (таблица переходов в этом смысле нагляднее). Поэтому состояние как класс оправдано именно тогда, когда важнее поведение внутри состояний, а не обзор переходов между ними.
Остаётся вопрос, который в описании паттерна обычно опускают: в базе-то лежит не объект, а строка. Мост между ними это словарь «значение перечисления в класс»:
STATES: dict[State, type[OrderState]] = {
State.NEW: NewState,
State.PAID: PaidState,
State.SHIPPED: ShippedState,
State.DELIVERED: DeliveredState,
State.CANCELLED: CancelledState,
}
def state_for(value: State) -> OrderState:
return STATES[value]()
Цикл замыкается: загрузили заказ, взяли state_for(State(row.status)), вызвали событие, получили новый объект состояния и записали в колонку его name. Сами объекты-состояния при этом остаются без полей, всё изменяемое живёт в заказе, а состояние только решает, что с ним можно сделать.
Готовая библиотека: transitions
Когда автомат большой и обвешан требованиями, свою реализацию поддерживать становится дорого. В Python самая распространённая библиотека — transitions: автомат описывается списком состояний и переходов, у перехода есть условия и действия, состояния умеют вкладываться друг в друга, а методы-события библиотека добавляет к вашему объекту сама.
from transitions import Machine, MachineError
TRANSITIONS = [
{"trigger": "pay", "source": "NEW", "dest": "PAID", "conditions": "is_fully_paid", "after": "reserve_stock"},
{"trigger": "ship", "source": "PAID", "dest": "SHIPPED"},
{"trigger": "deliver", "source": "SHIPPED", "dest": "DELIVERED"},
{"trigger": "cancel", "source": ["NEW", "PAID"], "dest": "CANCELLED"},
]
class Order:
def __init__(self, status: State, paid: int, total: int) -> None:
self.paid = paid
self.total = total
self.effects: list[SideEffect] = []
self.machine = Machine(model=self, states=list(State), transitions=TRANSITIONS, initial=status, auto_transitions=False)
def is_fully_paid(self) -> bool:
return self.paid == self.total
def reserve_stock(self) -> None:
self.effects.append(ReserveStock(order_id=self.id))
order = Order(State.NEW, paid=500, total=500)
order.pay()
order.state == "PAID"
order.may_ship()
Три вещи, которые стоит знать до того, как брать. Запрещённое событие это исключение MachineError с текстом вида Can't trigger event pay from state PAID!, а событие, разрешённое таблицей, но не прошедшее условие, исключения не даёт: order.pay() вернёт False, и состояние не изменится. Обе ситуации хочется превращать в свои коды, а не отдавать наружу как есть: MachineError в 409 с ILLEGAL_TRANSITION, ложный результат в проверку условия до вызова и свою ошибку. auto_transitions=False обязателен: без него библиотека добавит объекту методы to_PAID, to_CANCELLED и так далее, которые переводят в любое состояние откуда угодно в обход всех правил. И состояние библиотека держит в памяти объекта: прочитать order.state и записать в базу — ваша работа, как и для асинхронных действий есть отдельный AsyncMachine, где условия и обратные вызовы могут быть корутинами.
Вторая библиотека, python-statemachine, описывает автомат классом с объявленными состояниями и переходами, проверяет его при импорте (недостижимые состояния, конечные состояния с выходом) и умеет рисовать схему. Выбирают по вкусу к стилю описания; обе держат автомат внутри одного процесса, где переходы занимают секунды. Как только процесс растягивается на дни, уходит в несколько сервисов и начинает ждать решения человека, они перестают справляться, и не из-за размера автомата, а из-за того, чего в них нет: таймеров «напомнить через три дня», задач для людей, истории прохождения для бизнеса, компенсаций с порядком отката. Это уже территория движков процессов вроде Temporal или Camunda.
Когда брать. Автомат по-настоящему большой, переходов десятки, нужны подсостояния, действия на входе и выходе и схема автомата, которую можно генерировать, а не рисовать.
Когда избыточна. Для автомата из пяти состояний и десятка переходов библиотека это из пушки по воробьям: конфигурация, зависимость, методы, которые появляются у объекта во время выполнения и не видны проверке типов, и чужой словарь ошибок перевесят выгоду, а match или таблица дадут тот же результат меньшими средствами.
Хранение состояния и гонки
Любой из способов выше отвечает на вопрос «куда перейти». Остаётся второй вопрос: где хранить текущее состояние и как не сломать его при одновременных запросах. Обычно состояние живёт в колонке status таблицы заказа:
Два параллельных перехода из одного состояния без версии затирают друг друга; условие по версии пропускает ровно одного, второй получает конфликт.
CREATE TABLE orders (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
status text NOT NULL DEFAULT 'NEW'
CHECK (status IN ('NEW','PAID','SHIPPED','DELIVERED','CANCELLED')),
version bigint NOT NULL DEFAULT 0,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
Пара слов про типы, потому что здесь легко сделать по привычке и потом жалеть. text вместо varchar(20): в PostgreSQL они работают одинаково быстро, но у varchar(N) есть ограничение длины, которое однажды придётся менять, а у text нет. Сам список допустимых значений задаёт CHECK: его видно в схеме, он проверяется базой и правится обычной миграцией; тип Enum SQLAlchemy с native_enum=True создал бы тип PostgreSQL, в который новое значение добавляется только отдельной командой вне транзакции. Время это timestamptz, а не timestamp: без зоны момент времени восстановить невозможно, и первый же переезд сервера это покажет.
Проблема возникает, когда два обработчика одновременно читают заказ в состоянии PAID и оба решают его обработать: один шлёт SHIP, другой CANCEL. Каждый по отдельности видит допустимый переход, но вместе они конфликтуют, и без защиты выиграет тот, кто записал последним, затерев чужой результат. Есть два стандартных приёма.
Оптимистичная блокировка. В таблице держат колонку version. При чтении запоминают версию, при записи обновляют строку с условием «версия та же, что была». Условие пишется руками, а признак гонки это число затронутых строк:
from sqlalchemy import update
async def save_transition(session: AsyncSession, order: Order, next_state: State) -> None:
result = await session.execute(
update(OrderRow)
.where(OrderRow.id == order.id, OrderRow.version == order.version)
.values(status=next_state, version=OrderRow.version + 1, updated_at=func.now())
)
if result.rowcount == 0:
raise ConcurrentUpdate(order.id)
Если между чтением и записью кто-то уже поменял строку, версия не совпадёт, обновление затронет ноль строк, и мы понимаем, что случилась гонка. Повтор за вас не сделает никто: его пишут руками, и не внутри той же транзакции, а целиком заново: открыть новую, перечитать заказ, заново проверить переход и записать. Обработчик на ConcurrentUpdate отвечает 409, а повторяет либо клиент, либо сценарий, если команда идемпотентна. У ORM-слоя SQLAlchemy есть встроенный вариант того же приёма — version_id_col в настройках модели, тогда StaleDataError при commit играет роль нулевого rowcount.
Пессимистичная блокировка (SELECT ... FOR UPDATE). Второй обработчик блокируется на чтении строки до тех пор, пока первый не завершит транзакцию. Строку читают под блокировкой, проверяют переход, записывают, коммитят, и только потом её увидит второй:
row = (await session.execute(select(OrderRow).where(OrderRow.id == order_id).with_for_update())).scalar_one()
У репозитория есть вариант чтения с with_for_update(), и команда, которая сейчас будет менять состояние, зовёт именно его, а запрос на показ обходится без блокировки. Подробный разбор в статье Command side в CQRS на Python.
Смысл обоих приёмов один: проверка перехода и его запись должны быть одной неделимой операцией, чтобы между «прочитали PAID» и «записали SHIPPED» никто не успел вклиниться. Оптимистичная блокировка дешевле, когда конфликты редки (просто повторяем при неудаче); пессимистичная надёжнее, когда за одну и ту же запись часто конкурируют. Без любой из них корректный на бумаге автомат в реальной работе будет иногда переходить не туда. И отдельно про цикл событий: asyncio не защищает от этой гонки ничуть — два обработчика ждут ответа базы одновременно, и переключение между ними происходит ровно между чтением и записью.
Действия при переходе: что делать с письмами и деньгами
Все варианты выше отвечают на вопрос «куда перейти» и молчат про то, что в реальном сервисе висит на переходе: списать деньги, отправить письмо, зарезервировать товар, опубликовать событие. Это половина работы, и здесь же живёт самая частая авария: состояние сменилось, а действие не выполнилось, или наоборот.
Первое правило: автомат не выполняет действия, он их объявляет. Метод перехода меняет состояние и возвращает список того, что нужно сделать. Сам он никуда не ходит: ни в сеть, ни в почту, ни к платёжному провайдеру — и поэтому остаётся обычным синхронным методом без await.
def pay(self, amount: Money) -> list[SideEffect]:
if self.status is not State.NEW:
raise IllegalTransition(self.status, Event.PAY)
if amount != self.total:
raise PartialPayment()
self.status = State.PAID
return [
ReserveStock(order_id=self.id, lines=self.lines),
NotifyCustomer(order_id=self.id, status=State.PAID),
PublishEvent(OrderPaid(order_id=self.id, amount=amount)),
]
Почему так, а не «отправить письмо прямо здесь»: автомат остаётся проверяемым без всякой инфраструктуры (тест вызывает pay и смотрит, какие действия вернулись), и появляется одно место, где решается, как эти действия выполнить.
Второе правило: что уходит наружу, идёт через таблицу исходящих сообщений. Действие, которое обращается к внешнему миру, нельзя выполнять внутри транзакции перехода: транзакция может откатиться после успешной отправки, и письмо уйдёт про заказ, который не оплачен. И нельзя после коммита: между коммитом и отправкой процесс может умереть, и письмо не уйдёт никогда. Рабочая схема одна: в той же транзакции, что и переход, записать задание в таблицу, а отправку сделать отдельным процессом.
async with session.begin():
order = await orders.for_update(session, command.order_id)
effects = order.pay(command.amount)
await orders.save(session, order)
await history.record(session, order.id, State.NEW, State.PAID, "PAYMENT_RECEIVED", command.actor)
await outbox.save_all(session, effects)
Одна транзакция, три записи, ни одного обращения наружу. Дальше отправщик забирает задания из таблицы (FOR UPDATE SKIP LOCKED) и выполняет их, повторяя при сбоях, и вот здесь нужно третье правило.
Третье правило: действия должны переносить повтор. Отправщик не знает, дошло ли письмо, если оборвалась сеть, и попробует снова. Значит, у каждого задания есть ключ, а получатель обязан отличить повтор: письмо с тем же ключом не отправляется дважды, резерв с тем же ключом не создаётся дважды. Механика в статье про пакетную обработку и идемпотентность, и без неё «надёжная доставка» превращается в «двойное списание».
Что делать с действиями, которые обязаны выполниться синхронно. Бывает: оплата не может «случиться потом», пользователь ждёт ответа платёжного провайдера. Тогда порядок обратный: сначала внешний вызов, потом переход. Если провайдер ответил успехом, а наша транзакция упала, состояние не сменилось, деньги списаны, и это расхождение закрывают сверкой плюс ключом идемпотентности на стороне провайдера. Правило простое: внешний вызов либо до транзакции, либо после неё через таблицу исходящих, но никогда внутри.
Порядок внутри перехода, который стоит запомнить как последовательность: проверить переход, проверить условия, сменить состояние, записать историю, записать задания, закоммитить. Всё внешнее за пределами этого списка.
Как тестировать автомат
У автомата есть свойство, которого нет у обычного кода: его правила конечны и перечислимы, а значит, тест может проверить их все. В Python это parametrize по декартову произведению состояний и событий, и он же лучший аргумент в пользу таблицы переходов.
import pytest
from itertools import product
EXPECTED = {
(State.NEW, Event.PAY): State.PAID,
(State.NEW, Event.CANCEL): State.CANCELLED,
(State.PAID, Event.SHIP): State.SHIPPED,
(State.PAID, Event.CANCEL): State.CANCELLED,
(State.SHIPPED, Event.DELIVER): State.DELIVERED,
}
@pytest.mark.parametrize(("current", "event"), list(product(State, Event)))
def test_every_pair_behaves_as_specified(current: State, event: Event):
if (current, event) in EXPECTED:
assert next_state(current, event) is EXPECTED[(current, event)]
else:
with pytest.raises(IllegalTransition):
next_state(current, event)
Ценность здесь не в том, что проверены разрешённые переходы, а в том, что проверены запрещённые: пять состояний и четыре события дают 20 пар, из которых разрешено пять, и тест утверждает, что остальные пятнадцать отвечают именно IllegalTransition, а не KeyError и не пустым состоянием. Именно эти пятнадцать и есть содержание автомата, и обычными тестами их никто не покрывает.
Проверка достижимости. Второй тест ловит ошибки в правилах, а не в коде: обойти граф переходов от начального состояния и убедиться, что каждое состояние достижимо, а из каждого конечного нет выхода. Обход это десять строк с очередью по таблице:
from collections import deque
def reachable_from(start: State) -> set[State]:
seen = {start}
queue = deque([start])
while queue:
current = queue.popleft()
for (source, _event), target in TRANSITIONS.items():
if source is current and target not in seen:
seen.add(target)
queue.append(target)
return seen
def outgoing(state: State) -> list[Event]:
return [event for (source, event) in TRANSITIONS if source is state]
def test_all_states_reachable_and_terminals_final():
assert reachable_from(State.NEW) == set(State)
for terminal in (State.DELIVERED, State.CANCELLED):
assert outgoing(terminal) == [], f"из конечного {terminal} есть выход"
Недостижимое состояние это мёртвая константа (обычно остаток от старого процесса), а конечное состояние с выходом почти всегда ошибка в правилах, которую в проде обнаружит поддержка.
Тесты на условия и действия. Отдельно: guard проверяется парой тестов «условие выполнено, переход прошёл» и «не выполнено, отказ с нужным исключением» через pytest.raises. Действия тем, что метод перехода вернул нужный список, а не тем, что письмо ушло: это как раз та польза от «автомат объявляет действия», о которой раздел выше.
И тест на гонку, если состояние живёт в базе: сто корутин через asyncio.gather пытаются перевести один объект на настоящей PostgreSQL из контейнера, одна выигрывает, остальные получают понятную ConcurrentUpdate. Этот тест нельзя написать на SQLite и нельзя — на моке сессии: он проверяет ровно то, что делает база.
Состояние приходит извне: опоздания и повторы
Отдельный класс задач, который ломает всё описанное выше: состояние меняет не ваш код, а внешняя система: перевозчик, платёжный провайдер, партнёр. Приходит вебхук от них («посылка вручена»), и переход надо выполнить по чужим данным. Здесь три особенности, и все три встречаются в первый же месяц.
Сообщения приходят не по порядку. Перевозчик отправил «в пути» и «доставлено», а до вас дошло сначала второе. Если применять как пришло, заказ окажется «в пути» после того, как был доставлен. Правильный ответ: отбрасывать переход «назад», а не пытаться его выполнить. Для этого у состояний должен быть порядок или отметка времени события:
def apply_carrier_status(self, incoming: str, occurred_at: datetime) -> None:
if self.last_carrier_event_at and occurred_at < self.last_carrier_event_at:
log.info("опоздавшее событие перевозчика отброшено", status=incoming, at=occurred_at.isoformat())
return
target = CARRIER_TO_ORDER.get(incoming)
if target is None:
log.warning("неизвестный статус перевозчика", status=incoming)
return
if (self.status, target) not in ALLOWED_CARRIER_MOVES:
log.info("обратный переход отброшен", current=self.status, target=target)
return
self.status = target
self.last_carrier_event_at = occurred_at
Три вещи в этом коде важнее остального. Опоздавшее событие не ошибка, и на него нельзя отвечать ошибкой: внешняя система воспримет её как «не доставлено» и будет повторять бесконечно. Отвечать надо успехом и ничего не делать. Сравнение по времени события, а не по времени получения: время берут из полезной нагрузки, и если внешняя система его не присылает, это первое, о чём стоит попросить; две даты обязаны быть с часовым поясом, иначе Python откажется их сравнивать. И неизвестный чужой статус (перевозчик добавил новый) записывается в журнал, не меняет состояние и не роняет обработку: иначе один новый статус у партнёра остановит все уведомления.
Повторы приходят всегда. Внешняя система, не получившая подтверждения, отправит то же самое снова, иногда десятки раз. Значит, обработчик обязан быть идемпотентным: тот же переход, применённый дважды, не должен порождать двойных действий. Практически это отметка обработанных сообщений по идентификатору от внешней системы, в той же транзакции, что и переход, через INSERT ... ON CONFLICT DO NOTHING и проверку rowcount.
Чужие состояния не совпадают с вашими. У перевозчика двадцать статусов, у вас четыре. Отображение чужого на своё это решение, и его записывают явно, отдельным словарём соответствия; заодно в нём видно, какие чужие статусы вы игнорируете.
Новое состояние в работающей системе
Добавить значение в StrEnum это одна строка. Проблема в том, что в базе уже лежат миллионы записей со старыми значениями, а рядом работают копии сервиса с предыдущей версией кода.
Добавление состояния это совместимое изменение, если делать в правильном порядке. Сначала база, потом код, потом использование: миграция расширяет CHECK новым значением; выкатывается код, который умеет читать новое состояние, но ещё не переводит в него; и только потом включается переход, новым выкатом или флагом. Почему порядок именно такой: при выкате по одной копии старая версия обязана понимать то, что пишет новая. Пропустили второй шаг, и старая копия встретит незнакомое значение; State("RETURNED") упадёт с ValueError при чтении строки из базы, и заказ в новом состоянии станет недоступен ни для показа, ни для обработки. Это то же правило совместимости, что для любой схемы данных: сначала научиться читать, потом начать писать.
Удаление состояния это обратный порядок и гораздо дольше. Перестать переводить, дождаться, пока записей в этом состоянии не останется (а они могут висеть месяцами), убрать из кода, убрать из ограничения. Последний шаг часто не делают вовсе, и это нормально: лишнее значение в проверке никому не мешает, а код, который его не знает, мешает.
Чего нельзя делать ни в каком порядке: переименовывать значение (это одновременно удаление и добавление, и при выкате по одной копии часть записей окажется с одним именем, часть с другим) и менять смысл существующего значения (старые записи начнут врать). Тест, который это страхует: прогнать тесты предыдущей версии кода против базы с новой миграцией. Зелёные, значит изменение совместимо; красные, значит при выкате по одной копии будет авария.
Что выбрать
Единственно правильного способа нет, выбор зависит от размера автомата и от того, сколько логики висит на переходах.
StrEnumиmatchдля маленького автомата: несколько состояний, простые переходы, никаких зависимостей;assert_neverиmypyвместо проверки интерпретатором.- Таблица переходов для среднего: переходов много, но они сводятся к «сменить состояние»; правила хочется видеть списком, проверять перебором и менять, не трогая код.
- Состояние как класс когда в каждом состоянии много собственного поведения и его важно держать вместе, рядом с состоянием.
- Библиотека (
transitions,python-statemachine) для больших автоматов с условиями, действиями на входе и выходе, подсостояниями и схемой, которую генерируют; для мелких избыточна.
| Признак | StrEnum и match | Таблица переходов | Состояние как класс | Библиотека |
|---|---|---|---|---|
| Число состояний | до 5 | 5–20 | 3–10 | больше 10 |
| Число переходов | до 10 | десятки | десятки | десятки и больше |
| Условия перехода | case ... if | полем записи | внутри состояния | conditions у перехода |
| Много поведения в состоянии | нет | нет | да | да |
| Правила нужны как данные (менять, рисовать, проверять) | нет | да | нет | да, схема из библиотеки |
| Подсостояния | нет | нет | с трудом | да |
| Хранение состояния между запросами | своими руками | своими руками | своими руками | своими руками |
| Сторонняя зависимость | нет | нет | нет | да |
| Цена входа | минутная | часовая | часовая | день |
Как читать эту таблицу: два признака решают почти всегда. Если в состояниях много собственного поведения, состояние как класс, независимо от их числа. Если правила должны быть данными (их правят не только разработчики, по ним строят схему, их проверяют перебором), таблица переходов. Всё остальное при пяти состояниях это StrEnum и match, и это честный выбор, а не упрощение. Библиотеку берут по одному настоящему признаку: нужны подсостояния или схема, которую не хочется рисовать руками. И независимо от выбора не забудьте про хранение и гонки: колонка status, а поверх неё version или FOR UPDATE.
Глубже: переход, который ждёт внешнего ответарасширенное
Самый коварный переход в автомате заказа не внутренний, а тот, что ждёт чужого решения: отправили запрос в платёжный сервис и ждём подтверждения минуту, час, сутки. Пока ответа нет, заказ не в PAID и не в NEW, и если такого состояния в автомате нет, его придумают данные.
Состояние ожидания объявляют явно: AWAITING_PAYMENT со своим сроком. Переход в него выполняют одной транзакцией вместе с записью задания в outbox («запросить оплату», с ключом идемпотентности) и полем deadline_at; сам HTTP-вызов из транзакции не делают, его выполнит задача очереди.
Ответ приходит двумя путями, и оба обязаны быть идемпотентными: вебхук провайдера (POST /payments/callback) и опрос статуса по расписанию для тех заказов, у которых вебхук не пришёл. Обработчик ответа это обычное событие автомата payment_confirmed или payment_declined; повтор вебхука для заказа, который уже в PAID, автомат отбрасывает без ошибки по правилам раздела про опоздания.
Истечение срока это третье событие, payment_timeout, которое порождает фоновая задача по deadline_at <= now() и которое переводит заказ в CANCELLED с причиной; если после этого провайдер всё же подтвердит оплату, это уже не переход автомата, а отдельный сценарий возврата денег с записью в журнал.
Так состояние ожидания получает те же свойства, что остальные: его видно в таблице, у него есть срок, его переходы перечислены, и тест parametrize обходит их вместе со всеми.
Коротко
- Автомат на бумаге ничего не гарантирует, всё решает то, как перенести его в код, чтобы недопустимые переходы стали невозможны.
StrEnumиmatchэто простейший вариант; полнотуmatchинтерпретатор не проверяет, это делаютassert_neverсmypyи тесты. - Таблица переходов это словарь с ключом-кортежем «состояние плюс событие»: весь автомат виден списком, правила меняются без кода, а перечисление само отдаёт все значения для перебора в тестах.
- Состояние как класс делает каждое состояние объектом со своим поведением; базовый класс, который всё отвергает, заменяет
default-методы интерфейса. transitionsдаёт условия, действия, подсостояния иAsyncMachine; запрещённое событие этоMachineError, непройденное условие это тихийFalse, аauto_transitions=Falseобязателен, иначе появятся методыto_<состояние>в обход правил.- Запрещённый переход это свой класс исключения, который единый обработчик превращает в 409; отказ условия это своё исключение с кодом.
- Состояние хранят в колонке
statusсCHECK; гонку ловят оптимистично (update().where(version == n), нулевойrowcountзначит конфликт, повтор новой транзакцией) или пессимистично (with_for_update()в транзакции команды);asyncioот неё не защищает. - Автомат не выполняет действия, а объявляет их: состояние, история и задания пишутся одной транзакцией, наружу уходит отправщик из таблицы исходящих, а действия обязаны переносить повтор.
- Автомат проверяется
parametrizeпо всем парам «состояние на событие» (ценны именно запрещённые), обходом достижимости, тестами условий и тестом на гонку черезgatherна настоящей базе. - Состояние извне: время события в данных, опоздавшие и обратные переходы отбрасывать без ошибки, идемпотентность по идентификатору сообщения, явный словарь соответствия и безопасное поведение при неизвестном чужом статусе.
- Новое состояние вводят в три шага: разрешить в базе, научить код читать, включить переход; иначе старая копия упадёт на
State("...")сValueError; удаление в обратном порядке и дольше; переименование и смена смысла запрещены.
Что почитать дальше
- Что такое конечный автомат — состояния, переходы и когда автомат нужен.
- BPM-движки и оркестрация — долгие процессы через сервисы, сага и хореография.
- Command side в CQRS на Python — чтение агрегата под блокировкой и команда как единица работы.
- Пакетная обработка и идемпотентность на Python — отправщик из таблицы исходящих и повторы без двойных действий.