Merge pull request #90 from ikalnytskyi/chore/mypy-no-untyped-def

Turn MyPy's no-untyped-def/no-untyped-call on

pyproject.toml

@@ -44,7 +44,7 @@ dependencies = ["ruff == 0.4.*"]
 scripts.run = ["ruff check {args:.}", "ruff format --check --diff {args:.}"]
 
 [tool.hatch.envs.type]
-dependencies = ["mypy", "flask"]
+dependencies = ["mypy", "typing-extensions", "flask"]
 scripts.run = ["mypy {args}"]
 
 [tool.hatch.envs.docs]
@@ -97,6 +97,4 @@ strict = true
 disable_error_code = [
   "attr-defined",
   "comparison-overlap",
-  "no-untyped-call",
-  "no-untyped-def",
 ]

src/picobox/_box.py

@@ -7,6 +7,13 @@
 
 from . import _scopes
 
+if t.TYPE_CHECKING:
+    import typing_extensions
+
+    P = typing_extensions.ParamSpec("P")
+    T = typing_extensions.TypeVar("T")
+    R = t.Union[T, t.Awaitable[T]]
+
 # Missing is a special sentinel object that's used to indicate a value is
 # missing when "None" is a valid input. It's important to use a good name
 # because it appears in function signatures in API reference (see docs).
@@ -40,9 +47,9 @@ def do(magic):
         assert do() == 43
     """
 
-    def __init__(self):
-        self._store = {}
-        self._scope_instances = {}
+    def __init__(self) -> None:
+        self._store: t.Dict[t.Hashable, t.Tuple[_scopes.Scope, t.Callable[[], t.Any]]] = {}
+        self._scope_instances: t.Dict[t.Type[_scopes.Scope], _scopes.Scope] = {}
         self._lock = threading.RLock()
 
     def put(
@@ -81,18 +88,18 @@ def put(
         if value is not _unset and scope is not None:
             raise TypeError("Box.put() takes 'scope' when 'factory' provided")
 
+        def _factory() -> t.Any:
+            return value
+
+        factory = factory or _factory
+
         # Value is a syntax sugar Box supports to store objects "As Is"
         # with singleton scope. In other words it's essentially the same
         # as one pass "factory=lambda: value". Alternatively, Box could
         # have just one factory argument and check it for callable, but
         # in this case it wouldn't support values which are callable by
         # its nature.
         if value is not _unset:
-
-            def _factory():
-                return value
-
-            factory = _factory
             scope = _scopes.singleton
 
         # If scope is not explicitly passed, Box assumes "No Scope"
@@ -157,7 +164,12 @@ def get(self, key: t.Hashable, default: t.Any = _unset) -> t.Any:
 
         return value
 
-    def pass_(self, key: t.Hashable, *, as_: t.Optional[str] = None):
+    def pass_(
+        self,
+        key: t.Hashable,
+        *,
+        as_: t.Optional[str] = None,
+    ) -> "t.Callable[[t.Callable[P, R[T]]], t.Callable[P, R[T]]]":
         r"""Pass a dependency to a function if nothing explicitly passed.
 
         The decorator implements late binding which means it does not require
@@ -173,15 +185,15 @@ def pass_(self, key: t.Hashable, *, as_: t.Optional[str] = None):
         :raises KeyError: If no dependencies saved under `key` in the box.
         """
 
-        def decorator(fn):
+        def decorator(fn: "t.Callable[P, R[T]]") -> "t.Callable[P, R[T]]":
             # If pass_ decorator is called second time (or more), we can squash
             # the calls into one and reduce runtime costs of injection.
             if hasattr(fn, "__dependencies__"):
                 fn.__dependencies__.append((key, as_))
                 return fn
 
             @functools.wraps(fn)
-            def fn_with_dependencies(*args, **kwargs):
+            def fn_with_dependencies(*args: "P.args", **kwargs: "P.kwargs") -> "R[T]":
                 signature = inspect.signature(fn)
                 arguments = signature.bind_partial(*args, **kwargs)
 
@@ -200,10 +212,10 @@ def fn_with_dependencies(*args, **kwargs):
             if inspect.iscoroutinefunction(fn):
 
                 @functools.wraps(fn)
-                async def wrapper(*args, **kwargs):
-                    return await fn_with_dependencies(*args, **kwargs)
+                async def wrapper(*args: "P.args", **kwargs: "P.kwargs") -> "T":
+                    return await t.cast(t.Awaitable["T"], fn_with_dependencies(*args, **kwargs))
             else:
-                wrapper = fn_with_dependencies
+                wrapper = fn_with_dependencies  # type: ignore[assignment]
 
             wrapper.__dependencies__ = [(key, as_)]
             return wrapper
@@ -244,7 +256,7 @@ def do(magic_a, magic_b):
     .. versionadded:: 1.1
     """
 
-    def __init__(self, *boxes: Box):
+    def __init__(self, *boxes: Box) -> None:
         self._boxes = boxes or (Box(),)
 
     def put(

src/picobox/_scopes.py

@@ -33,8 +33,8 @@ def get(self, key: t.Hashable) -> t.Any:
 class singleton(Scope):
     """Share instances across application."""
 
-    def __init__(self):
-        self._store = {}
+    def __init__(self) -> None:
+        self._store: t.Dict[t.Hashable, t.Any] = {}
 
     def set(self, key: t.Hashable, value: t.Any) -> None:
         self._store[key] = value
@@ -46,7 +46,7 @@ def get(self, key: t.Hashable) -> t.Any:
 class threadlocal(Scope):
     """Share instances across the same thread."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         self._local = threading.local()
 
     def set(self, key: t.Hashable, value: t.Any) -> None:
@@ -76,8 +76,8 @@ class contextvars(Scope):
     .. versionadded:: 2.1
     """
 
-    def __init__(self):
-        self._store = {}
+    def __init__(self) -> None:
+        self._store: t.Dict[t.Hashable, _contextvars.ContextVar[t.Any]] = {}
 
     def set(self, key: t.Hashable, value: t.Any) -> None:
         try:

src/picobox/_stack.py

@@ -1,54 +1,54 @@
 """Picobox API to work with a box at the top of the stack."""
 
 import contextlib
-import functools
 import threading
 import typing as t
 
-from ._box import Box, ChainBox
-
-_ERROR_MESSAGE_EMPTY_STACK = "No boxes found on the stack, please `.push()` a box first."
-
-
-def _copy_signature(method, instance=None):
-    # This is a workaround to overcome 'sphinx.ext.autodoc' inability to
-    # retrieve a docstring of a bound method. Here's the trick - we create
-    # a partial function, and autodoc can deal with partially applied
-    # functions.
-    if instance:
-        method = functools.partial(method, instance)
-
-    # The reason behind empty arguments is to reuse a signature of wrapped
-    # function while preserving "__doc__", "__name__" and other accompanied
-    # attributes. They are very helpful for troubleshooting as well as
-    # necessary for Sphinx API reference.
-    return functools.wraps(method, (), ())
+from . import _scopes
+from ._box import Box, ChainBox, _unset
 
+if t.TYPE_CHECKING:
+    import typing_extensions
 
-def _create_stack_proxy(stack):
-    """Create an object that proxies all calls to the top of the stack."""
-
-    class _StackProxy:
-        def __getattribute__(self, name):
-            try:
-                return getattr(stack[-1], name)
-            except IndexError:
-                raise RuntimeError(_ERROR_MESSAGE_EMPTY_STACK)
+    P = typing_extensions.ParamSpec("P")
+    T = typing_extensions.TypeVar("T")
+    R = t.Union[T, t.Awaitable[T]]
 
-    return _StackProxy()
+_ERROR_MESSAGE_EMPTY_STACK = "No boxes found on the stack, please `.push()` a box first."
 
 
 @contextlib.contextmanager
-def _create_push_context_manager(box, pop_callback):
+def _create_push_context_manager(
+    box: Box,
+    pop_callback: t.Callable[[], Box],
+) -> t.Generator[Box, None, None]:
     """Create a context manager that calls something on exit."""
     try:
         yield box
     finally:
+        popped_box = pop_callback()
+
         # Ensure the poped box is the same that was submitted by this exact
         # context manager. It may happen if someone messed up with order of
-        # push() and pop() calls. Normally, push() should be used a context
+        # push() and pop() calls. Normally, push() should be used as a context
         # manager to avoid this issue.
-        assert pop_callback() is box
+        assert popped_box is box
+
+
+class _CurrentBoxProxy(Box):
+    """Delegates operations to the Box instance at the top of the stack."""
+
+    def __init__(self, stack: t.List[Box]) -> None:
+        self._stack = stack
+
+    def __getattribute__(self, name: str) -> t.Any:
+        if name == "_stack":
+            return super().__getattribute__(name)
+
+        try:
+            return getattr(self._stack[-1], name)
+        except IndexError:
+            raise RuntimeError(_ERROR_MESSAGE_EMPTY_STACK)
 
 
 class Stack:
@@ -95,7 +95,7 @@ def do(magic):
     .. versionadded:: 2.2
     """
 
-    def __init__(self, name: t.Optional[str] = None):
+    def __init__(self, name: t.Optional[str] = None) -> None:
         self._name = name or f"0x{id(t):x}"
         self._stack: t.List[Box] = []
         self._lock = threading.Lock()
@@ -105,9 +105,9 @@ def __init__(self, name: t.Optional[str] = None):
         # that mimic Box interface but deal with a box on the top instead.
         # While it's not completely necessary for `put()` and `get()`, it's
         # crucial for `pass_()` due to its laziness and thus late evaluation.
-        self._topbox = _create_stack_proxy(self._stack)
+        self._current_box = _CurrentBoxProxy(self._stack)
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"<Stack ({self._name})>"
 
     def push(self, box: Box, *, chain: bool = False) -> t.ContextManager[Box]:
@@ -159,56 +159,70 @@ def pop(self) -> Box:
             except IndexError:
                 raise RuntimeError(_ERROR_MESSAGE_EMPTY_STACK)
 
-    @_copy_signature(Box.put)
-    def put(self, *args, **kwargs):
-        """The same as :meth:`Box.put` but for a box at the top."""
-        return self._topbox.put(*args, **kwargs)
-
-    @_copy_signature(Box.get)
-    def get(self, *args, **kwargs):
+    def put(
+        self,
+        key: t.Hashable,
+        value: t.Any = _unset,
+        *,
+        factory: t.Optional[t.Callable[[], t.Any]] = None,
+        scope: t.Optional[t.Type[_scopes.Scope]] = None,
+    ) -> None:
+        """The same as :meth:`Box.put` but for a box at the top of the stack."""
+        return self._current_box.put(key, value, factory=factory, scope=scope)
+
+    def get(self, key: t.Hashable, default: t.Any = _unset) -> t.Any:
         """The same as :meth:`Box.get` but for a box at the top."""
-        return self._topbox.get(*args, **kwargs)
-
-    @_copy_signature(Box.pass_)
-    def pass_(self, *args, **kwargs):
+        return self._current_box.get(key, default=default)
+
+    def pass_(
+        self,
+        key: t.Hashable,
+        *,
+        as_: t.Optional[str] = None,
+    ) -> "t.Callable[[t.Callable[P, R[T]]], t.Callable[P, R[T]]]":
         """The same as :meth:`Box.pass_` but for a box at the top."""
-        return Box.pass_(self._topbox, *args, **kwargs)
+        return Box.pass_(self._current_box, key, as_=as_)
 
 
 _instance = Stack("shared")
 
 
-@_copy_signature(Stack.push, _instance)
-def push(*args, **kwargs):
+def push(box: Box, *, chain: bool = False) -> t.ContextManager[Box]:
     """The same as :meth:`Stack.push` but for a shared stack instance.
 
     .. versionadded:: 1.1 ``chain`` parameter
     """
-    return _instance.push(*args, **kwargs)
+    return _instance.push(box, chain=chain)
 
 
-@_copy_signature(Stack.pop, _instance)
-def pop(*args, **kwargs):
+def pop() -> Box:
     """The same as :meth:`Stack.pop` but for a shared stack instance.
 
     .. versionadded:: 2.0
     """
-    return _instance.pop(*args, **kwargs)
+    return _instance.pop()
 
 
-@_copy_signature(Stack.put, _instance)
-def put(*args, **kwargs):
+def put(
+    key: t.Hashable,
+    value: t.Any = _unset,
+    *,
+    factory: t.Optional[t.Callable[[], t.Any]] = None,
+    scope: t.Optional[t.Type[_scopes.Scope]] = None,
+) -> None:
     """The same as :meth:`Stack.put` but for a shared stack instance."""
-    return _instance.put(*args, **kwargs)
+    return _instance.put(key, value, factory=factory, scope=scope)
 
 
-@_copy_signature(Stack.get, _instance)
-def get(*args, **kwargs):
+def get(key: t.Hashable, default: t.Any = _unset) -> t.Any:
     """The same as :meth:`Stack.get` but for a shared stack instance."""
-    return _instance.get(*args, **kwargs)
+    return _instance.get(key, default=default)
 
 
-@_copy_signature(Stack.pass_, _instance)
-def pass_(*args, **kwargs):
+def pass_(
+    key: t.Hashable,
+    *,
+    as_: t.Optional[str] = None,
+) -> "t.Callable[[t.Callable[P, R[T]]], t.Callable[P, R[T]]]":
     """The same as :meth:`Stack.pass_` but for a shared stack instance."""
-    return _instance.pass_(*args, **kwargs)
+    return _instance.pass_(key, as_=as_)

src/picobox/ext/asgiscopes.py

@@ -9,6 +9,11 @@
 if t.TYPE_CHECKING:
     Store = weakref.WeakKeyDictionary[picobox.Scope, t.MutableMapping[t.Hashable, t.Any]]
     StoreCtxVar = contextvars.ContextVar[Store]
+    ASGIScope = t.MutableMapping[str, t.Any]
+    ASGIMessage = t.MutableMapping[str, t.Any]
+    ASGIReceive = t.Callable[[], t.Awaitable[ASGIMessage]]
+    ASGISend = t.Callable[[ASGIMessage], t.Awaitable[None]]
+    ASGIApplication = t.Callable[[ASGIScope, ASGIReceive, ASGISend], t.Awaitable[None]]
 
 
 _current_app_store: "StoreCtxVar" = contextvars.ContextVar(f"{__name__}.current-app-store")
@@ -30,14 +35,14 @@ class ScopeMiddleware:
     :param app: The ASGI application to wrap.
     """
 
-    def __init__(self, app):
+    def __init__(self, app: "ASGIApplication") -> None:
         self.app = app
         # Since we want stored objects to be garbage collected as soon as the
         # storing scope instance is destroyed, scope instances have to be
         # weakly referenced.
         self.store: Store = weakref.WeakKeyDictionary()
 
-    async def __call__(self, scope, receive, send):
+    async def __call__(self, scope: "ASGIScope", receive: "ASGIReceive", send: "ASGISend") -> None:
         """Define scopes and invoke the ASGI application."""
         # Storing the ASGI application's scope state within a ScopeMiddleware
         # instance because it's assumed that each ASGI middleware is typically