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

Автор Сообщение
news_bot ®

Стаж: 6 лет 9 месяцев
Сообщений: 27286

Создавать темы news_bot ® написал(а)
25-Мар-2021 14:31
Каждый уважающий себя техлид \ архитектор ПО \ руководитель разработки, должен написать в своей жизни хотя бы одну CRM
народная мудрость
Всем привет! Меня зовут Михаил я техлид в компании ДомКлик. Сегодня я хочу поговорить про автоматизацию бизнес-процессов. У нас есть объекты, граф состояний \ набор статусов и в каждый момент времени объект находится в одном из возможных состояний. Это позволяет описать workflow или конечный автомат для рассматриваемого процесса и строить сервис автоматизации на этой абстракции.В основе многих сервисов, которые мы используем в повседневной жизни, лежат процессы которые можно описать с помощью этих абстракций - это покупки в интернете, еда, такси, CRM, ERP, ...Рассмотрим для примера, процесс оформления и доставки некоторого заказа.Описание объекта
class Order:
  status
  responsible
  price
  paid
I
WF_STATUSES = (NEW, ORDERED, RESERVED, CANCELLED,
               RETURNED, PAID, 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: PAID,
        DIR_CANCEL: CANCELLED,
    },
    PAID: {
        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: PAID,
        DIR_FAIL: CANCELED,
        'log_status': True,
    },
Затем нужно прописать логику, те действия, которые мы хотим выполнить, при наличии этого флага в очередном состоянии объекта.
@staticmethod
    def log_status(order):
        pass
Заключение Я описал наш подход к реализации workflow, который обеспечивает, по моему мнению
  • наглядное описание процесса в коде
  • удобство расширения логики обработки переходов по состояниям
  • удобство изменения схемы процесса и расширения описания процесса
Предложенный подход имеет свои сильные и слабые стороны. Возможно, он не подойдет для реализации любого процесса, но послужит хорошей отправной точкой для ваших проектов по автоматизации процессов. Спасибо что дочитали! Всем добра!
===========
Источник:
habr.com
===========

Похожие новости: Теги для поиска: #_programmirovanie (Программирование), #_fsm, #_workflow, #_konechnyj_avtomat (конечный автомат), #_blog_kompanii_domklik (
Блог компании ДомКлик
)
, #_programmirovanie (
Программирование
)
Профиль  ЛС 
Показать сообщения:     

Вы не можете начинать темы
Вы не можете отвечать на сообщения
Вы не можете редактировать свои сообщения
Вы не можете удалять свои сообщения
Вы не можете голосовать в опросах
Вы не можете прикреплять файлы к сообщениям
Вы не можете скачивать файлы

Текущее время: 22-Ноя 19:18
Часовой пояс: UTC + 5