Workflow (Рабочие процессы) Planiqum — Документация для разработчика¶

Архитектура и ключевые сущности¶

Основные модели¶

WorkflowTemplate — шаблон рабочего процесса (процесс верхнего уровня).
key, name, summary, description, define_assignees, define_users
Метод create_workflow() — создание экземпляра процесса по шаблону.
WorkflowTemplateElement — элемент (шаг) шаблона процесса.
template (FK), parent (иерархия), prev (последовательность), summary, duration_days, assignees_groups, owners_groups
Вложенность через parent, последовательность через prev/next_task.
Ключевой метод: автоматическая синхронизация длительности по подзадачам.
Issue — экземпляр задачи (шаг процесса).
Наследует структуру от WorkflowTemplateElement.
parent, prev, subtasks, duration_days, start_date, due_date, state, resolution, assignees, owners
Методы для переходов статусов, уведомлений, расчёта длительности, каскадного обновления.
Resolution — справочник решений (статусов закрытия задачи).

Ключевые поля и связи¶

parent — иерархия (вложенность задач).
prev/next_task — последовательность (критический путь).
subtasks — подзадачи (children).
duration_days — длительность задачи (автоматически синхронизируется).
assignees/owners — исполнители и владельцы (группы и пользователи).

Бизнес-логика и методы¶

Критический путь и длительность¶

Критический путь: для родительской задачи длительность определяется максимальной суммой длительностей по всем параллельным и последовательным веткам (по prev).
Функция расчёта:
get_critical_path_duration(obj) (см. workflow/utils.py)
Рекурсивно обходит дерево задач, учитывает вложенность и последовательность.
Для параллельных веток — берёт максимум, для последовательных — сумму.
Синхронизация длительности:
Метод sync_duration_days_from_subtasks() в абстрактном классе WorkflowTemplateElementAbstract и его наследниках.
При изменении подзадачи автоматически пересчитывает длительность всех родителей до root.
Расчёт дат:
calculate_due_date(start_date, duration_days) — учитывает рабочие дни (если настроен календарь).
calculate_duration(start_date, due_date) — считает длительность между датами.

Примеры кода и тестов¶

Пример структуры процесса¶

root
├─ A (1) → B (2) → C (3) → D (4) → E (5)
├─ F (2)
│   └─ G (2)
│       └─ Z (1)
└─ H (1)
    ├─ LONG (12)
    ├─ S1 (3) → S2 (4)

- Критический путь: root→A→B→C→D→E (1+2+3+4+5=15) - Для H: duration_days = 12 (по LONG), а не 7 (по S1→S2)

Пример теста (test_workflow_smoke.py)¶

@pytest.mark.django_db
def test_critical_path_and_cascade_update_with_nested(load_workflow_hierarchy_and_groups):
    """
    Проверяет:
    - Расчёт критического пути для структуры с параллельными и последовательными задачами
    - Каскадное обновление duration_days при изменении глубоко вложённой задачи
    - Корректную работу рекурсии при создании и обновлении
    - Синхронизацию duration_days у элементов шаблона и задач

    Структура:
    root
    ├─ A (1) → B (2) → C (3) → D (4) → E (5)
    ├─ F (2)
    │   └─ G (2)
    │       └─ Z (1)
    └─ H (1)
        ├─ LONG (12)
        ├─ S1 (3) → S2 (4)
    """
    # ... создание шаблона, элементов, задач ...
    # ... assert'ы на duration_days, каскадные обновления ...

Пример использования функции расчёта критического пути¶

from planiqum.core.workflow.utils import get_critical_path_duration

duration = get_critical_path_duration(issue_or_template_element)

Как изучать и тестировать логику¶

Тесты:
Все ключевые сценарии покрыты в test_workflow_smoke.py.
Используйте docstring с ASCII-нотацией для визуализации структуры.
Проверяйте не только значения, но и каскадные обновления при изменении/добавлении подзадач.
Рекомендации:
Для сложных сценариев используйте вложенность, параллельные и последовательные ветки.
Изучайте методы моделей: save, sync_duration_days_from_subtasks, create_workflow, get_critical_path_duration.

Важные детали¶

duration_days родителя всегда определяется только по подзадачам (максимум по критическим путям), своё значение игнорируется, если есть подзадачи.
Каскадный пересчёт длительности происходит при изменении любой вложенной задачи.
Вся логика расчёта централизована и универсальна для шаблонов и экземпляров задач.
Все изменения отражаются в тестах и docstring.

Где искать код¶

Модели: src/planiqum/core/workflow/models.py
Бизнес-логика: src/planiqum/core/workflow/utils.py
Тесты: src/planiqum/core/workflow/tests/test_workflow_smoke.py

Если требуется добавить примеры миграций, сериализации, интеграции с другими модулями — сообщите! Документация готова для глубокого изучения и расширения рабочей логики процессов в Planiqum.

Работа с календарём рабочих дней¶

Все расчёты дат и длительностей в workflow используют механизм календаря рабочих дней, если он настроен (см. отдельную статью).
Если календарь не настроен — используется обычный календарь.

⚠️ Критическая ошибка конфигурации¶

Проблема: Если в SiteSettings указан ключ меры календаря рабочих дней (WORKING_DAYS_CALENDAR_MEASURE_KEY), но сама мера отсутствует, то при создании задач (Issue.save()) выбрасывается ValueError.

Места возникновения ошибки: - core/workflow/models.py, метод Issue.calculate_due_date() (строки 2015-2017) - core/workflow/models.py, метод Issue.calculate_duration() (строка 1407)

Текст ошибки:

ValueError: В системных настройках указан ключ меры 'working_days_calendar__is_working_day', но мера не найдена. Проверьте настройки или создайте меру для календаря рабочих дней.

Последствия: Ошибка приводит к FAILURE при выполнении сценариев создания workflow, включая create_workflow_by_template.

Технические детали реализации Workflow¶

Основные классы и файлы¶

Модели: src/planiqum/core/workflow/models.py
WorkflowTemplate, WorkflowTemplateElement, Issue, Resolution
Бизнес-логика: src/planiqum/core/workflow/utils.py
Функции: calculate_due_date, calculate_duration, get_critical_path_duration, _get_working_days_from_db
Обработка сохранения задач: src/planiqum/core/workflow/issue_save_handler.py
Класс: IssueSaveHandler, метод: prepare_dates()
Тесты: src/planiqum/core/workflow/tests/test_issue_working_days.py, test_workflow_smoke.py

Ключевые поля моделей¶

Issue.start_date, Issue.due_date, Issue.duration_days, Issue.parent, Issue.prev, Issue.subtasks
WorkflowTemplateElement.duration_days, prev, parent

Алгоритмы расчёта дат и длительности¶

Все расчёты дат и длительностей используют функции из workflow/utils.py:
calculate_due_date(start_date, duration_days) — учитывает рабочие дни, если календарь настроен (см. dev/working_days_calendar.md)
calculate_duration(start_date, due_date) — считает рабочие дни, если календарь настроен
Если календарь не настроен — fallback на календарные дни
Каскадный пересчёт дат и длительности реализован в IssueSaveHandler.prepare_dates()
При изменении даты начала, окончания или длительности автоматически пересчитываются связанные поля
Каскадное обновление по иерархии задач (родитель/подзадачи, prev/next_task)

Взаимодействие с календарём рабочих дней¶

Ключ меры календаря рабочих дней берётся из SiteSettings (WORKING_DAYS_CALENDAR_MEASURE_KEY)
Для получения рабочих дней используется _get_working_days_from_db
Если мера не найдена — расчёты ведутся по календарным дням

Примеры кода¶

Пример пересчёта дат и длительности:

from planiqum.core.workflow.utils import calculate_due_date, calculate_duration
due_date = calculate_due_date(issue.start_date, issue.duration_days)
duration = calculate_duration(issue.start_date, issue.due_date)

Пример каскадного обновления дат:

handler = IssueSaveHandler(issue)
handler.prepare_dates()

Где искать¶

Модели: src/planiqum/core/workflow/models.py
Логика расчёта: src/planiqum/core/workflow/utils.py
Обработка изменений: src/planiqum/core/workflow/issue_save_handler.py
Тесты: src/planiqum/core/workflow/tests/

Автоматическое создание рабочих процессов по расписанию ¶

Ключевые модели¶

WorkflowSchedule (src/planiqum/core/workflow/models.py)
Описывает расписание для автоматического создания процессов по шаблону.
Поля:
- template — ссылка на шаблон (WorkflowTemplate)
- is_active — активность расписания
- start_date, end_date — диапазон действия расписания
- frequency — периодичность (ежедневно, еженедельно, ежемесячно, ежегодно)
- interval — интервал между запусками (например, каждые 2 недели)
- workflow_creation_offset_days — за сколько дней до старта создавать процесс
- summary — переопределяет шаблон описания
Основные методы:
- should_create_workflow(date=None) — определяет, нужно ли создавать процесс для указанной даты
- get_next_start_date(reference_date=None) — вычисляет дату следующего запуска
WorkflowTemplate (src/planiqum/core/workflow/models.py)
Метод create_workflow(...) — создаёт процесс по шаблону, учитывая расписание и параметры.
Аргумент scheduler — ссылка на объект WorkflowSchedule, если процесс создаётся по расписанию.

Как работает автоматизация¶

Периодическая задача Celery (см. workflow_maintenance_task в src/planiqum/core/workflow/scripts/workflow_maintenance.py) перебирает все активные объекты WorkflowSchedule.
Для каждого расписания вызывается should_create_workflow(). Если пора — вызывается WorkflowTemplate.create_workflow(...).
Вся логика расчёта дат, проверки интервалов, учёта offset и т.д. реализована в методах модели WorkflowSchedule.
Созданные процессы и задачи получают ссылку на объект расписания (scheduler).

Где искать код¶

Модели: src/planiqum/core/workflow/models.py (классы WorkflowSchedule, WorkflowTemplate)
Логика запуска: src/planiqum/core/workflow/scripts/create_scheduled_workflows.py, workflow_maintenance.py
Тесты: src/planiqum/core/workflow/tests/ (например, test_workflow_smoke.py)

Пример теста¶

@pytest.mark.django_db
def test_workflow_schedule_creates_process():
    # Создаём шаблон и расписание
    template = WorkflowTemplate.objects.create(...)
    schedule = WorkflowSchedule.objects.create(
        template=template,
        is_active=True,
        start_date=date(2024, 1, 1),
        frequency='daily',
        interval=1
    )
    # Проверяем, что процесс будет создан для нужной даты
    should_create, next_start_date = schedule.should_create_workflow(date(2024, 1, 1))
    assert should_create
    # Запускаем создание процесса
    process = template.create_workflow(scheduler=schedule, start_date=next_start_date)
    assert process is not None

Важные детали¶

Расписание полностью управляется через Django-модели, а не через celery-beat напрямую.
Можно создавать несколько расписаний для одного шаблона.
Вся логика проверки, когда создавать процесс, инкапсулирована в модели.
Для отладки используйте логику и тесты из test_workflow_smoke.py и других тестов.