Skip to content

runner

django_spire.contrib.progress.runner

TaskProgressUpdater

Source code in django_spire/contrib/progress/runner.py 
20
21
22
23
def __init__(self, tracker: ProgressTracker, task: Task) -> None:
    self._messages = ProgressMessages()
    self._task = task
    self._tracker = tracker

complete

Source code in django_spire/contrib/progress/runner.py 
25
26
27
28
29
30
31
def complete(self, message: str | None = None) -> None:
    self._tracker._update_task(
        self._task.name,
        ProgressStatus.COMPLETE,
        self._format(message or self._messages.complete),
        100
    )

error

Source code in django_spire/contrib/progress/runner.py 
33
34
35
36
37
38
39
def error(self, message: str) -> None:
    self._tracker._update_task(
        self._task.name,
        ProgressStatus.ERROR,
        self._format(message),
        0
    )

start

Source code in django_spire/contrib/progress/runner.py 
41
42
43
44
45
46
47
def start(self) -> None:
    self._tracker._update_task(
        self._task.name,
        ProgressStatus.PROCESSING,
        self._format(self._messages.starting),
        2
    )

update

Source code in django_spire/contrib/progress/runner.py 
49
50
51
52
53
54
55
def update(self, message: str, progress: int) -> None:
    self._tracker._update_task(
        self._task.name,
        ProgressStatus.PROCESSING,
        self._format(message),
        progress
    )

ProgressSimulator

Source code in django_spire/contrib/progress/runner.py 
62
63
64
65
66
67
68
69
70
71
def __init__(
    self,
    updater: TaskProgressUpdater,
    max_progress: int = 90,
    update_interval: float = 0.15
) -> None:
    self._max_progress = max_progress
    self._messages = ProgressMessages()
    self._update_interval = update_interval
    self._updater = updater

run

Source code in django_spire/contrib/progress/runner.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def run(self, stop_event: threading.Event) -> None:
    start_time = time.time()
    duration = random.uniform(8.0, 15.0)

    while not stop_event.is_set():
        elapsed = time.time() - start_time
        t = min(elapsed / duration, 1.0)

        progress = self._ease_out_expo(t) * self._max_progress
        progress = min(int(progress), self._max_progress)

        jitter = random.uniform(-1.5, 1.5) if progress < self._max_progress - 5 else 0
        progress = max(2, min(int(progress + jitter), self._max_progress))

        message = self._get_message_for_progress(progress)
        self._updater.update(message, progress)

        if progress >= self._max_progress:
            break

        time.sleep(self._update_interval + random.uniform(0, 0.1))

TaskRunner

Source code in django_spire/contrib/progress/runner.py 
109
110
111
112
113
def __init__(self, tracker: ProgressTracker, task: Task) -> None:
    self._stop_event = threading.Event()
    self._task = task
    self._tracker = tracker
    self._updater = TaskProgressUpdater(tracker, task)

run_parallel

Source code in django_spire/contrib/progress/runner.py 
115
116
def run_parallel(self, results: dict[str, TaskResult]) -> None:
    self._execute(results, lambda: self._task.func())

run_sequential

Source code in django_spire/contrib/progress/runner.py 
118
119
120
def run_sequential(self, results: dict[str, TaskResult]) -> None:
    unwrapped = {name: result.value for name, result in results.items()}
    self._execute(results, lambda: self._task.func(unwrapped))

stop

Source code in django_spire/contrib/progress/runner.py 
122
123
def stop(self) -> None:
    self._stop_event.set()