Source code for qpageview.backgroundjob

# -*- coding: utf-8 -*-
# This file is part of the qpageview package.
# Copyright (c) 2019 - 2019 by Wilbert Berendsen
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
# See for more information.

Run jobs in the background using QThread.

from PyQt5.QtCore import QThread

# as soon as a Job start()s, it is saved here, to prevent it being
# destroyed while running
_runningJobs = set()
_pendingJobs = []
maxjobs = 12

[docs]class Job(QThread): """A simple wrapper around QThread. Before calling start() you should put the work function in the work attribute, and an optional finalize function (which will be called with the result) in the finalize attribute. Or alternatively, inherit from this class and implement work() and finish() yourself. The result of the work function is stored in the result attribute. """ finalize = None running = False done = False result = None def __init__(self, parent=None): """Init ourselves; the parent can be a QObject which will be our parent.""" super().__init__(parent) self.finished.connect(self._slotFinished)
[docs] def start(self): self.result = None self.running = True # this is more robust than isRunning() self.done = False if len(_runningJobs) < maxjobs: _runningJobs.add(self) super().start() else: _pendingJobs.append(self)
[docs] def run(self): """Call the work function in the background thread.""" self.result =
[docs] def work(self): """Implement this to get the work done. If you have long tasks you can Qt's isInterruptionRequested() functionality. Instead of implementing this method, you can put the work function in the work instance attribute. """ pass
[docs] def finish(self): """This slot is called in the main thread when the work is done. The default implementation calls the finalize function with the result. """ if self.finalize: self.finalize(self.result)
def _slotFinished(self): _runningJobs.discard(self) if _pendingJobs: _pendingJobs.pop().start() self.running = False self.done = True self.finish()
[docs]class SingleRun: """Run a function in a background thread. The outcome is silently discarded if another function is called before the old one finishes. """ def __init__(self): self._job = None
[docs] def cancel(self): """Forgets the running job. The job is not terminated but the callback is not called. """ j = self._job if j: j.finalize = None self._job = None
def __call__(self, func, callback=None): self.cancel() j = self._job = Job() = func def finalize(result): self.cancel() if callback: callback(result) j.finalize = finalize j.start()
[docs]def run(func, callback=None): """Run specified function in a background thread. The thread is immediately started. If a callback is specified, it is called in the main thread with the result when the function is ready. """ j = Job() = func j.finalize = callback j.start()