Skip to content

progress

django_spire.contrib.progress

__all__ = ['ProgressMessages', 'ProgressStatus', 'ProgressTracker', 'ProgressTrackingMixin', 'Task', 'TaskProgressUpdater', 'TaskResult', 'TaskState', 'TrackerState'] module-attribute

ProgressStatus

Bases: StrEnum

COMPLETE = 'complete' class-attribute instance-attribute

ERROR = 'error' class-attribute instance-attribute

PENDING = 'pending' class-attribute instance-attribute

PROCESSING = 'processing' class-attribute instance-attribute

ProgressTrackingMixin

tracker property

get_tracker_key

Source code in django_spire/contrib/progress/mixins.py 
16
17
def get_tracker_key(self) -> str:
    raise NotImplementedError

progress_error

Source code in django_spire/contrib/progress/mixins.py 
26
27
def progress_error(self, message: str) -> None:
    self.tracker.error(message)

update_progress

Source code in django_spire/contrib/progress/mixins.py 
29
30
31
32
33
34
35
36
def update_progress(
    self,
    status: ProgressStatus,
    message: str,
    progress: int,
    **kwargs: Any
) -> None:
    self.tracker.update(status, message, progress, **kwargs)

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
    )

TaskState dataclass

message = 'Waiting...' class-attribute instance-attribute

progress = 0 class-attribute instance-attribute

status = ProgressStatus.PENDING class-attribute instance-attribute

from_dict classmethod

Source code in django_spire/contrib/progress/states.py 
18
19
20
21
22
23
24
@classmethod
def from_dict(cls, data: dict[str, Any]) -> TaskState:
    return cls(
        message=data.get('message', 'Waiting...'),
        progress=data.get('progress', 0),
        status=ProgressStatus(data.get('step', ProgressStatus.PENDING)),
    )

to_dict

Source code in django_spire/contrib/progress/states.py 
26
27
28
29
30
31
def to_dict(self) -> dict[str, Any]:
    return {
        'message': self.message,
        'progress': self.progress,
        'step': self.status.value,
    }

TrackerState dataclass

message = 'Initializing...' class-attribute instance-attribute

progress = 0 class-attribute instance-attribute

status = ProgressStatus.PENDING class-attribute instance-attribute

task_order = field(default_factory=list) class-attribute instance-attribute

tasks = field(default_factory=dict) class-attribute instance-attribute

from_dict classmethod

Source code in django_spire/contrib/progress/states.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@classmethod
def from_dict(cls, data: dict[str, Any]) -> TrackerState:
    tasks = {
        name: TaskState.from_dict(task_data)
        for name, task_data in data.get('tasks', {}).items()
    }

    return cls(
        message=data.get('message', 'Initializing...'),
        progress=data.get('progress', 0),
        status=ProgressStatus(data.get('step', ProgressStatus.PENDING)),
        task_order=data.get('task_order', []),
        tasks=tasks,
    )

to_dict

Source code in django_spire/contrib/progress/states.py 
57
58
59
60
61
62
63
64
def to_dict(self) -> dict[str, Any]:
    return {
        'message': self.message,
        'progress': self.progress,
        'step': self.status.value,
        'task_order': self.task_order,
        'tasks': {name: task.to_dict() for name, task in self.tasks.items()},
    }

ProgressMessages dataclass

complete = 'Complete' class-attribute instance-attribute

error = 'Error' class-attribute instance-attribute

starting = 'Starting...' class-attribute instance-attribute

steps = field(default_factory=(lambda: ['Initializing...', 'Processing...', 'Analyzing...', 'Working...', 'Almost there...', 'Finalizing...'])) class-attribute instance-attribute

Task dataclass

func instance-attribute

label instance-attribute

name instance-attribute

parallel = False class-attribute instance-attribute

TaskResult dataclass

error = None class-attribute instance-attribute

value = None class-attribute instance-attribute

failed property

ProgressTracker

Source code in django_spire/contrib/progress/tracker.py 
24
25
26
27
28
def __init__(self, key: str, timeout: int = 300) -> None:
    self._key = f'progress_tracker_{key}'
    self._lock = threading.Lock()
    self._tasks: list[Task] = []
    self._timeout = timeout

clear

Source code in django_spire/contrib/progress/tracker.py 
30
31
def clear(self) -> None:
    cache.delete(self._key)

complete

Source code in django_spire/contrib/progress/tracker.py 
33
34
35
36
37
38
39
def complete(self, message: str = 'Complete!') -> None:
    with self._lock:
        state = self._get_state()
        state.message = message
        state.progress = 100
        state.status = ProgressStatus.COMPLETE
        self._save_state(state)

error

Source code in django_spire/contrib/progress/tracker.py 
41
42
43
44
45
46
47
def error(self, message: str) -> None:
    with self._lock:
        state = self._get_state()
        state.message = message
        state.progress = 0
        state.status = ProgressStatus.ERROR
        self._save_state(state)

execute

Source code in django_spire/contrib/progress/tracker.py 
49
50
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
def execute(self) -> dict[str, Any] | None:
    self._create_state()

    error = False
    results: dict[str, TaskResult] = {}

    try:
        error = self._execute_parallel_tasks(results)

        if not error:
            error = self._execute_sequential_tasks(results)

        if not error:
            self.complete()
            time.sleep(1)
    except BaseException:
        log.exception('Progress tracker failed')
        error = True

    if error:
        self.error('An unexpected error occurred')
        time.sleep(2)

    self.clear()

    if error:
        return None

    return {name: result.value for name, result in results.items()}

get

Source code in django_spire/contrib/progress/tracker.py 
79
80
def get(self) -> dict[str, Any] | None:
    return cache.get(self._key)

parallel

Source code in django_spire/contrib/progress/tracker.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def parallel(
    self,
    name: str,
    func: Callable[[], Any],
    label: str | None = None
) -> ProgressTracker:
    task = Task(
        func=func,
        label=label or self._generate_label(name),
        name=name,
        parallel=True,
    )

    self._tasks.append(task)

    return self

sequential

Source code in django_spire/contrib/progress/tracker.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def sequential(
    self,
    name: str,
    func: Callable[[dict[str, Any]], Any],
    label: str | None = None
) -> ProgressTracker:
    task = Task(
        func=func,
        label=label or self._generate_label(name),
        name=name,
        parallel=False,
    )

    self._tasks.append(task)

    return self

start

Source code in django_spire/contrib/progress/tracker.py 
116
117
118
def start(self) -> None:
    state = TrackerState()
    cache.set(self._key, state.to_dict(), timeout=self._timeout)

update

Source code in django_spire/contrib/progress/tracker.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
def update(
    self,
    status: ProgressStatus,
    message: str,
    progress: int,
    **kwargs: Any
) -> None:
    with self._lock:
        state = self._get_state()
        state.message = message
        state.progress = progress
        state.status = status
        self._save_state(state, **kwargs)