Реализация конечного автомата для автоматизации процессов

Каждый уважающий себя техлид \ архитектор ПО \ руководитель разработки, должен написать в своей жизни хотя бы одну CRM
_{народная мудрость}

Всем привет! Меня зовут Михаил я техлид в компании ДомКлик. Сегодня я хочу поговорить про автоматизацию бизнес-процессов. У нас есть объекты, граф состояний \ набор статусов и в каждый момент времени объект находится в одном из возможных состояний. Это позволяет описать workflow или конечный автомат для рассматриваемого процесса и строить сервис автоматизации на этой абстракции.

В основе многих сервисов, которые мы используем в повседневной жизни, лежат процессы которые можно описать с помощью этих абстракций - это покупки в интернете, еда, такси, CRM, ERP, ...

Рассмотрим для примера, процесс оформления и доставки некоторого заказа.

Описание объекта

class Order:
  status
  responsible
  price
  payed

Статусная модель

WF_STATUSES = (NEW, ORDERED, RESERVED, CANCELLED,
               RETURNED, PAYED, SHIPPED, DELIVERED, COMPLETED,)

Borland Developer Studio, ODBC, все как положено, на дворе 2006 год.. именно тогда мне довелось поработать над первой в своей жизни CRM. Человеческая психика так устроена, что все плохое вытесняет и замещает, поэтому, знакомясь с очередной реализацией workflow или создавая проект с нуля, я старался найти ту самую серебряную пулю - общий подход, который будет наиболее удобен в использовании, интуитивно понятен и эффективен. За время своей работы у меня скопилась хорошая подборка решений из серии, как не надо делать, но удалось выработать и кое-что полезное.

Как не надо делать

def set_ordered(self, request):
  ...
  object = self.get_object()
  object.status = ORDERED
  object.save()
  ...

Наиболее неудачное решение, это размазывание, по коду программы, всей логики движения объекта по workflow. Мы изменяем состояние объекта в API-handlers, сигналах, триггерах, методах класса, везде где только можно. При таком подходе нет общего понимания процесса, вносить изменения крайне сложно.

class Order:
  ...
  def set_ordered(self):
    pass

  def set_reserved(self):
    pass

Достаточно удобно, реализовать workflow через класс, где для перехода в каждый статус будет своя функция. Мне этот вариант не нравится тем, что, часто есть общий подход к смене статуса - какие-то общие действия, которые должны быть в каждой функции, например:

• валидация состояния объекта

• назначение ответственного

• логирование смены статуса

Кроме того, поддержка и развитие такого workflow в нашем изменчивом мире, так же создаст проблемы. Например включение новых шагов в процесс, потребует заново оценить все места в проекте, где происходит смена состояния объекта. Кроме того, логика workflow утекает между пальцев и нам снова не понятно, какой статус идет за каким. Чтобы разобраться, потребуется глубокое понимание проекта.

К чему мы пришли

Итак, у нас есть объект - заказ в интернет магазине и статусная модель, описывающая процесс покупки товара. Вначале мы определили варианты для манипуляций с объектом, мы можем

• двигать объект по позитивному сценарию DIR_NEXT,

• двигать в альтернативные ветки DIR_FAIL, DIR_WAIT, DIR_RETURN,

• отменять обработку объекта DIR_CANCEL,

• завершить обработку объекта DIR_COMPLETE.

DIRECTIONS = (DIR_NEXT, DIR_FAIL, DIR_RETURN, DIR_WAIT, DIR_CANCEL, DIR_COMPLETE,)

Далее, чтобы получить наглядное представление о процессе, его нужно описать. Мы выбрали JSON-схему, это хорошая отправная точка для построения визуального представления процесса. Кроме того схема процесса всегда есть в коде под рукой, чтобы вспомнить что за чем идет.

Workflow процесса

ORDER_WORKFLOW = {
    NEW: {
        DIR_NEXT: ORDERED,
        'responsible': AUTHOR,
    },
    ORDERED: {
        DIR_NEXT: RESERVED,
        DIR_RETURN: RETURNED,
        DIR_CANCEL: CANCELLED,
        'responsible': MANAGER,
    },
    RESERVED: {
        DIR_NEXT: PAYED,
        DIR_CANCEL: CANCELLED,
    },
    PAYED: {
        DIR_NEXT: SHIPPED,
        'notify_manager': True,
        'responsible': STOREKEEPER,
    },
    SHIPPED: {
        DIR_NEXT: DELIVERED,
        'notify_client': True,
        'responsible': DRIVER,
    },
    DELIVERED: {
      DIR_NEXT: COMPLETED,
      DIR_CANCEL: CANCELLED,
    },
    COMPLETED: {
        'notify_manager': True,
        'finished': True,
    },
                RETURNED: {
      DIR_NEXT: ORDERED,
      DIR_CANCEL: CANCELLED,
    },
    CANCELLED: {
        'notify_manager': True,
        'finished': True,
    },
}

Диаграмма процесса

Собственно реализацию workflow делаем через класс. При инициализации связываем объект с instance Workflow и все манипуляции со сменой состояния \ статуса объекта делаем через этот класс. Интерфейс работы с workflow имеет следующий вид:

• у нас есть метод get_state для получения состояния объекта, которое включает в себя доступный набор переходов и необходимую информацию для отображения объекта,

• есть метод step, который обеспечивает смену состояния объекта с учетом доступных переходов.

Реализация workflow

class Workflow:

    def __init__(self, order, workflow):
        self.order = order
        self.workflow = workflow

    def get_state(self):
        """
        получить состояние заявки в workflow
          - возможные переходы
          - finished true | false
          - какая-то дополнительная информация,
          описывающая состояние заявки в рамках процесса
        """
        order = self.order
        stage = self.workflow[task.status]

        state = {
            'status': order.status,  # actual status
            'finished': stage.get('finished', False),
            'directions': tuple(),
        }
                                for direction in DIRECTIONS:
            dir_status = stage.get(direction)
            if dir_status:
                state['directions'] += (direction, dir_status),

        return state

    def _step_assert(self, task, direction, user):
        assert task.status in self.workflow, 'wrong workflow status'
        assert direction in DIRECTIONS, 'wrong direction'

    def get_direction(self, stage, direction):
        return stage.get(direction)

    def step(self, direction=DIR_NEXT, **kwargs):
        """
        перемещение заявки в следующий возможный статус в рамках workflow
        :param direction:
        :return: moved - true | false, int_code, text_reason
        """
        order = self.order
        user = get_current_user()

        self._step_assert(order, direction, user)

        stage = self.workflow[order.status]
        if stage.get('finished'):
            return False, 2, 'Обработка заявки завершена'

        next_status = self.get_direction(stage, direction)

        if next_status:
            next_stage = self.workflow[next_status]
            notify_manager = next_stage.get('notify_manager')
            notify_client = next_stage.get('notify_client')

            if notify_manager:
                self.notify_manager(order)

            if notify_client:
                self.notify_client(order)

            order.set_status(next_status)

            if 'responsible' in next_stage:
                order.responsible = self.set_responsible(
                    order, next_stage['responsible']
                )

            order.save()
            return True, 0, 'Переход произведен'

        return False, 1, 'Переход не был произведен'

    @staticmethod
    def notify_manager(order):
        raise NotImplemented

    @staticmethod
    def notify_client(order):
        raise NotImplemented

    @staticmethod
    def set_responsible(order, role):
        raise NotImplemented


class OrderWorkflow(Workflow):
    """
    Order Workflow
    """

    def __init__(self, order):
        super().__init__(order, ORDER_WORKFLOW)

    @staticmethod
    def set_responsible(order, role):
        return order.set_responsible(role=role)

Данная реализация собирает всю логику процесса, внутри класса описывающего workflow. В нашем случае, у нас было несколько процессов, которые имеют разные статусные модели, но при этом мы смогли использовать общий движок для изменения состояний сущностей разного типа. Я считаю что, для этой реализации расширение статусной модели и внесение новой логики может проходить относительно безболезненно.

Добавление нового состояния сведется к обновлению собственно JSON-схемы. Если мы хотим добавить новую логику при смене состояний, это удобно сделать через навешивание новых флагов \ признаков в описание каждого состояния в схеме.

    RESERVED: {
        DIR_NEXT: PAYED,
        DIR_FAIL: CANCELED,
        'log_status': True,
    },

Затем нужно прописать логику, те действия, которые мы хотим выполнить, при наличии этого флага в очередном состоянии объекта.

               @staticmethod
    def log_status(order):
        pass

Заключение

Я описал наш подход к реализации workflow, который обеспечивает, по моему мнению

• наглядное описание процесса в коде

• удобство расширения логики обработки переходов по состояниям

• удобство изменения схемы процесса и расширения описания процесса

Предложенный подход имеет свои сильные и слабые стороны. Возможно, он не подойдет для реализации любого процесса, но послужит хорошей отправной точкой для ваших проектов по автоматизации процессов. Спасибо что дочитали! Всем добра!

Let's block ads! (Why?)