101 lines
2.8 KiB
Python
101 lines
2.8 KiB
Python
"""Application-level use cases for the task automation API.
|
|
|
|
The automation engine owns rule evaluation and mutations. This module owns
|
|
the orchestration required by HTTP and other adapters: access-scoped task
|
|
lookup, eligible button selection and interpretation of manual run outcomes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from django.db.models import Q, QuerySet
|
|
|
|
from tasks.access import task_board_access_q
|
|
from tasks.automation.engine import conditions_match, execute_automation
|
|
from tasks.models import (
|
|
Task,
|
|
TaskAutomation,
|
|
TaskAutomationKind,
|
|
TaskAutomationRun,
|
|
TaskAutomationRunStatus,
|
|
TaskBoard,
|
|
)
|
|
|
|
|
|
class TaskAutomationApplicationError(Exception):
|
|
"""Base error raised by task automation use cases."""
|
|
|
|
|
|
class AccessibleTaskNotFound(TaskAutomationApplicationError):
|
|
"""The task does not exist or is outside the actor's accessible boards."""
|
|
|
|
|
|
class ManualAutomationUnavailable(TaskAutomationApplicationError):
|
|
"""The selected rule cannot be launched as a manual button."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManualAutomationOutcome:
|
|
run: TaskAutomationRun | None
|
|
conflict_message: str | None = None
|
|
|
|
|
|
def get_accessible_automation_task(*, actor, task_id: int) -> Task:
|
|
task = (
|
|
Task.objects.select_related("board", "column", "created_by", "assignee")
|
|
.filter(
|
|
board__in=TaskBoard.objects.filter(task_board_access_q(actor)),
|
|
id=task_id,
|
|
)
|
|
.first()
|
|
)
|
|
if task is None:
|
|
raise AccessibleTaskNotFound
|
|
return task
|
|
|
|
|
|
def eligible_button_automations(
|
|
queryset: QuerySet[TaskAutomation],
|
|
*,
|
|
task: Task,
|
|
) -> list[TaskAutomation]:
|
|
candidates = (
|
|
queryset.filter(kind=TaskAutomationKind.BUTTON, is_active=True)
|
|
.filter(Q(applies_to_all_boards=True) | Q(boards=task.board_id))
|
|
.distinct()
|
|
)
|
|
return [
|
|
automation
|
|
for automation in candidates
|
|
if conditions_match(task, automation.conditions)
|
|
]
|
|
|
|
|
|
def run_manual_automation(
|
|
automation: TaskAutomation,
|
|
*,
|
|
task: Task,
|
|
actor,
|
|
) -> ManualAutomationOutcome:
|
|
if automation.kind != TaskAutomationKind.BUTTON or not automation.is_active:
|
|
raise ManualAutomationUnavailable
|
|
|
|
run = execute_automation(
|
|
automation,
|
|
task,
|
|
source="manual",
|
|
actor=actor,
|
|
)
|
|
if run is None:
|
|
return ManualAutomationOutcome(
|
|
run=None,
|
|
conflict_message="Автоматизация не была запущена.",
|
|
)
|
|
if run.status == TaskAutomationRunStatus.SKIPPED:
|
|
return ManualAutomationOutcome(
|
|
run=run,
|
|
conflict_message=(run.error or "Условия автоматизации не выполнены."),
|
|
)
|
|
return ManualAutomationOutcome(run=run)
|