Maintenance Cleanup 2021/07 (#53)

* constrained decorator type

* core slots

* mypy is mandatory

* no mypy for pypy :(

* PEP 561 -- Distributing and Packaging Type Information

* tee cleans up like any other iterator

* restricted python versions

* overload for sorted

* moved all typing to separate module

Author: maxfischer2781

.github/workflows/verification.yml

@@ -21,3 +21,6 @@ jobs:
     - name: Format with black
       run: |
         black asyncstdlib unittests --diff --check
+    - name: Verify with MyPy
+      run: |
+        mypy --pretty

asyncstdlib/_core.py

@@ -1,6 +1,5 @@
 from inspect import iscoroutinefunction
 from typing import (
-    TypeVar,
     AsyncIterator,
     Iterable,
     AsyncIterable,
@@ -11,25 +10,21 @@
     Callable,
 )
 
-T = TypeVar("T")
-
-
-AnyIterable = Union[Iterable[T], AsyncIterable[T]]
+from ._typing import T, AnyIterable
 
 
 class Sentinel:
     """Placeholder with configurable ``repr``"""
 
+    __slots__ = ("name",)
+
     def __init__(self, name):
         self.name = name
 
     def __repr__(self):
         return self.name
 
 
-__ITER_SENTINEL = Sentinel("<no default>")
-
-
 def aiter(subject: AnyIterable[T]) -> AsyncIterator[T]:
     """
     An async iterator object yielding elements from ``subject``
@@ -58,19 +53,23 @@ async def _aiter_sync(iterable: Iterable[T]) -> AsyncIterator[T]:
 class ScopedIter(Generic[T]):
     """Context manager that provides and cleans up an iterator for an iterable"""
 
+    __slots__ = ("_iterable", "_iterator")
+
     def __init__(self, iterable: AnyIterable[T]):
-        self._iterable = iterable
+        self._iterable: Optional[AnyIterable[T]] = iterable
         self._iterator: Optional[AsyncIterator[T]] = None
 
     async def __aenter__(self) -> AsyncIterator[T]:
-        assert self._iterator is None, f"{self.__class__.__name__} is not re-entrant"
+        assert (
+            self._iterable is not None
+        ), f"{self.__class__.__name__} is not re-entrant"
         self._iterator = aiter(self._iterable)
         self._iterable = None
         return self._iterator
 
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> bool:
         try:
-            aclose = self._iterator.aclose()
+            aclose = self._iterator.aclose()  # type: ignore
         except AttributeError:
             pass
         else:
@@ -89,7 +88,7 @@ def awaitify(
 ) -> Callable[..., Awaitable[T]]:
     """Ensure that ``function`` can be used in ``await`` expressions"""
     if iscoroutinefunction(function):
-        return function
+        return function  # type: ignore
     else:
         return Awaitify(function)
 
@@ -108,10 +107,10 @@ def __call__(self, *args, **kwargs) -> Awaitable[T]:
         if async_call is None:
             value = self.__wrapped__(*args, **kwargs)
             if isinstance(value, Awaitable):
-                self._async_call = self.__wrapped__
+                self._async_call = self.__wrapped__  # type: ignore
                 return value
             else:
-                self._async_call = force_async(self.__wrapped__)
+                self._async_call = force_async(self.__wrapped__)  # type: ignore
                 return await_value(value)
         else:
             return async_call(*args, **kwargs)

asyncstdlib/_lrucache.py

@@ -8,28 +8,19 @@
 from typing import (
     NamedTuple,
     Callable,
-    Awaitable,
-    TypeVar,
+    Any,
     Optional,
     Tuple,
     Dict,
     Union,
 )
 from functools import update_wrapper
 from collections import OrderedDict
-import sys
-
-if sys.version_info[:2] >= (3, 8):
-    from typing import Protocol, TypedDict
-else:
-    from typing_extensions import Protocol, TypedDict
 
+from ._typing import Protocol, TypedDict, C
 from ._utility import public_module
 
 
-R = TypeVar("R")
-
-
 @public_module("asyncstdlib.functools")
 class CacheInfo(NamedTuple):
     """
@@ -65,16 +56,16 @@ class CacheParameters(TypedDict):
 
 
 @public_module("asyncstdlib.functools")
-class LRUAsyncCallable(Protocol[R]):
+class LRUAsyncCallable(Protocol[C]):
     """
     :py:class:`~typing.Protocol` of a LRU cache wrapping a callable to an awaitable
     """
 
     #: The callable wrapped by this cache
-    __wrapped__: Callable[..., Awaitable[R]]
+    __wrapped__: C
 
-    async def __call__(self, *args, **kwargs) -> R:
-        """Get the result of ``await __wrapped__(...)`` from the cache or evaluation"""
+    #: Get the result of ``await __wrapped__(...)`` from the cache or evaluation
+    __call__: C
 
     def cache_parameters(self) -> CacheParameters:
         """Get the parameters of the cache"""
@@ -90,7 +81,7 @@ def cache_clear(self) -> None:
 
 
 @public_module("asyncstdlib.functools")
-def lru_cache(maxsize: Optional[int] = 128, typed: bool = False):
+def lru_cache(maxsize: Optional[Union[int, Callable]] = 128, typed: bool = False):
     """
     Least Recently Used cache for async functions
 
@@ -143,7 +134,8 @@ def lru_cache(maxsize: Optional[int] = 128, typed: bool = False):
             "first argument to 'lru_cache' must be an int, a callable or None"
         )
 
-    def lru_decorator(function: Callable[..., Awaitable[R]]) -> LRUAsyncCallable[R]:
+    def lru_decorator(function: C) -> LRUAsyncCallable[C]:
+        assert not callable(maxsize)
         if maxsize is None:
             wrapper = _unbound_lru(function=function, typed=typed)
         elif maxsize == 0:
@@ -190,14 +182,12 @@ def from_call(
         return cls(key)
 
 
-def _empty_lru(
-    function: Callable[..., Awaitable[R]], typed: bool
-) -> LRUAsyncCallable[R]:
+def _empty_lru(function: C, typed: bool) -> LRUAsyncCallable[C]:
     """Wrap the async ``function`` in an async LRU cache without any capacity"""
     # cache statistics
     misses = 0
 
-    async def wrapper(*args, **kwargs) -> R:
+    async def wrapper(*args, **kwargs):
         nonlocal misses
         misses += 1
         return await function(*args, **kwargs)
@@ -212,25 +202,23 @@ def cache_clear():
         nonlocal misses
         misses = 0
 
-    wrapper.cache_parameters = cache_parameters
-    wrapper.cache_info = cache_info
-    wrapper.cache_clear = cache_clear
-    return wrapper
+    wrapper.cache_parameters = cache_parameters  # type: ignore
+    wrapper.cache_info = cache_info  # type: ignore
+    wrapper.cache_clear = cache_clear  # type: ignore
+    return wrapper  # type: ignore
 
 
-def _unbound_lru(
-    function: Callable[..., Awaitable[R]], typed: bool
-) -> LRUAsyncCallable[R]:
+def _unbound_lru(function: C, typed: bool) -> LRUAsyncCallable[C]:
     """Wrap the async ``function`` in an async LRU cache with infinite capacity"""
     # local lookup
     make_key = CallKey.from_call
     # cache statistics
     hits = 0
     misses = 0
     # cache content
-    cache: Dict[Union[CallKey, int, str], R] = {}
+    cache: Dict[Union[CallKey, int, str], Any] = {}
 
-    async def wrapper(*args, **kwargs) -> R:
+    async def wrapper(*args, **kwargs):
         nonlocal hits, misses
         key = make_key(args, kwargs, typed=typed)
         try:
@@ -259,26 +247,24 @@ def cache_clear():
         hits = 0
         cache.clear()
 
-    wrapper.cache_parameters = cache_parameters
-    wrapper.cache_info = cache_info
-    wrapper.cache_clear = cache_clear
-    return wrapper
+    wrapper.cache_parameters = cache_parameters  # type: ignore
+    wrapper.cache_info = cache_info  # type: ignore
+    wrapper.cache_clear = cache_clear  # type: ignore
+    return wrapper  # type: ignore
 
 
-def _bounded_lru(
-    function: Callable[..., Awaitable[R]], typed: bool, maxsize: int
-) -> LRUAsyncCallable[R]:
+def _bounded_lru(function: C, typed: bool, maxsize: int) -> LRUAsyncCallable[C]:
     """Wrap the async ``function`` in an async LRU cache with fixed capacity"""
     # local lookup
     make_key = CallKey.from_call
     # cache statistics
     hits = 0
     misses = 0
     # cache content
-    cache: OrderedDict[Union[int, str, CallKey], R] = OrderedDict()
+    cache: OrderedDict[Union[int, str, CallKey], Any] = OrderedDict()
     filled = False
 
-    async def wrapper(*args, **kwargs) -> R:
+    async def wrapper(*args, **kwargs):
         nonlocal hits, misses, filled
         key = make_key(args, kwargs, typed=typed)
         try:
@@ -319,7 +305,7 @@ def cache_clear():
         filled = False
         cache.clear()
 
-    wrapper.cache_parameters = cache_parameters
-    wrapper.cache_info = cache_info
-    wrapper.cache_clear = cache_clear
-    return wrapper
+    wrapper.cache_parameters = cache_parameters  # type: ignore
+    wrapper.cache_info = cache_info  # type: ignore
+    wrapper.cache_clear = cache_clear  # type: ignore
+    return wrapper  # type: ignore

asyncstdlib/_typing.py

@@ -0,0 +1,71 @@
+"""
+Helper module to simplify version specific typing imports
+
+This module is for internal use only. Do *not* put any new
+"async typing" definitions here.
+"""
+import sys
+from typing import TypeVar, Hashable, Union, AsyncIterable, Iterable, Callable
+
+if sys.version_info[:2] >= (3, 8):
+    from typing import Protocol, AsyncContextManager, ContextManager, TypedDict
+else:
+    from typing_extensions import (
+        Protocol,
+        AsyncContextManager,
+        ContextManager,
+        TypedDict,
+    )
+
+__all__ = [
+    "Protocol",
+    "AsyncContextManager",
+    "ContextManager",
+    "TypedDict",
+    "T",
+    "T1",
+    "T2",
+    "T3",
+    "T4",
+    "T5",
+    "R",
+    "C",
+    "HK",
+    "LT",
+    "ADD",
+    "AnyIterable",
+]
+
+# TypeVars for argument/return type
+T = TypeVar("T")
+T1 = TypeVar("T1")
+T2 = TypeVar("T2")
+T3 = TypeVar("T3")
+T4 = TypeVar("T4")
+T5 = TypeVar("T5")
+R = TypeVar("R", covariant=True)
+C = TypeVar("C", bound=Callable)
+
+#: Hashable Key
+HK = TypeVar("HK", bound=Hashable)
+
+# LT < LT
+LT = TypeVar("LT", bound="SupportsLT")
+
+
+class SupportsLT(Protocol):
+    def __lt__(self: LT, other: LT) -> bool:
+        raise NotImplementedError
+
+
+# ADD + ADD
+ADD = TypeVar("ADD", bound="SupportsAdd")
+
+
+class SupportsAdd(Protocol):
+    def __add__(self: ADD, other: ADD) -> bool:
+        raise NotImplementedError
+
+
+#: (async) iter T
+AnyIterable = Union[Iterable[T], AsyncIterable[T]]

asyncstdlib/_utility.py

@@ -1,12 +1,25 @@
-from typing import TypeVar, Any
+from typing import TypeVar, Any, Optional
 
-T = TypeVar("T")
+from ._typing import Protocol
 
 
-def public_module(module_name: str, qual_name: str = None):
+class Definition(Protocol):
+    """
+    Type of objects created from a class or function definition
+    """
+
+    __name__: str
+    __module__: str
+    __qualname__: str
+
+
+D = TypeVar("D", bound=Definition)
+
+
+def public_module(module_name: str, qual_name: Optional[str] = None):
     """Set the module name of a function or class"""
 
-    def decorator(thing: T) -> T:
+    def decorator(thing: D) -> D:
         thing.__module__ = module_name
         if qual_name is not None:
             thing.__qualname__ = qual_name