Про конечные автоматы (finite state machine, fsm) много кто слышал, но используют их явно в реальных проектах редко. Чаще встречаются конструкции, которые поведением напоминают КА, но ими не являются.
Почему же автоматы обходят стороной и/или изобретают велосипеды, превращая код в спагетти?
По-моему, тут дело в стереотипе: мол, автоматы — это что-то сложное из теоретической математики и к реальной жизни не относится. А применять их можно только в лексических анализаторах или еще чем-нибудь специфичном.

На самом деле, область применения КА куда шире и понятнее. Давайте разберем на примере автоматизации процессов в любимом кровавом enterprise.

Начнем с определений

Дисклеймер №1: Тема автоматного программирования и машин состояний очень широкая. А у нас нет цели объять необъятное. Поэтому часть информации останется за кадром.

Формально, конечный автомат — это математическая модель, которая описывает абстрактную систему с конечным числом допустимых состояний. В любой момент времени система может находиться только в одном из них.

Сегодня поговорим только об одном виде КА — детерминированных автоматах. Их можно описать еще проще: это набор состояний, связанных переходами. Такой КА легко представить в виде графа, где вершинами будут состояния, а рёбрами — переходы.

Если вас не остановила теория КА, значит, вы сможете оценить и его практическую реализацию. Переходим к практике.

Дисклеймер №2. В этой статье мы не будем в сотый раз писать имплементацию КА. Варианты реализации fsm на любых языках программирования можно найти в том числе на Хабре. Мы же применим готовое решение — open source библиотеку transitions для python.

Выбираем объект автоматизации

Все, кто использует Agile, наверняка знакомы с инструментом Atlassian Jira (далее просто jira). Мы будем автоматизировать модель состояний и переходов задачи в jira.

Дефолтный workflow — жизненный цикл задачи — в jira можно представить графом:

Разложим граф с диаграммы на артефакты fsm:

• Возможные состояния автомата (states): open, in progress, resolved, reopened, closed
• Начальное состояние автомата (initial state): open
• Переходы (transitions): start progress, stop progress, resolve, close, resolve & close, reopen

Теперь, когда все артефакты fsm формализованы, приступим к реализации.

Инициализируем fsm

В библиотеке transitions есть специальный класс Machine. Он и инкапсулирует всю логику fsm. Конструктор класса принимает следующие аргументы:

1. объект автоматизации, который мы хотим наделить артефактами fsm;
2. список возможных состояний (states);
3. модель переходов (transitions);
4. начальное состояние объекта (initial state);
5. прочие параметры (опционально).

Создадим класс JiraTask. Пока нам достаточно пустышки:

class JiraTask:
    pass

Теперь создадим экземпляр этого класса и привяжем его к машине:

task = JiraTask()

# init jira task states and transitions
states = ['open', 'closed', 'resolved', 'inprogress', 'reopened']
transitions = [
    {'trigger': 'start_progress', 'source': 'open', 'dest': 'inprogress'},
    {'trigger': 'resolve_and_close', 'source': 'open', 'dest': 'closed'},
    {'trigger': 'stop_progress', 'source': 'inprogress', 'dest': 'open'},
    {'trigger': 'resolve', 'source': 'inprogress', 'dest': 'resolved'},
    {'trigger': 'resolve_and_close', 'source': 'open', 'dest': 'closed'},
    {'trigger': 'close', 'source': 'resolved', 'dest': 'closed'},
    {'trigger': 'reopen', 'source': 'closed', 'dest': 'reopened'},
    {'trigger': 'resolve', 'source': 'reopened', 'dest': 'resolved'},
    {'trigger': 'resolve_and_close', 'source': 'reopened', 'dest': 'closed'},
    {'trigger': 'start_progress', 'source': 'reopened', 'dest': 'inprogress'}
]

# Initialize fsm and bind it
machine = Machine(task, states=states, transitions=transitions, initial='open')

Давайте разберем, что мы сделали.

Переменная states содержит список всех возможных состояний.
Самая интересная штука — transitions. По сути, это список всех возможных переходов в нашем автомате, где:

• trigger – это некое действие, которое может привести к смене состояния объекта,
• source – исходное состояние объекта,
• dest – целевое состояние объекта.

Вот, собственно, и вся инициализация fsm.

Тестируем fsm

Напишем два сценария, чтобы проверить корректность работы fsm:

• позитивный (sunny day scenario),
• негативный (rainy day scenario).

Sunny day scenario реализует логику штатного движения задачи по workflow. Возьмем самый простой линейный путь задачи:

try:
    task.start_progress()
    print(task.state)

    task.resolve()
    print(task.state)

    task.close()
    print(task.state)

except MachineError as error:
    print(error)

В консоли увидим следующий вывод:
inprogress
resolved
closed

Но постойте, откуда у экземпляра JiraTask взялись методы start_progress(), resolve(), close() ? Мы же их явно не описывали в классе. А они появились автоматически при связывании (binding) машины состояний с экземпляром нашего класса.

Rainy day scenario реализует негативный кейс, нарушающий штатную цепочку движения задачи по workflow. Попытаемся закрыть задачу, пропустив шаг resolve.

try:
    task.start_progress()
    print(task.state)

    task.close() # exception - there is no such transition in model
    print(task.state)

    task.resolve()
    print(task.state)

    task.close()
    print(task.state)

except MachineError as error:
    print(error)

В консоли увидим следующий вывод:

inprogress
"Can't trigger event close from state inprogress!"

То есть fsm бросил exception (MachineError) при попытке выполнить недопустимый переход.

Добавляем условия

Все работает, как задумано, но пока не выглядит решением для промышленной разработки. Не хватает условий (conditions) для осуществления переходов.

В реальной jira для смены статуса задачи с А на Б должен выполниться целый ряд требований, даже если данный переход условно допустим по workflow. Например, пользователь должен обладать определенными правами и/или заполнить ряд обязательных полей в задаче. Давайте попробуем это имплементировать.

Реализуем упрощенную логику условий перехода:

• создадим сущность «Пользователь» в виде словаря с атрибутами: идентификатор, имя, роль;
• создадим реестр пользователей, зарегистрированных в jira;
• разрешим/запретим ряд переходов в зависимости от роли пользователя, ответственного (responsible) по задаче.

Поехали!

Для создания реестра пользователей применим паттерн singleton, так как нам нужен только один экземпляр этой сущности в приложении:

class MetaSingleton(type):
    _instances = {}
    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super(MetaSingleton, cls).__call__(*args, **kwargs)
        return cls._instances[cls]

Ну а теперь реализуем реестр (UserStorage) и все сопутствующие артефакты:

class UserRoles(Enum):
    tester = 1
    developer = 2

class UserStorage(metaclass=MetaSingleton):
def __init__(self):
    self._users = [{'id': 'v', 'name': 'Вася', 'role': UserRoles.developer},
                    {'id': 'p', 'name': 'Петя', 'role': UserRoles.developer},
                    {'id': 'm', 'name': 'Миша', 'role': UserRoles.developer},
                    {'id': 's', 'name': 'Саша', 'role': UserRoles.tester},
                    {'id': 'k', 'name': 'Катя', 'role': UserRoles.tester}
    ]

def get_user_by_id(self, id):
    return next((item for item in self._users if item['id'] == id), None)

def is_tester(self, id):
    user = self.get_user_by_id(id)
    if user:
        return user['role'] == UserRoles.tester
    return False

def is_developer(self, id):
    user = self.get_user_by_id(id)
    if user:
        return user['role'] == UserRoles.developer
    return False

Сразу добавляем в реестр 5 пользователей: 3 разработчика (developer) и 2 тестировщика (tester).

Реализуем несколько helper методов для проверки роли пользователя по идентификатору:

• is_developer()
• is_tester()

Усложним наш класс JiraTask — привяжем его по атрибуту user_id к ответственному исполнителю из реестра UserStorage. В качестве «жертвы» возьмем разработчика Мишу. Потому что его не жалко. Шутка :)

class JiraTask:
    def __init__(self):
        self._user_id = 'm'

    # checks if the user is tester
    def is_tester(self):
        return UserStorage().is_tester(self._user_id)

    # checks if the user is developer
    def is_developer(self):
        return UserStorage().is_developer(self._user_id)

Итак, что мы сделали. Добавили user_id в класс JiraTask и реализовали proxy-вызовы методов is_tester()/is_developer() для обмена с реестром пользователей.

Пора добавлять в fsm наши условия.

Разрешим закрывать задачи только пользователям с ролью «тестировщик» (tester). С учетом новых вводных инициализация transitions будет выглядеть так:

transitions = [
    {'trigger': 'start_progress', 'source': 'open', 'dest': 'inprogress'},
    {'trigger': 'resolve_and_close', 'source': 'open', 'dest': 'closed', 'conditions': ['is_tester']},
    {'trigger': 'stop_progress', 'source': 'inprogress', 'dest': 'open'},
    {'trigger': 'resolve', 'source': 'inprogress', 'dest': 'resolved'},
    {'trigger': 'resolve_and_close', 'source': 'open', 'dest': 'closed', 'conditions': ['is_tester']},
    {'trigger': 'close', 'source': 'resolved', 'dest': 'closed', 'conditions': ['is_tester']},
    {'trigger': 'reopen', 'source': 'closed', 'dest': 'reopened'},
    {'trigger': 'resolve', 'source': 'reopened', 'dest': 'resolved'},
    {'trigger': 'resolve_and_close', 'source': 'reopened', 'dest': 'closed', 'conditions': ['is_tester']},
    {'trigger': 'start_progress', 'source': 'reopened', 'dest': 'inprogress'}
]

Атрибут conditions указывает на список всех условий, объединенных по «И». В нашем случае это методы класса JiraTask, к экземпляру которого привязана модель переходов fsm.

Тестируем снова

Изменим логику обработки результатов методов в прежнем сценарии. Теперь надо учитывать результат выполнения условий — true/false. Ранее все методы, которые инициировали смену состояния, возвращали true либо бросали MachineError, если переход отсутствует в модели.
После доработок достаточно прогнать только «позитивный» сценарий. Хотя он уже не такой позитивный, ведь Мише, увы, теперь нельзя закрывать задачу из-за ролевых игр ограничений:

try:
    if not task.start_progress():
        print('conditions fail')
    else:
        print(task.state)

    if not task.resolve():
        print('conditions fail')
    else:
        print(task.state)

    if not task.close():
        print('conditions fail')
    else:
        print(task.state)

except MachineError as error:
    print(error)

Что же мы видим в консоли?
inprogress
resolved
conditions fail

Как и ожидалось, попытка несанкционированного закрытия задачи провалилась. Значит, все работает автоматом!

Теперь добавим логирование. Оно в библиотеке transitions идет из коробки. Инициализируем logger:

import logging
logging.basicConfig(level=logging.DEBUG)
# Set transitions' log level to INFO; DEBUG messages will be omitted
logging.getLogger('transitions').setLevel(logging.DEBUG)

Запустим приложение с включенным логированием:

DEBUG:transitions.core:Executed machine preparation callbacks before conditions.
DEBUG:transitions.core:Initiating transition from state open to state inprogress…
DEBUG:transitions.core:Executed callbacks before conditions.
DEBUG:transitions.core:Executed callback before transition.
DEBUG:transitions.core:Exiting state open. Processing callbacks…
INFO:transitions.core:Finished processing state open exit callbacks.
DEBUG:transitions.core:Entering state inprogress. Processing callbacks…
INFO:transitions.core:Finished processing state inprogress enter callbacks.
DEBUG:transitions.core:Executed callback after transition.
DEBUG:transitions.core:Executed machine finalize callbacks
inprogress
DEBUG:transitions.core:Executed machine preparation callbacks before conditions.
DEBUG:transitions.core:Initiating transition from state inprogress to state resolved…
DEBUG:transitions.core:Executed callbacks before conditions.
DEBUG:transitions.core:Executed callback before transition.
DEBUG:transitions.core:Exiting state inprogress. Processing callbacks…
INFO:transitions.core:Finished processing state inprogress exit callbacks.
DEBUG:transitions.core:Entering state resolved. Processing callbacks…
INFO:transitions.core:Finished processing state resolved enter callbacks.
DEBUG:transitions.core:Executed callback after transition.
DEBUG:transitions.core:Executed machine finalize callbacks
resolved
DEBUG:transitions.core:Executed machine preparation callbacks before conditions.
DEBUG:transitions.core:Initiating transition from state resolved to state closed…
DEBUG:transitions.core:Executed callbacks before conditions.
DEBUG:transitions.core:Transition condition failed: is_tester() does not return True. Transition halted.
DEBUG:transitions.core:Executed machine finalize callbacks
conditions fail

Теперь наша машина детально логирует все события. Очень удобно для трассировки в режиме промышленной эксплуатации.

Визуализируем автомат

И вишенка на торте — коробочная визуализация запрограммированного КА.
Внутри библиотеки transitions для генерации изображений используется пакет GraphViz.
Импортируем его:

from transitions.extensions import GraphMachine

Добавим пару строчек кода: сменим тип сущности fsm на GraphMachine и экспортируем граф в картинку.

machine = GraphMachine(task, states=states, transitions=transitions, initial='open')

machine.get_graph().draw('jira_task_state_diagram.png', prog='dot')

Вуаля, библиотека сформировала нам диаграмму — граф КА в формате png. Красота!

Мы рассмотрели далеко не все возможности transitions. Но я и не планировал делать из статьи документацию к библиотеке объемом с «Войну и мир». Если тема будет интересной читателям, то обязательно напишу продолжение.

Подводим итоги

1. Мы спроектировали модель реального бизнес-процесса на примере jira workflow;
2. Успешно реализовали логику переходов;
3. Готовый прототип можно использовать как референс при разработке промышленных решений.

Надеюсь, мне удалось показать, что fsm не так страшны и совсем незаслуженно обделены вниманием разработчиков. Автоматы можно и нужно применять в задачах автоматизации бизнес-процессов. Они подходят для систем любого уровня сложности, в том числе — в рамках промышленной разработки. В следующих статьях обсудим, в каких еще сферах и системах они будут полезны.