File size: 3,318 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 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 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """Module for runtime."""
from __future__ import annotations
import logging
import time
from collections.abc import Callable
from dataclasses import dataclass
from django.conf import settings
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class TaskRunContext:
started_monotonic: float
soft_deadline_monotonic: float
timeout_seconds: float
@classmethod
def from_django_q(
cls,
*,
timeout_seconds: float | None = None,
safety_margin_seconds: float = 30.0,
min_soft_deadline_seconds: float = 60.0,
) -> TaskRunContext:
start = time.monotonic()
q_cluster = getattr(settings, "Q_CLUSTER", {}) or {}
effective_timeout = timeout_seconds
if effective_timeout is None:
try:
effective_timeout = float(q_cluster.get("timeout") or 600)
except (TypeError, ValueError, KeyError):
effective_timeout = 600.0
soft = start + max(min_soft_deadline_seconds, float(effective_timeout) - float(safety_margin_seconds))
return cls(started_monotonic=start, soft_deadline_monotonic=soft, timeout_seconds=float(effective_timeout))
def is_past_soft_deadline(self) -> bool:
return time.monotonic() > float(self.soft_deadline_monotonic)
class CancellationToken:
def __init__(self, should_cancel: Callable[[], bool]) -> None:
self._should_cancel = should_cancel
def is_cancelled(self) -> bool:
try:
return bool(self._should_cancel())
except (TypeError, AttributeError):
return False
class ProgressReporter:
def __init__(
self,
*,
update_fn: Callable[[int, int, int, str], None],
min_interval_seconds: float = 0.5,
) -> None:
self._update_fn = update_fn
self._min_interval_seconds = float(min_interval_seconds)
self._last_update_ts = 0.0
self._last_progress: int | None = None
self._last_message: str | None = None
def report(self, *, current: int, total: int, message: str, force: bool = False) -> None:
try:
progress = int(current * 100 / total) if total else 0
except (ZeroDivisionError, TypeError, ValueError):
progress = 0
progress = min(max(progress, 0), 100)
now_ts = time.time()
if not force:
if self._last_progress == progress and self._last_message == message:
if now_ts - self._last_update_ts < self._min_interval_seconds:
return
elif now_ts - self._last_update_ts < self._min_interval_seconds:
return
self._update_fn(progress, int(current), int(total), str(message or ""))
self._last_update_ts = now_ts
self._last_progress = progress
self._last_message = message
def report_extra(self, *, progress: int, current: int, total: int, message: str, force: bool = False) -> None:
now_ts = time.time()
if not force and now_ts - self._last_update_ts < self._min_interval_seconds:
return
self._update_fn(int(progress), int(current), int(total), str(message or ""))
self._last_update_ts = now_ts
self._last_progress = int(progress)
self._last_message = message
|