File size: 2,417 Bytes
f8c73f9 | 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 | """Coalesced background reads; widget updates always return to the Qt thread."""
from __future__ import annotations
import time
from typing import Callable
from PySide6.QtCore import QObject, QRunnable, QThreadPool, Signal, Slot
class _Result(QObject):
finished = Signal(object, object)
class _Read(QRunnable):
def __init__(self, read: Callable, result: _Result) -> None:
super().__init__()
self.read = read
self.result = result
def run(self) -> None:
try:
value = self.read()
except Exception as exc:
self._emit(None, str(exc))
else:
self._emit(value, None)
def _emit(self, value: object, error: object) -> None:
"""Ignore a late completion after Qt has already torn down the window."""
try:
self.result.finished.emit(value, error)
except RuntimeError:
# A screenshot/normal application shutdown may delete the receiver
# while this harmless read is still completing.
pass
class BackgroundRead(QObject):
ready = Signal(object)
failed = Signal(str)
def __init__(self, parent: QObject, *, max_age: float = 30) -> None:
super().__init__(parent)
self.max_age = max_age
self._completed_at: float | None = None
self._running = False
self._pending: Callable | None = None
# The runnable retains this object if its page closes during a read.
self._result = _Result()
self._result.finished.connect(self._finished)
def request(self, read: Callable, *, force: bool = False) -> None:
if self._running:
if force:
self._pending = read
return
if not force and self._completed_at is not None and time.monotonic() - self._completed_at < self.max_age:
return
self._running = True
QThreadPool.globalInstance().start(_Read(read, self._result))
@Slot(object, object)
def _finished(self, value: object, error: object) -> None:
self._running = False
if self._pending is not None:
read, self._pending = self._pending, None
self.request(read, force=True)
return
if error is not None:
self.failed.emit(str(error))
else:
self._completed_at = time.monotonic()
self.ready.emit(value)
|