[Программирование] Реализация конечного автомата для автоматизации процессов
Автор
Сообщение
news_bot ®
Стаж: 6 лет 9 месяцев
Сообщений: 27286
Каждый уважающий себя техлид \ архитектор ПО \ руководитель разработки, должен написать в своей жизни хотя бы одну 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
===========
Похожие новости:
- [Программирование, Алгоритмы, R, Изучение языков] Дети, русский язык и R
- [Программирование, Анализ и проектирование систем, Проектирование и рефакторинг, Микросервисы] Разложение монолита: Декомпозиция БД (часть 1)
- [Программирование, Rust] Почему вы должны попробовать Rust
- [Программирование, Функциональное программирование, TypeScript] Функциональное программирование на TypeScript: задачи (tasks) как альтернатива промисам
- [Программирование, Java, Параллельное программирование, Конференции] Обзор программы JPoint 2021: воркшопы, Spring, «игра вдолгую»
- [Программирование, Java] Удаленный доступ к IDE при помощи Projector
- [Программирование, Математика, Учебный процесс в IT] Computer Science Center открыл приём заявок на новый учебный год
- [Программирование, IT-инфраструктура, Серверная оптимизация, Лайфхаки для гиков] Уход от проблемы TTL или Стратегии корректного и быстрого кэширования (перевод)
- [Ненормальное программирование, Разработка веб-сайтов, Программирование, Haskell] Зачем мы транспилируем Haskell в JavaScript
- [Программирование, Разработка мобильных приложений, Dart, Flutter] Dart 2.12: Sound null safety и Dart FFI отправлены на стабильный канал (перевод)
Теги для поиска: #_programmirovanie (Программирование), #_fsm, #_workflow, #_konechnyj_avtomat (конечный автомат), #_blog_kompanii_domklik (
Блог компании ДомКлик
), #_programmirovanie (
Программирование
)
Вы не можете начинать темы
Вы не можете отвечать на сообщения
Вы не можете редактировать свои сообщения
Вы не можете удалять свои сообщения
Вы не можете голосовать в опросах
Вы не можете прикреплять файлы к сообщениям
Вы не можете скачивать файлы
Текущее время: 22-Ноя 19:18
Часовой пояс: UTC + 5
Автор | Сообщение |
---|---|
news_bot ®
Стаж: 6 лет 9 месяцев |
|
Каждый уважающий себя техлид \ архитектор ПО \ руководитель разработки, должен написать в своей жизни хотя бы одну CRM
народная мудрость class Order:
status responsible price paid WF_STATUSES = (NEW, ORDERED, RESERVED, CANCELLED,
RETURNED, PAID, SHIPPED, DELIVERED, COMPLETED,) def set_ordered(self, request):
... object = self.get_object() object.status = ORDERED object.save() ... class Order:
... def set_ordered(self): pass def set_reserved(self): pass
DIRECTIONS = (DIR_NEXT, DIR_FAIL, DIR_RETURN, DIR_WAIT, DIR_CANCEL, DIR_COMPLETE,)
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 имеет следующий вид:
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) Добавление нового состояния сведется к обновлению собственно JSON-схемы. Если мы хотим добавить новую логику при смене состояний, это удобно сделать через навешивание новых флагов \ признаков в описание каждого состояния в схеме. RESERVED: {
DIR_NEXT: PAID, DIR_FAIL: CANCELED, 'log_status': True, }, @staticmethod
def log_status(order): pass
=========== Источник: habr.com =========== Похожие новости:
Блог компании ДомКлик ), #_programmirovanie ( Программирование ) |
|
Вы не можете начинать темы
Вы не можете отвечать на сообщения
Вы не можете редактировать свои сообщения
Вы не можете удалять свои сообщения
Вы не можете голосовать в опросах
Вы не можете прикреплять файлы к сообщениям
Вы не можете скачивать файлы
Вы не можете отвечать на сообщения
Вы не можете редактировать свои сообщения
Вы не можете удалять свои сообщения
Вы не можете голосовать в опросах
Вы не можете прикреплять файлы к сообщениям
Вы не можете скачивать файлы
Текущее время: 22-Ноя 19:18
Часовой пояс: UTC + 5