/
/
/
1"""Task execution context helpers for long running background tasks."""
2
3from __future__ import annotations
4
5from contextvars import ContextVar
6from dataclasses import dataclass
7from typing import TYPE_CHECKING
8
9if TYPE_CHECKING:
10 from collections.abc import Callable
11
12 from music_assistant_models.background_task import BackgroundTask
13
14
15@dataclass(slots=True)
16class TaskExecutionContext:
17 """Runtime task context exposed to long-running task code."""
18
19 task_id: str
20 get_task: Callable[[str], BackgroundTask]
21 update_progress: Callable[[str, int | None, str | None], None]
22 update_progress_text: Callable[[str, str | None], None]
23 add_failure: Callable[[str, str], None]
24 update_report: Callable[[str, str | None], None]
25
26 @property
27 def task(self) -> BackgroundTask:
28 """Return the attached background task object."""
29 return self.get_task(self.task_id)
30
31 def set_progress(self, progress: int | None, text: str | None = None) -> None:
32 """Set an absolute progress percentage and optional phase text."""
33 self.update_progress(self.task_id, progress, text)
34
35 def set_progress_text(self, text: str | None) -> None:
36 """Update the human-readable progress text only."""
37 self.update_progress_text(self.task_id, text)
38
39 def set_progress_from_index(self, current: int, total: int, text: str | None = None) -> int:
40 """Set progress from the current item index and total item count."""
41 progress = calculate_progress(current, total)
42 self.update_progress(self.task_id, progress, text)
43 return progress
44
45 def record_failure(self, message: str) -> None:
46 """Record a non-fatal failure for the current task."""
47 self.add_failure(self.task_id, message)
48
49 def set_report(self, markdown: str | None) -> None:
50 """Set the Markdown report for the current task."""
51 self.update_report(self.task_id, markdown)
52
53
54ACTIVE_TASK_CONTEXT: ContextVar[TaskExecutionContext | None] = ContextVar(
55 "active_background_task_context",
56 default=None,
57)
58
59
60def calculate_progress(current: int, total: int) -> int:
61 """Convert the current item index and total item count into a percentage."""
62 if total <= 0:
63 raise ValueError("Task progress total must be > 0")
64 current = max(current, 0)
65 return min(int((current * 100) / total), 100)
66
67
68def get_current_task_context() -> TaskExecutionContext | None:
69 """Return the task context active in the current async/thread context."""
70 return ACTIVE_TASK_CONTEXT.get()
71
72
73def get_current_task() -> BackgroundTask | None:
74 """Return the active background task for the current async/thread context."""
75 if task_context := get_current_task_context():
76 return task_context.task
77 return None
78
79
80def get_current_task_id() -> str | None:
81 """Return the active task id for the current async/thread context."""
82 if task_context := get_current_task_context():
83 return task_context.task_id
84 return None
85
86
87def update_current_task_progress(progress: int | None, text: str | None = None) -> None:
88 """Update progress for the task active in the current async/thread context."""
89 if task_context := get_current_task_context():
90 task_context.set_progress(progress, text)
91
92
93def update_current_task_progress_text(text: str | None) -> None:
94 """Update progress text for the task active in the current async/thread context."""
95 if task_context := get_current_task_context():
96 task_context.set_progress_text(text)
97
98
99def update_current_task_progress_from_index(
100 current: int, total: int, text: str | None = None
101) -> int | None:
102 """Update progress from item counts for the current async/thread context."""
103 if task_context := get_current_task_context():
104 return task_context.set_progress_from_index(current, total, text)
105 return None
106
107
108def report_current_task_failure(message: str) -> None:
109 """Record a non-fatal failure for the current async/thread context."""
110 if task_context := get_current_task_context():
111 task_context.record_failure(message)
112
113
114def set_current_task_report(markdown: str | None) -> None:
115 """Set the Markdown report for the current async/thread context."""
116 if task_context := get_current_task_context():
117 task_context.set_report(markdown)
118