diff --git a/qutebrowser/utils/qtutils.py b/qutebrowser/utils/qtutils.py index d2b7f2ffc..5373e76aa 100644 --- a/qutebrowser/utils/qtutils.py +++ b/qutebrowser/utils/qtutils.py @@ -31,11 +31,12 @@ Module attributes: import io import operator import contextlib +import typing # pylint: disable=unused-import,useless-suppression import pkg_resources from PyQt5.QtCore import (qVersion, QEventLoop, QDataStream, QByteArray, QIODevice, QSaveFile, QT_VERSION_STR, - PYQT_VERSION_STR) + PYQT_VERSION_STR, QFileDevice, QObject) try: from PyQt5.QtWebKit import qWebKitVersion except ImportError: # pragma: no cover @@ -61,19 +62,22 @@ class QtOSError(OSError): qt_errno: The error attribute of the given QFileDevice, if applicable. """ - def __init__(self, dev, msg=None): + def __init__(self, dev: QFileDevice, msg: str = None) -> None: if msg is None: msg = dev.errorString() super().__init__(msg) + self.qt_errno = None # type: typing.Optional[QFileDevice.FileError] try: self.qt_errno = dev.error() except AttributeError: - self.qt_errno = None + pass -def version_check(version, exact=False, compiled=True): +def version_check(version: str, + exact: bool = False, + compiled: bool = True) -> bool: """Check if the Qt runtime version is the version supplied or newer. Args: @@ -103,14 +107,14 @@ def version_check(version, exact=False, compiled=True): MAX_WORLD_ID = 256 if version_check('5.11.2') else 11 -def is_new_qtwebkit(): +def is_new_qtwebkit() -> bool: """Check if the given version is a new QtWebKit.""" assert qWebKitVersion is not None return (pkg_resources.parse_version(qWebKitVersion()) > pkg_resources.parse_version('538.1')) -def check_overflow(arg, ctype, fatal=True): +def check_overflow(arg: int, ctype: str, fatal: bool = True) -> int: """Check if the given argument is in bounds for the given type. Args: @@ -138,13 +142,13 @@ def check_overflow(arg, ctype, fatal=True): return arg -def ensure_valid(obj): +def ensure_valid(obj: QObject) -> None: """Ensure a Qt object with an .isValid() method is valid.""" if not obj.isValid(): raise QtValueError(obj) -def check_qdatastream(stream): +def check_qdatastream(stream: QDataStream) -> None: """Check the status of a QDataStream and raise OSError if it's not ok.""" status_to_str = { QDataStream.Ok: "The data stream is operating normally.", @@ -158,7 +162,7 @@ def check_qdatastream(stream): raise OSError(status_to_str[stream.status()]) -def serialize(obj): +def serialize(obj: QObject) -> QByteArray: """Serialize an object into a QByteArray.""" data = QByteArray() stream = QDataStream(data, QIODevice.WriteOnly) @@ -166,20 +170,20 @@ def serialize(obj): return data -def deserialize(data, obj): +def deserialize(data: QByteArray, obj: QObject) -> None: """Deserialize an object from a QByteArray.""" stream = QDataStream(data, QIODevice.ReadOnly) deserialize_stream(stream, obj) -def serialize_stream(stream, obj): +def serialize_stream(stream: QDataStream, obj: QObject) -> None: """Serialize an object into a QDataStream.""" check_qdatastream(stream) stream << obj # pylint: disable=pointless-statement check_qdatastream(stream) -def deserialize_stream(stream, obj): +def deserialize_stream(stream: QDataStream, obj: QObject) -> None: """Deserialize a QDataStream into an object.""" check_qdatastream(stream) stream >> obj # pylint: disable=pointless-statement @@ -195,11 +199,14 @@ def savefile_open(filename, binary=False, encoding='utf-8'): open_ok = f.open(QIODevice.WriteOnly) if not open_ok: raise QtOSError(f) + if binary: new_f = PyQIODevice(f) else: new_f = io.TextIOWrapper(PyQIODevice(f), encoding=encoding) + yield new_f + new_f.flush() except: f.cancelWriting() @@ -219,29 +226,29 @@ class PyQIODevice(io.BufferedIOBase): dev: The underlying QIODevice. """ - def __init__(self, dev): + def __init__(self, dev: QIODevice) -> None: super().__init__() self.dev = dev - def __len__(self): + def __len__(self) -> int: return self.dev.size() - def _check_open(self): + def _check_open(self) -> None: """Check if the device is open, raise ValueError if not.""" if not self.dev.isOpen(): raise ValueError("IO operation on closed device!") - def _check_random(self): + def _check_random(self) -> None: """Check if the device supports random access, raise OSError if not.""" if not self.seekable(): raise OSError("Random access not allowed!") - def _check_readable(self): + def _check_readable(self) -> None: """Check if the device is readable, raise OSError if not.""" if not self.dev.isReadable(): raise OSError("Trying to read unreadable file!") - def _check_writable(self): + def _check_writable(self) -> None: """Check if the device is writable, raise OSError if not.""" if not self.writable(): raise OSError("Trying to write to unwritable file!") @@ -263,7 +270,7 @@ class PyQIODevice(io.BufferedIOBase): raise QtOSError(self.dev) return contextlib.closing(self) - def close(self): + def close(self) -> None: """Close the underlying device.""" self.dev.close() @@ -289,18 +296,18 @@ class PyQIODevice(io.BufferedIOBase): raise io.UnsupportedOperation @property - def closed(self): + def closed(self) -> bool: return not self.dev.isOpen() - def flush(self): + def flush(self) -> None: self._check_open() self.dev.waitForBytesWritten(-1) - def isatty(self): + def isatty(self) -> bool: self._check_open() return False - def readable(self): + def readable(self) -> bool: return self.dev.isReadable() def readline(self, size=-1): @@ -326,18 +333,18 @@ class PyQIODevice(io.BufferedIOBase): raise QtOSError(self.dev) return buf - def seekable(self): + def seekable(self) -> bool: return not self.dev.isSequential() - def tell(self): + def tell(self) -> int: self._check_open() self._check_random() return self.dev.pos() - def writable(self): + def writable(self) -> bool: return self.dev.isWritable() - def write(self, b): + def write(self, b: bytes) -> int: self._check_open() self._check_writable() num = self.dev.write(b) @@ -361,7 +368,7 @@ class QtValueError(ValueError): """Exception which gets raised by ensure_valid.""" - def __init__(self, obj): + def __init__(self, obj: QObject) -> None: try: self.reason = obj.errorString() except AttributeError: @@ -379,7 +386,7 @@ class EventLoop(QEventLoop): Raises an exception when doing exec_() multiple times. """ - def __init__(self, parent=None): + def __init__(self, parent: QObject = None) -> None: super().__init__(parent) self._executing = False @@ -388,5 +395,6 @@ class EventLoop(QEventLoop): if self._executing: raise AssertionError("Eventloop is already running!") self._executing = True - super().exec_(flags) + status = super().exec_(flags) self._executing = False + return status