Реализация кода на основе UML–диаграмм: пример на Python

Практическое использование UML-диаграмм: реализация классов для работы с документами на Python. Пример СЭД на минималках с блэк-джеком задачами согласования и девушками пользователями.

Простые конструкторы классов на Python

Чтобы реализовать постановку задачи, описанную в прошлой статье, создадим классы, определенные на UML-диаграмме.

Класс – это конструктор для создания объектов. Помимо определения статической структуры объекта, т.е. набора его полей (атрибутов) с типом данных для каждого из них, в конструкторе класса также описываются его методы – функции для работы с объектом этого класса. Например, согласно UML-диаграмме класса, документ можно опубликовать, а в маршрут можно добавить задачу. Для этого в конструкторах классов document и route предусмотрены соответствующие методы.

Чтобы вы могли повторить это упражнения, не устанавливая среду разработки, далее я приведу код на Python для запуска его в интерактивной среде Google Colab. Для удобства разделим код на ячейки. Сперва импортируем необходимые модули: datetime для работы с датой и временем, Enum для работы с перечислениями, а также List и Optional из пакета typing, который предоставляет инструменты для указания типов данных в аннотациях.

################################# ячейка 1 - импорт модулей ###############
from datetime import datetime
from enum import Enum
from typing import List, Optional

Для работы с перечислениями в Python используется модуль Enum. Для перечислений не нужен метод инициализации, поскольку Enum – базовый класс, на основе которого создаются перечисления, автоматически управляет созданием своих элементов. Когда создается класс-перечисление, Python автоматически генерирует нужные методы для создания и управления элементами перечисления. Создадим классы-перечисления, в которых определены ограниченные списки значений, такие как роли пользователей, состояния жизненного цикла документов, задач и маршрутов, а также допустимые форматы файлов. В терминах домено-ориентированного проектирования (DDD, Domain-Driven Design) эти классы будут объектами-значениями (Value Object), которые не имеют собственной идентичности.

Код классов-перечислений:

################################# ячейка 2 - создание перечислений ############### 
class role(Enum):
    secretary = 'секретарь'
    administrator = 'администратор'
    employee = 'сотрудник'

class task_status(Enum):
    new = 'новая'
    assigned = 'поставлена'
    started = 'принята к исполнению'
    declined_by_executor = 'отклонена исполнителем'
    performing = 'выполняется'
    canceled_by_initiator = 'отменена инициатором'
    completed_by_executor = 'согласована исполнителем'
    rejected_by_executor = 'несогласована исполнителем'
    frozen = 'заморожена'
    accepted_by_initiator = 'принята инициатором'
    finished = 'завершена'

class route_status(Enum):
    new = 'новый'
    started = 'начат'
    in_progress = 'выполняется'
    canceled = 'отменен'    
    completed = 'завершен'

class file_format(Enum):
    pdf = 'pdf'
    doc = 'doc'
    xls = 'xls'
    ppt = 'ppt'

class document_status(Enum):
    new = 'новый'
    published = 'опубликован'
    on_review = 'на согласовании'
    approved = 'согласован'
    rejected = 'несогласован'

Далее создадим сами классы сущностей и агрегатов в терминах DDD. Поскольку для сущностей и агрегатов явно создаются объекты, нужно использовать метод __init__ в объявлении класса. Метод __init__ в Python называется конструктором класса: он используется для инициализации объектов класса и вызывается автоматически при создании нового экземпляра. Первый аргумент метода __init__ всегда должен быть self, который представляет собой экземпляр создаваемого класса. Это позволяет методу обращаться к атрибутам и методам объекта. Дополнительные аргументы могут быть добавлены для передачи данных при создании объекта. В методе __init__ обычно задаются начальные значения атрибутов объекта.

Для наглядности визуализации данных об объектах каждого класса добавим в конструкторы метод info. Он будет выводить информацию о значениях атрибутов объекта в виде словаря формата JSON (ключ-значение). Это особенно пригодится, чтобы посмотреть информацию об объектах-агрегатах: маршруте согласования документа, в который пользователь-инициатор добавил несколько задач. Например, следующий код показывает объявление класса user, который вызывается при создании нового пользователя.

Код класса user:

class user:
    def __init__(self, login: str, email: str, phone: str, role: role):
        self.login = login
        self.email = email
        self.phone = phone
        self.role = role

    def info(self):
        return {
            "login": self.login,
            "email": self.email,
            "phone": self.phone,
            "role": self.role.value
        }

Код метода  инициализации принимает четыре аргумента:

• login — строка, представляющая логин пользователя.
• email — строка, представляющая адрес электронной почты пользователя.
• phone — строка, представляющая номер телефона пользователя.
• role — объект типа данных role, представляющий роль пользователя как объект класса role, ранее описанного как перечисление.

Метод info() возвращает информацию о пользователе в виде словаря. Ключи словаря — это строки, представляющие названия атрибутов, а значения — соответствующие атрибуты объекта класса user.

Аналогичный по набору методов код для создания объекта файл документа выглядит так:

class doc_file:
    def __init__(self, file_format: file_format, size: int):
        self.file_format = file_format
        self.size = size

    def info(self):
        return {
            "file_format": self.file_format.value,
            "size": self.size
        }

Класс, описывающий документ, будет сложнее, поскольку он содержит методы управления документом, а также атрибуты документа:

• number – строка с номером документа;
• created_timestamp – время и дата создания документа, устанавливаемая при инициализации объекта;
• published_timestamp – время и дата публикации документа, которая не задана (None), когда документ ещё не опубликован;
• status – статус, состояние жизненного цикла документа, который имеет значения из перечисления document_status, описанного ранее, например, new, published, on_review, approved, rejected;
• author – автор документа, т.е. объект класса user, создавший документ;
• doc_file — файл, связанный с документом, объект класса doc_file, который может быть не задан (None), если к документу не присоединен файл.

Код класса document:

class document:
    def __init__(self, number: str, author: user, published_timestamp: Optional[datetime] = None, doc_file: Optional[doc_file] = None):
        self.number = number
        self.created_timestamp = datetime.now()
        self.published_timestamp = published_timestamp
        self.status = document_status.new
        self.author = author
        self.doc_file = doc_file

    def add_file(self, doc_file: doc_file):
        self.doc_file = doc_file

    def delete_file(self):
        self.doc_file = None

    def publish_document(self):
        self.status = document_status.published
        self.published_timestamp = datetime.now()

    def review_document(self):
        self.status = document_status.on_review

    def approve_document(self):
        self.status = document_status.approved

    def reject_document(self):
        self.status = document_status.rejected

    def info(self):
        return {
            "number": self.number,
            "created_timestamp": self.created_timestamp.isoformat() if self.created_timestamp else None,
            "published_timestamp": self.published_timestamp.isoformat() if self.published_timestamp else None,
            "author": self.author.info(),
            "status": self.status.value,
            "file": self.doc_file.info() if self.doc_file else None
        }

Класс document содержит следующие методы:

• __init__() — конструктор, который инициализирует объект документа с номером, автором, временной меткой публикации и файлом. При создании он заполняет временную метку создания документа и устанавливает его начальный статус new;
• add_file() – функция добавления или замены файла, связанного с документом;
• delete_file() — функция удаления файла, связанного с документом, устанавливает значение атрибута doc_file в None;
• publish_document () — функция публикации документа, устанавливает статус документа в значение published и временную метку публикации текущим датой и временем;
• review_document() — функция перевода документа в статус согласования (on_review);
• approve_document() — функция согласования документа, устанавливает ему статус approved;
• reject_document() – функция отказа в согласовании документа, устанавливает ему статус rejected;
• info() – функция получения информации о документе в виде словаря: номер, временные метки создания и публикации, информацию об авторе, статус и информацию о файле (если файл существует). Для форматирования временных меток в атрибутах created_timestamp и published_timestamp вызывается метод isoformat(), если они установлены.

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

DDD, ООП и UML для аналитика

Код курса

BUML

Ближайшая дата курса

9 декабря, 2024

Продолжительность

16 ак.часов

Стоимость обучения

36 000 руб.

Взаимозависимые состояния и обратные вызовы

Как уже было отмечено выше, с классами task и route немного сложнее. Опишем класс задачи со следующими атрибутами:

• name – строка с названием;
• created_timestamp – время и дата создания задачи, устанавливается в момент создания объекта;
• status — текущий статус задачи, по умолчанию равен значению, заданном в перечислении, task_status.new;
• started_timestamp – время и дата начала выполнения задачи, может быть None;
• finished_timestamp – время и дата завершения задачи, может быть None;
• executor — исполнитель задачи, объект класса пользователь;
• _status_change_callback**: Приватный атрибут для хранения функции обратного вызова, которая будет вызываться при изменении статуса задачи.

Код класса task:

class task:
    def __init__(self, name: str, executor: user, started_timestamp: Optional[datetime] = None, finished_timestamp: Optional[datetime] = None):
        self.name = name
        self.created_timestamp = datetime.now()
        self.status = task_status.new
        self.started_timestamp = started_timestamp
        self.finished_timestamp = finished_timestamp
        self.executor = executor
        self._status_change_callback = None

    def set_status_change_callback(self, callback):
        self._status_change_callback = callback

    def _notify_status_change(self):
        if self._status_change_callback:
            self._status_change_callback(self)
        self.update_task_status()

    def start_task(self):
        self.status = task_status.started
        self.started_timestamp = datetime.now()
        self._notify_status_change()

    def receive_task(self):
        self.status = task_status.assigned
        self._notify_status_change()

    def decline_task(self):
        self.status = task_status.declined_by_executor
        self._notify_status_change()

    def cancel_task(self):
        self.status = task_status.canceled_by_initiator
        self.finished_timestamp = datetime.now()
        self._notify_status_change()

    def perform_task(self):
        self.status = task_status.performing
        self._notify_status_change()

    def accept_task(self):
        self.status = task_status.accepted_by_initiator
        self._notify_status_change()

    def complete_task(self):
        self.status = task_status.completed_by_executor
        self._notify_status_change()

    def reject_task(self):
        self.status = task_status.rejected_by_executor
        self._notify_status_change()

    def freeze_task(self):
        self.status = task_status.frozen
        self._notify_status_change()

    def finish_task(self):
        self.status = task_status.finished
        self.finished_timestamp = datetime.now()
        self._notify_status_change()

    def update_task_status(self):
        if self.status in [task_status.accepted_by_initiator]:
            self.finish_task()

    def info(self):
        return {
            "name": self.name,
            "created_timestamp": self.created_timestamp.isoformat() if self.created_timestamp else None,
            "started_timestamp": self.started_timestamp.isoformat() if self.started_timestamp else None,
            "finished_timestamp": self.finished_timestamp.isoformat() if self.finished_timestamp else None,
            "executor": self.executor.info(),
            "status": self.status.value
        }

В этом коде определены следующие методы класса:

• __init__() – конструктор, инициализирует объект задачи с заданными параметрами;
• set_status_change_callback() — функция обратного вызова для уведомления об изменении статуса;
• _notify_status_change() — внутренний метод, вызывает функцию обратного вызова при изменении статуса задачи;
• start_task() – функции старта задачи, устанавливает статус задачи как task_status.started и фиксирует временную метку ее старта;
• receive_task() – функция назначения задачи на исполнителя, устанавливает статус задачи как task_status.assigned;
• decline_task() – функция отклонения задачи исполнителем, устанавливает ее статус как task_status.declined_by_executor;
• cancel_task() – функция отмены задачи инициатором, устанавливает ее статус как task_status.canceled_by_initiator и фиксирует временную метку завершения текущим временем;
• perform_task() – функция выполнения задачи, задает ее статус как task_status.performing;
• freeze_task() – функция заморозки задачи, устанавливает ее статус как task_status.frozen;
• complete_task() – функция согласования задачи, устанавливает ее статус как task_status.completed_by_executor и вызывает метод update_task_status;
• reject_task() – функция несогласования задачи, устанавливает ее статус как task_status.rejected_by_executor и вызывает метод update_task_status;
• accept_task() – функция принятия задачи инициатором, устанавливает ее статус как task_status.accepted_by_initiator и вызывает метод update_task_status;
• finish_task() – функция завершения задачи, устанавливает ее статус как task_status.finished и фиксирует временную метку завершения;
• update_task_status() – функция, которая проверяет статус задачи и вызывает метод finish_task, если задача была принята инициатором;
• info() – функция получения информации о задаче в виде словаря: имя, временные метки, информацию об исполнителе и текущий статус. Для форматирования временных меток в соответствующих атрибутах вызывается метод isoformat(), если они установлены.

Поскольку состояние маршрута меняется в зависимости от входящих в него задач, их необходимо связать. В рассмотренном примере это делается с помощью функций обратного вызова (callback): когда статус задачи изменяется, маршрут обновляет свой статус соответственно. Сallback передается как аргумент другой функции и вызывается в определённый момент, например, когда происходит какое-то событие. Например, функция обратного вызова используется для уведомления маршрута о том, что статус задачи изменился.

Когда задача добавляется в маршрут с помощью метода add_task, для этой задачи устанавливается функция обратного вызова с помощью метода set_status_change_callback:

def add_task(self, task: task):
    task.set_status_change_callback(self.update_route_status)
    self.tasks.append(task)
    self.update_route_status()

Сам метод set_status_change_callback() описан в классе task. Он сохраняет переданную функцию обратного вызова в атрибуте _status_change_callback задачи:

def set_status_change_callback(self, callback):
        self._status_change_callback = callback

Каждый раз, когда статус задачи изменяется, вызывается внутренний метод задачи _notify_status_change(), который проверяет наличие функции обратного вызова и вызывает её, передавая текущую задачу в качестве аргумента. Также здесь вызывается функция обновления статуса задачи:

def _notify_status_change(self):
    if self._status_change_callback:
       self._status_change_callback(self)
    self.update_task_status()

Методы, изменяющие статус задачи, такие как start_task(), receive_task(), decline_task() и пр., вызывают метод _notify_status_change() после изменения статуса. Например, метод, который переводит задачу в состояние начата (started) и устанавливает ей время старта:

def start_task(self):
        self.status = task_status.started
        self.started_timestamp = datetime.now()
        self._notify_status_change()

При вызове метода _notify_status_change() переданная функция обратного вызова, которая является методом update_route_status() у класса route, вызывается и обновляет статус маршрута в зависимости от текущего состояния всех задач:

def update_route_status(self, _=None):
        if any(task.status == task_status.rejected_by_executor for task in self.tasks):
            self.document.reject_document()
        elif all(task.status == task_status.completed_by_executor for task in self.tasks):
            self.document.approve_document()
        elif all(task.finished_timestamp is not None for task in self.tasks):
            self.complete_route()
        elif any(task.status == task_status.started for task in self.tasks):
            self.start_route()
            self.document.publish_document()
        elif any(task.status == task_status.assigned or task.status == task_status.performing for task in self.tasks):
            self.status = route_status.in_progress
            self.document.review_document()
        else:
            self.status = route_status.new
            self.created_timestamp = datetime.now()

Таким образом, класс маршрута (route), который управляет процессом маршрутизации документа и связанными с ним задачами, будет иметь следующие атрибуты:

• created_timestamp – дата и время создания маршрута;
• status — статус маршрута, который может принимать значения из перечисления route_status;
• started_timestamp – дата и время старта маршрута (может быть None);
• finished_timestamp – дата и время завершения маршрута (может быть None);
• document – объект класса документ, связанный с маршрутом;
• initiator — инициатор маршрута, объекта класса пользователь;
• tasks — список задач, добавленных в маршрут, набор объектов класса task.

Код класса route:

class route:
    def __init__(self, document: 'document', initiator: 'user', started_timestamp: Optional[datetime] = None, finished_timestamp: Optional[datetime] = None):
        self.created_timestamp = datetime.now()
        self.status = route_status.new
        self.started_timestamp = started_timestamp
        self.finished_timestamp = finished_timestamp
        self.document = document
        self.initiator = initiator
        self.tasks: List[task] = []

    def add_task(self, task: task):
        task.set_status_change_callback(self.update_route_status)
        self.tasks.append(task)
        self.update_route_status()

    def delete_task(self, task: task):
        self.tasks.remove(task)
        self.update_route_status()

    def start_route(self):
        self.status = route_status.started
        self.started_timestamp = datetime.now()
        self.document.publish_document()
        for task in self.tasks :
            task.status = task_status.assigned

    def cancel_route(self):
        self.status = route_status.canceled
        self.finished_timestamp = datetime.now()
        for task in self.tasks :
            task.status = task_status.canceled_by_initiator

    def perform_route(self):
        self.status = route_status.in_progress
        self.update_route_status()

    def complete_route(self):
        self.status = route_status.completed
        self.finished_timestamp = datetime.now()

    def get_all_tasks(self) -> List[dict]:
        return [task.info() for task in self.tasks]

    def update_route_status(self, _=None):
        if any(task.status == task_status.rejected_by_executor for task in self.tasks):
            self.document.reject_document()
        elif all(task.status == task_status.completed_by_executor for task in self.tasks):
            self.document.approve_document()
        elif all(task.finished_timestamp is not None for task in self.tasks):
            self.complete_route()
        elif any(task.status == task_status.started for task in self.tasks):
            self.start_route()
            self.document.publish_document()
        elif any(task.status == task_status.assigned or task.status == task_status.performing for task in self.tasks):
            self.status = route_status.in_progress
            self.document.review_document()
        else:
            self.status = route_status.new
            self.created_timestamp = datetime.now()

    def info(self):
        return {
            "created_timestamp": self.created_timestamp.isoformat() if self.created_timestamp else None,
            "started_timestamp": self.started_timestamp.isoformat() if self.started_timestamp else None,
            "finished_timestamp": self.finished_timestamp.isoformat() if self.finished_timestamp else None,
            "initiator": self.initiator.info(),
            "document": self.document.info(),
            "status": self.status.value,
            "tasks": self.get_all_tasks()
        }

В этом коде определены следующие методы:

• __init__() — инициализация маршрута с заданными документом, инициатором и временными метками;
• add_task() – функция добавления задачи к маршруту и обновления его статуса;
• delete_task() – функция удаления задачи из маршрута и обновления его статуса;
• start_route() – функция старта маршрута с установкой соответствующего статуса и временной метки старта;
• cancel_route() – функция отмены маршрута с установкой соответствующего статуса и временной метки завершения;
• perform_route() – функция выполнения маршрута с установкой соответствующего статуса;
• complete_route() – функция завершения маршрута с установкой соответствующего статуса и временной метки завершения;
• get_all_tasks() – функция получения информации обо всех задачах маршрута в формате словарей, включая название, временные метки и исполнителя;
• update_route_status() – функция обновления статуса маршрута в зависимости от статусов входящих в него задач. Если все задачи завершены, маршрут помечается как завершённый. Если любая задача отклонена исполнителем, документ отклоняется, а маршрут переводится в состояние выполняется. Если все задачи согласованы исполнителями, документ становится согласован, а маршрут находится в состоянии выполняется до тех пор, пока инициатор не примет каждую задачу. Когда любая задача начата, маршрут тоже стартует, а неопубликованный документ публикуется. Если любая задача назначена или выполняется, маршрут переводится в состояние выполняется, а документ отправляется на согласование. Иначе маршрут остаётся новым и время создания обновляется.
• info() – функция получения информации о маршруте в виде словаря, включая информацию о документе, инициаторе и задачах. Для форматирования временных меток в соответствующих атрибутах вызывается метод isoformat(), если они установлены.

После определения всех классов, наконец, можно приступить к созданию объектов и работе с ними. Это рассмотрим далее.

DDD, ООП и UML для аналитика

Код курса

BUML

Ближайшая дата курса

9 декабря, 2024

Продолжительность

16 ак.часов

Стоимость обучения

36 000 руб.

Манипулирование объектами

Сперва создадим объекты классов:

################################# ячейка 3 - создание объектов ###############
# Создание пользователей
user_obj_1 = user(login="user1", email="user1@example.com", phone="1234567890", role=role.secretary)
user_obj_2 = user(login="user2", email="user2@example.com", phone="0987654321", role=role.administrator)
user_obj_3 = user(login="user3", email="user3@example.com", phone="1122334455", role=role.employee)

# Создание файла
doc_file_obj = doc_file(file_format=file_format.pdf, size=1024)

# Создание документа
doc_1 = document(number="DOC123", author=user_obj_1, doc_file=doc_file_obj)

# Создание задачи
task_obj_1 = task(name='задача 1', executor=user_obj_3)
task_obj_2 = task(name='задача 2', executor=user_obj_2)

# Создание маршрута
route_obj = route(document=doc_1, initiator=user_obj_2)

# Добавление задачи в маршрут
route_obj.add_task(task_obj_1)
route_obj.add_task(task_obj_2)

Выведем информацию о созданном маршруте:

#вывод информации о маршруте
route_obj.info()

Выведем информацию о задачах маршрута, вызвав метод get_all_tasks() у объекта route_obj:

#вывод информации о всех задачах конкретного маршрута
route_obj.get_all_tasks()

Вывод информации о сложных объектах

Теперь реализуем логику управления документами, стартовав маршрут. Входящие в него задачи перейдут в состояние поставлена.

#стартовать маршрут
route_obj.start_route()

#показать информацию о маршруте
route_obj.info()

Предположим, один из исполнителей согласовал документ, а другой – нет.

############## новая ячейка - управление задачами #######################
task_obj_1.start_task() #стартовать задачу 1
print("---------------- задача 1 стартована ---------------- ")
print("Статус задачи: ", task_obj_1.status.value)
print("Время старта задачи: ", task_obj_1.started_timestamp)
print("\n---------------- задача 1 выполнена - согласовано ---------------- ")
task_obj_1.complete_task() #согласовать задачу 1
print("Статус задачи: ", task_obj_1.status.value)
print("\n---------------- выполненная задача 1 принята инициатором---------------- ")
task_obj_1.accept_task() #принять задачу 1
print("Статус задачи: ", task_obj_1.status.value)
print("Время завершения задачи: ", task_obj_1.finished_timestamp)

task_obj_2.start_task() #стартовать задачу 2
print("\n---------------- задача 2 стартована---------------- ")
print("Статус задачи: ", task_obj_2.status.value)
print("Время старта задачи: ", task_obj_2.started_timestamp)
task_obj_2.reject_task() #отказать в согласовании документа по задачу 2
print("\n---------------- задача 2 выполнена - не согласовано---------------- ")
print("Статус задачи: ", task_obj_2.status.value)
print("\n---------------- выполненная задача 2 принята инициатором---------------- ")
task_obj_2.accept_task() #принять задачу 2
print("Статус задачи: ", task_obj_2.status.value)
print("Время завершения задачи: ", task_obj_2.finished_timestamp)

#узнаем статус маршрута
print("\nСтатус маршрута: ", route_obj.status.value)
print("Время завершения маршрута: ", route_obj.finished_timestamp)
#узнаем статус документа
print("Статус документа: ",  route_obj.document.status.value)

DDD, ООП и UML для аналитика

Код курса

BUML

Ближайшая дата курса

9 декабря, 2024

Продолжительность

16 ак.часов

Стоимость обучения

36 000 руб.

Хотя итоговый код немного отличается от первичной постановки задачи и проекта в виде набора UML-диаграмм, этап проектирования помог реализовать не только структуру доменных объектов, но и логику их поведения. Поэтому умение разрабатывать UML-диаграммы очень полезный навык для аналитика. Освоить его вы сможете на моих курсах в Школе прикладного бизнес-анализа на базе нашего лицензированного учебного центра обучения и повышения квалификации системных и бизнес-аналитиков в Москве:

• UML для бизнес-аналитика