/
/
/
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 run_token: str = ""
31
32 @property
33 def is_scheduled(self) -> bool:
34 """Return if this task has a recurring schedule."""
35 return self.task_info.schedule is not None
36
37 @property
38 def is_active(self) -> bool:
39 """Return if this task is pending or running."""
40 return self.task_info.status in (TaskStatus.PENDING, TaskStatus.RUNNING)
41
42 @property
43 def can_remove(self) -> bool:
44 """Return if this task can be removed from history."""
45 return not self.is_scheduled and not self.is_active
46