/
/
1"""Runtime models for the background tasks controller."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import Awaitable, Callable
7from dataclasses import dataclass
8from typing import TYPE_CHECKING, Any
9
10from music_assistant_models.enums import TaskStatus
11
12from .constants import DEFAULT_TASK_LOG_LINES
13
14if TYPE_CHECKING:
15 from music_assistant_models.background_task import BackgroundTask
16
17
18@dataclass
19class ManagedTask:
20 """Runtime state for a managed background task."""
21
22 task_info: BackgroundTask
23 handler: Callable[[], Awaitable[Any]]
24 max_log_lines: int = DEFAULT_TASK_LOG_LINES
25 current_task: asyncio.Task[Any] | None = None
26 timer_delay: float | None = None
27 priority: bool = False
28 removed: bool = False
29 clear_persisted_state_on_remove: bool = True
30
31 @property
32 def is_scheduled(self) -> bool:
33 """Return if this task has a recurring schedule."""
34 return self.task_info.schedule is not None
35
36 @property
37 def is_active(self) -> bool:
38 """Return if this task is pending or running."""
39 return self.task_info.status in (TaskStatus.PENDING, TaskStatus.RUNNING)
40
41 @property
42 def can_remove(self) -> bool:
43 """Return if this task can be removed from history."""
44 return not self.is_scheduled and not self.is_active
45