added more type hints, full pyright and mypy support and added py.typed file

Author: wolph

.github/workflows/main.yml

@@ -32,7 +32,7 @@ jobs:
       - name: pytest
         run: py.test
 
-  docs:
+  docs_and_lint:
     runs-on: ubuntu-latest
     timeout-minutes: 2
     steps:
@@ -44,7 +44,13 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip setuptools
-        pip install -e '.[docs]'
+        pip install -e '.[docs,tests]' pyright flake8 mypy
     - name: build docs
       run: make html
       working-directory: docs/
+    - name: flake8
+      run: flake8 -v python_utils setup.py
+    - name: mypy
+      run: mypy python_utils setup.py
+    - name: pyright
+      run: pyright

_python_utils_tests/test_time.py

@@ -71,6 +71,9 @@ def test_timeout_generator(
 
 @pytest.mark.asyncio
 async def test_aio_generator_timeout_detector():
+    # Make pyright happy
+    i = None
+
     async def generator():
         for i in range(10):
             await asyncio.sleep(i / 100.0)
@@ -106,52 +109,55 @@ async def generator():
 
 @pytest.mark.asyncio
 async def test_aio_generator_timeout_detector_decorator():
+    # Make pyright happy
+    i = None
+
     # Test regular timeout with reraise
     @python_utils.aio_generator_timeout_detector_decorator(timeout=0.05)
-    async def generator():
+    async def generator_timeout():
         for i in range(10):
             await asyncio.sleep(i / 100.0)
             yield i
 
     with pytest.raises(asyncio.TimeoutError):
-        async for i in generator():
+        async for i in generator_timeout():
             pass
 
     # Test regular timeout with clean exit
     @python_utils.aio_generator_timeout_detector_decorator(
         timeout=0.05, on_timeout=None
     )
-    async def generator():
+    async def generator_clean():
         for i in range(10):
             await asyncio.sleep(i / 100.0)
             yield i
 
-    async for i in generator():
+    async for i in generator_clean():
         pass
 
     assert i == 4
 
     # Test total timeout with reraise
     @python_utils.aio_generator_timeout_detector_decorator(total_timeout=0.1)
-    async def generator():
+    async def generator_reraise():
         for i in range(10):
             await asyncio.sleep(i / 100.0)
             yield i
 
     with pytest.raises(asyncio.TimeoutError):
-        async for i in generator():
+        async for i in generator_reraise():
             pass
 
     # Test total timeout with clean exit
     @python_utils.aio_generator_timeout_detector_decorator(
         total_timeout=0.1, on_timeout=None
     )
-    async def generator():
+    async def generator_clean_total():
         for i in range(10):
             await asyncio.sleep(i / 100.0)
             yield i
 
-    async for i in generator():
+    async for i in generator_clean_total():
         pass
 
     assert i == 4

pyrightconfig.json

@@ -0,0 +1,8 @@
+{
+  "include": [
+    "python_utils"
+  ],
+  "exclude": [
+    "python_utils/types.py",
+  ],
+}

python_utils/aio.py

@@ -5,8 +5,15 @@
 import asyncio
 import itertools
 
+from . import types
 
-async def acount(start=0, step=1, delay=0, stop=None):
+
+async def acount(
+    start: float = 0,
+    step: float = 1,
+    delay: float = 0,
+    stop: types.Optional[float] = None,
+) -> types.AsyncIterator[float]:
     '''Asyncio version of itertools.count()'''
     for item in itertools.count(start, step):  # pragma: no branch
         if stop is not None and item >= stop:

python_utils/containers.py

@@ -31,8 +31,8 @@ def __init__(
         self,
         key_cast: KT_cast = None,
         value_cast: VT_cast = None,
-        *args,
-        **kwargs,
+        *args: DictUpdateArgs,
+        **kwargs: VT,
     ) -> None:
         self._value_cast = value_cast
         self._key_cast = key_cast
@@ -53,7 +53,7 @@ def __setitem__(self, key: Any, value: Any) -> None:
         return super().__setitem__(key, value)
 
 
-class CastedDict(CastedDictBase):
+class CastedDict(CastedDictBase[KT, VT]):
     '''
     Custom dictionary that casts keys and values to the specified typing.
 
@@ -88,14 +88,14 @@ class CastedDict(CastedDictBase):
     {1: 2, '3': '4', '5': '6', '7': '8'}
     '''
 
-    def __setitem__(self, key, value):
+    def __setitem__(self, key: Any, value: Any) -> None:
         if self._value_cast is not None:
             value = self._value_cast(value)
 
         super().__setitem__(key, value)
 
 
-class LazyCastedDict(CastedDictBase):
+class LazyCastedDict(CastedDictBase[KT, VT]):
     '''
     Custom dictionary that casts keys and lazily casts values to the specified
     typing. Note that the values are cast only when they are accessed and
@@ -141,13 +141,13 @@ class LazyCastedDict(CastedDictBase):
     '4'
     '''
 
-    def __setitem__(self, key, value):
+    def __setitem__(self, key: Any, value: Any) -> None:
         if self._key_cast is not None:
             key = self._key_cast(key)
 
         super().__setitem__(key, value)
 
-    def __getitem__(self, key) -> VT:
+    def __getitem__(self, key: Any) -> VT:
         if self._key_cast is not None:
             key = self._key_cast(key)
 

python_utils/converters.py

@@ -10,7 +10,7 @@ def to_int(
     input_: typing.Optional[str] = None,
     default: int = 0,
     exception: types.ExceptionsType = (ValueError, TypeError),
-    regexp: types.O[types.Pattern] = None,
+    regexp: types.Optional[types.Pattern] = None,
 ) -> int:
     r'''
     Convert the given input to an integer or return default
@@ -100,7 +100,7 @@ def to_float(
     input_: str,
     default: int = 0,
     exception: types.ExceptionsType = (ValueError, TypeError),
-    regexp: types.O[types.Pattern] = None,
+    regexp: types.Optional[types.Pattern] = None,
 ) -> types.Number:
     r'''
     Convert the given `input_` to an integer or return default

python_utils/generators.py

@@ -9,7 +9,7 @@ async def abatcher(
     generator: types.AsyncIterator,
     batch_size: types.Optional[int] = None,
     interval: types.Optional[types.delta_type] = None,
-):
+) -> types.AsyncIterator[list]:
     '''
     Asyncio generator wrapper that returns items with a given batch size or
     interval (whichever is reached first).
@@ -64,7 +64,9 @@ async def abatcher(
             next_yield = time.perf_counter() + interval_s
 
 
-def batcher(iterable, batch_size):
+def batcher(
+    iterable: types.Iterable, batch_size: int = 10
+) -> types.Iterator[list]:
     '''
     Generator wrapper that returns items with a given batch size
     '''

python_utils/loguru.py

@@ -1,15 +1,15 @@
 from __future__ import annotations
 
-from . import logger
-
 import loguru
 
+from . import logger as logger_module
+
 __all__ = ['Logurud']
 
 
-class Logurud(logger.LoggerBase):
+class Logurud(logger_module.LoggerBase):
     logger: loguru.Logger
 
     def __new__(cls, *args, **kwargs):
-        cls.logger: loguru.Loguru = loguru.logger.opt(depth=1)
+        cls.logger: loguru.Logger = loguru.logger.opt(depth=1)
         return super().__new__(cls)

python_utils/py.typed

@@ -3,8 +3,11 @@
 
 from . import converters
 
+Dimensions = typing.Tuple[int, int]
+OptionalDimensions = typing.Optional[Dimensions]
 
-def get_terminal_size() -> typing.Tuple[int, int]:  # pragma: no cover
+
+def get_terminal_size() -> Dimensions:  # pragma: no cover
     '''Get the current size of your terminal
 
     Multiple returns are not always a good idea, but in this case it greatly
@@ -62,35 +65,36 @@ def get_terminal_size() -> typing.Tuple[int, int]:  # pragma: no cover
         pass
 
     try:
-        w, h = _get_terminal_size_linux()
-        if w and h:
-            return w, h
+        # The method can return None so we don't unpack it
+        wh = _get_terminal_size_linux()
+        if wh is not None and all(wh):
+            return wh
     except Exception:  # pragma: no cover
         pass
 
     try:
         # Windows detection doesn't always work, let's try anyhow
-        w, h = _get_terminal_size_windows()
-        if w and h:
-            return w, h
+        wh = _get_terminal_size_windows()
+        if wh is not None and all(wh):
+            return wh
     except Exception:  # pragma: no cover
         pass
 
     try:
         # needed for window's python in cygwin's xterm!
-        w, h = _get_terminal_size_tput()
-        if w and h:
-            return w, h
+        wh = _get_terminal_size_tput()
+        if wh is not None and all(wh):
+            return wh
     except Exception:  # pragma: no cover
         pass
 
     return 79, 24
 
 
-def _get_terminal_size_windows():  # pragma: no cover
+def _get_terminal_size_windows() -> OptionalDimensions:  # pragma: no cover
     res = None
     try:
-        from ctypes import windll, create_string_buffer
+        from ctypes import windll, create_string_buffer  # type: ignore
 
         # stdin handle is -10
         # stdout handle is -11
@@ -115,7 +119,7 @@ def _get_terminal_size_windows():  # pragma: no cover
         return None
 
 
-def _get_terminal_size_tput():  # pragma: no cover
+def _get_terminal_size_tput() -> OptionalDimensions:  # pragma: no cover
     # get terminal width src: http://stackoverflow.com/questions/263890/
     try:
         import subprocess
@@ -141,19 +145,19 @@ def _get_terminal_size_tput():  # pragma: no cover
         return None
 
 
-def _get_terminal_size_linux():  # pragma: no cover
+def _get_terminal_size_linux() -> OptionalDimensions:  # pragma: no cover
     def ioctl_GWINSZ(fd):
         try:
             import fcntl
             import termios
             import struct
 
-            size = struct.unpack(
-                'hh', fcntl.ioctl(fd, termios.TIOCGWINSZ, '1234')
+            return struct.unpack(
+                'hh',
+                fcntl.ioctl(fd, termios.TIOCGWINSZ, '1234'),  # type: ignore
             )
         except Exception:
             return None
-        return size
 
     size = ioctl_GWINSZ(0) or ioctl_GWINSZ(1) or ioctl_GWINSZ(2)