Skip to content

meta

django_spire.celery.meta

CeleryTaskMeta

Bases: BaseModel

model_config = ConfigDict(extra='allow') class-attribute instance-attribute

progress = None class-attribute instance-attribute

started_time = None class-attribute instance-attribute

last_update_time = None class-attribute instance-attribute

estimated_completed_time = None class-attribute instance-attribute

completed_time = None class-attribute instance-attribute

completed_datetime property

estimated_progress property

estimated_progress_per_second property

estimated_remaining_seconds property

estimated_run_time_seconds property

run_time_seconds property

started_datetime property

__setattr__

Source code in django_spire/celery/meta.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __setattr__(self, name: str, value: Any) -> None:
    if name == 'progress' and isinstance(value, float):
        if self.started_time is None:
            self.started_time = time.time()

        if self._progress_updates_count >= 2:
            current_time = time.time()
            elapsed_seconds = current_time - self.started_time
            estimated_total_seconds = elapsed_seconds / value
            remaining_seconds = estimated_total_seconds - elapsed_seconds

            new_estimated_end_time_seconds = current_time + (
                remaining_seconds * (_ESTIMATED_TIME_BUFFER + 1.0)
            )

            if self.estimated_completed_time is None:
                self.estimated_completed_time = new_estimated_end_time_seconds

            else:
                difference_seconds = (
                    self.estimated_completed_time - new_estimated_end_time_seconds
                )
                change_threshold = _ESTIMATED_TIME_BUFFER * self.estimated_run_time_seconds

                if abs(difference_seconds) > abs(change_threshold):
                    self.estimated_completed_time = new_estimated_end_time_seconds

        else:
            self._progress_updates_count += 1

        self.last_update_time = time.time()

    super().__setattr__(name, value)

set_completed

Source code in django_spire/celery/meta.py 
106
107
108
def set_completed(self) -> None:
    self.progress = 1.0
    self.completed_time = time.time()

set_started

Source code in django_spire/celery/meta.py 
110
111
112
def set_started(self) -> None:
    self.progress = 0.02
    self.started_time = time.time()

set_started_and_completing_soon

Source code in django_spire/celery/meta.py 
114
115
116
117
118
def set_started_and_completing_soon(self) -> None:
    self.set_started()
    self.progress = 1.0
    self.estimated_completed_time = time.time() + 5
    self.last_update_time = time.time() + 1