Skip to content

tracker

django_spire.celery.tracker

CeleryTaskTracker

Used for tracking the state of a celery task inside the running task function

Source code in django_spire/celery/tracker.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(self, celery_task: Task, update_interval_seconds: int = 5) -> None:
    if update_interval_seconds < 5:
        message = f'{self.__class__.__name__}: Update Interval must be at least 5 seconds'
        raise ValueError(message)

    self._celery_task = celery_task
    self._update_interval_seconds = update_interval_seconds
    self._start_time_seconds = time.time()
    self._last_update_time_seconds = 0
    self._state = states.PENDING
    self._meta = CeleryTaskMeta()
    self._additional_meta = None
    self._cumulative_progress = 0
    self._pending_future: Future | None = None
    self._cumulative_target_value: int | None = None

task property

set_cumulative_progress_target_value

Source code in django_spire/celery/tracker.py 
70
71
def set_cumulative_progress_target_value(self, value: int) -> None:
    self._cumulative_target_value = value

update_state

Source code in django_spire/celery/tracker.py 
87
88
89
90
91
def update_state(self, state: str = states.PENDING, meta: dict | None = None) -> None:
    self._state = state.upper()
    self._additional_meta = meta

    self._process_overdue_update()

update_count_progress

Source code in django_spire/celery/tracker.py 
 93
 94
 95
 96
 97
 98
 99
100
101
def update_count_progress(
    self, current_count: int, target_count: int, range_min: float = 0.0, range_max: float = 1.0
) -> None:
    if range_min < 0.0 or range_min > range_max or range_max > 1.0:
        message = 'Progress range is invalid'
        raise ValueError(message)

    self._meta.progress = range_min + (range_max - range_min) * (current_count / target_count)
    self._process_overdue_update()

update_cumulative_progress

Source code in django_spire/celery/tracker.py 
103
104
105
106
107
108
109
110
111
def update_cumulative_progress(self, added_value: int) -> None:
    self._cumulative_progress += added_value

    if self._cumulative_target_value is None:
        message = f'{self.__class__.__name__}: Cumulative Progress Target Value is None'
        raise ValueError(message)

    self._meta.progress = self._cumulative_progress / self._cumulative_target_value
    self._process_overdue_update()

set_completed

Source code in django_spire/celery/tracker.py 
113
114
115
def set_completed(self):
    self._meta.set_completed()
    self._flush_futures()

set_started

Source code in django_spire/celery/tracker.py 
117
118
119
def set_started(self):
    self._meta.set_started()
    self.update_state(states.STARTED)

set_started_and_completing_soon

Source code in django_spire/celery/tracker.py 
121
122
123
def set_started_and_completing_soon(self):
    self._meta.set_started_and_completing_soon()
    self.update_state(states.STARTED)