Add partial type annotations for qtutils

This commit is contained in:
Florian Bruhin 2018-11-29 13:29:40 +01:00
parent 7cbba4b3f1
commit e576847b91

View File

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