session

django_spire.contrib.progress.session

SIMULATION_MESSAGES = ['Initializing...', 'Processing...', 'Analyzing...', 'Generating...', 'Finalizing...'] module-attribute

ProgressSession

Source code in django_spire/contrib/progress/session.py

def __init__(self, session_id: str, tasks: dict[str, str]) -> None:
    self._lock = threading.Lock()
    self._simulation_threads: dict[str, tuple[threading.Event, threading.Thread]] = {}

    self._tasks: dict[str, dict[str, Any]] = {
        task_id: {
            'complete_message': '',
            'message': 'Waiting...',
            'name': name,
            'percent': 0.0,
            'status': ProgressStatus.PENDING,
        }
        for task_id, name in tasks.items()
    }

    self.session_id = session_id

session_id = session_id instance-attribute

has_error property

is_complete property

is_running property

overall_percent property

status property

add_parallel

Source code in django_spire/contrib/progress/session.py

def add_parallel(self, task_id: str, future: Any) -> ParallelTask:
    return ParallelTask(self, task_id, future)

add_sequential

Source code in django_spire/contrib/progress/session.py

def add_sequential(self, task_id: str, func: Callable, *args: Any, **kwargs: Any) -> SequentialTask:
    return SequentialTask(self, task_id, func, *args, **kwargs)

complete

Source code in django_spire/contrib/progress/session.py

def complete(self, task_id: str, message: str | None = None) -> None:
    with self._lock:
        if task_id in self._tasks:
            task = self._tasks[task_id]
            task['status'] = ProgressStatus.COMPLETING
            task['complete_message'] = message or f'{task["name"]} complete'
            task['message'] = 'Completing...'
            self._save()

error

Source code in django_spire/contrib/progress/session.py

def error(self, task_id: str, message: str | None = None) -> None:
    if task_id in self._simulation_threads:
        stop_event, _ = self._simulation_threads[task_id]
        stop_event.set()
        del self._simulation_threads[task_id]

    with self._lock:
        if task_id in self._tasks:
            task = self._tasks[task_id]
            task['status'] = ProgressStatus.ERROR
            task['message'] = message or f'{task["name"]} failed'
            self._save()

start

Source code in django_spire/contrib/progress/session.py

def start(self, task_id: str) -> None:
    with self._lock:
        if task_id in self._tasks:
            task = self._tasks[task_id]
            task['percent'] = 0.0
            task['status'] = ProgressStatus.RUNNING
            task['message'] = SIMULATION_MESSAGES[0]
            self._save()

    stop_event = threading.Event()

    thread = threading.Thread(
        target=self._simulate_progress,
        args=(task_id, stop_event),
        daemon=True,
    )

    self._simulation_threads[task_id] = (stop_event, thread)
    thread.start()

stream

Source code in django_spire/contrib/progress/session.py

def stream(self, poll_interval: float = 0.1) -> Generator[str, None, None]:
    with self._lock:
        data = self.to_dict()

    yield f'{json.dumps(data)}\n'

    while True:
        time.sleep(poll_interval)

        with self._lock:
            data = self.to_dict()

        yield f'{json.dumps(data)}\n'

        if self.is_complete or self.has_error:
            self._delete()
            break

to_dict

Source code in django_spire/contrib/progress/session.py

def to_dict(self) -> dict[str, Any]:
    return {
        'overall_percent': self.overall_percent,
        'session_id': self.session_id,
        'status': self.status.value,
        'tasks': {
            task_id: {
                'message': task['message'],
                'name': task['name'],
                'percent': int(task['percent']),
                'status': task['status'].value if task['status'] != ProgressStatus.COMPLETING else ProgressStatus.RUNNING.value,
            }
            for task_id, task in self._tasks.items()
        },
    }

create classmethod

Source code in django_spire/contrib/progress/session.py

@classmethod
def create(cls, tasks: dict[str, str]) -> ProgressSession:
    session_id = str(uuid.uuid4())
    session = cls(session_id, tasks)
    session._save()
    return session

get classmethod

Source code in django_spire/contrib/progress/session.py

@classmethod
def get(cls, session_id: str) -> ProgressSession | None:
    data = cache.get(f'{cls._CACHE_PREFIX}{session_id}')

    if data is None:
        return None

    session = cls(data['session_id'], {})

    session._tasks = {
        task_id: {
            'complete_message': task['complete_message'],
            'message': task['message'],
            'name': task['name'],
            'percent': task['percent'],
            'status': ProgressStatus(task['status']),
        }
        for task_id, task in data['tasks'].items()
    }

    return session