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
def get_tracker_key(self) -> str:
    raise NotImplementedError

progress_error

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

update_progress

Source code in django_spire/contrib/progress/mixins.py
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
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
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
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
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
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
@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
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
@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
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
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
def clear(self) -> None:
    cache.delete(self._key)

complete

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

parallel

Source code in django_spire/contrib/progress/tracker.py
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
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
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
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)