Merge pull request #100 from ikalnytskyi/chore/typing

Drop 'typing_extensions', use modern annotations

pyproject.toml

@@ -44,7 +44,7 @@ dependencies = ["ruff == 0.5.*"]
 scripts.run = ["ruff check {args:.}", "ruff format --check --diff {args:.}"]
 
 [tool.hatch.envs.type]
-dependencies = ["mypy", "typing-extensions", "flask"]
+dependencies = ["mypy", "flask"]
 scripts.run = ["mypy {args}"]
 
 [tool.hatch.envs.docs]
@@ -68,7 +68,6 @@ known-first-party = ["picobox"]
 "tests/*" = ["D", "S101", "ARG001", "BLE001", "INP001"]
 
 [tool.mypy]
-python_version = "3.8"
 files = ["src"]
 pretty = true
 strict = true

src/picobox/_box.py

@@ -5,16 +5,17 @@
 import functools
 import inspect
 import threading
-import typing as t
+import typing
 
 from . import _scopes
 
-if t.TYPE_CHECKING:
-    import typing_extensions
+if typing.TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable, Hashable
+    from typing import Any, ParamSpec, TypeVar, Union
 
-    P = typing_extensions.ParamSpec("P")
-    T = typing_extensions.TypeVar("T")
-    R = t.Union[T, t.Awaitable[T]]
+    P = ParamSpec("P")
+    T = TypeVar("T")
+    R = Union[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
@@ -50,16 +51,16 @@ def do(magic):
     """
 
     def __init__(self) -> None:
-        self._store: dict[t.Hashable, tuple[_scopes.Scope, t.Callable[[], t.Any]]] = {}
+        self._store: dict[Hashable, tuple[_scopes.Scope, Callable[[], Any]]] = {}
         self._scope_instances: dict[type[_scopes.Scope], _scopes.Scope] = {}
         self._lock = threading.RLock()
 
     def put(
         self,
-        key: t.Hashable,
-        value: t.Any = _unset,
+        key: Hashable,
+        value: Any = _unset,
         *,
-        factory: t.Callable[[], t.Any] | None = None,
+        factory: Callable[[], Any] | None = None,
         scope: type[_scopes.Scope] | None = None,
     ) -> None:
         """Define a dependency (aka service) within the box instance.
@@ -93,7 +94,7 @@ def put(
             error_message = "Box.put() takes 'scope' when 'factory' provided"
             raise TypeError(error_message)
 
-        def _factory() -> t.Any:
+        def _factory() -> Any:
             return value
 
         factory = factory or _factory
@@ -129,7 +130,7 @@ def _factory() -> t.Any:
         with self._lock:
             self._store[key] = (scope_instance, factory)
 
-    def get(self, key: t.Hashable, default: t.Any = _unset) -> t.Any:
+    def get(self, key: Hashable, default: Any = _unset) -> Any:
         """Retrieve a dependency (aka service) out of the box instance.
 
         The process involves creation of requested dependency by calling an
@@ -171,10 +172,10 @@ def get(self, key: t.Hashable, default: t.Any = _unset) -> t.Any:
 
     def pass_(
         self,
-        key: t.Hashable,
+        key: Hashable,
         *,
         as_: str | None = None,
-    ) -> t.Callable[[t.Callable[P, R[T]]], t.Callable[P, R[T]]]:
+    ) -> Callable[[Callable[P, R[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
@@ -190,7 +191,7 @@ def pass_(
         :raises KeyError: If no dependencies saved under `key` in the box.
         """
 
-        def decorator(fn: t.Callable[P, R[T]]) -> t.Callable[P, R[T]]:
+        def decorator(fn: Callable[P, R[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__"):
@@ -218,7 +219,8 @@ def fn_with_dependencies(*args: P.args, **kwargs: P.kwargs) -> R[T]:
 
                 @functools.wraps(fn)
                 async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
-                    return await t.cast(t.Awaitable["T"], fn_with_dependencies(*args, **kwargs))
+                    coroutine = fn_with_dependencies(*args, **kwargs)
+                    return await typing.cast(typing.Awaitable["T"], coroutine)
             else:
                 wrapper = fn_with_dependencies  # type: ignore[assignment]
 
@@ -266,16 +268,16 @@ def __init__(self, *boxes: Box) -> None:
 
     def put(
         self,
-        key: t.Hashable,
-        value: t.Any = _unset,
+        key: Hashable,
+        value: Any = _unset,
         *,
-        factory: t.Callable[[], t.Any] | None = None,
+        factory: Callable[[], Any] | None = None,
         scope: type[_scopes.Scope] | None = None,
     ) -> None:
         """Same as :meth:`Box.put` but applies to first underlying box."""
         return self._boxes[0].put(key, value, factory=factory, scope=scope)
 
-    def get(self, key: t.Hashable, default: t.Any = _unset) -> t.Any:
+    def get(self, key: Hashable, default: Any = _unset) -> Any:
         """Same as :meth:`Box.get` but looks up for key in underlying boxes."""
         for box in self._boxes:
             try:

src/picobox/_scopes.py

@@ -5,9 +5,13 @@
 import abc
 import contextvars as _contextvars
 import threading
-import typing as t
+import typing
 import weakref
 
+if typing.TYPE_CHECKING:
+    from collections.abc import Hashable
+    from typing import Any
+
 
 class Scope(metaclass=abc.ABCMeta):
     """Scope is an execution context based storage interface.
@@ -25,24 +29,24 @@ class Scope(metaclass=abc.ABCMeta):
     """
 
     @abc.abstractmethod
-    def set(self, key: t.Hashable, value: t.Any) -> None:
+    def set(self, key: Hashable, value: Any) -> None:
         """Bind `value` to `key` in current execution context."""
 
     @abc.abstractmethod
-    def get(self, key: t.Hashable) -> t.Any:
+    def get(self, key: Hashable) -> Any:
         """Get `value` by `key` for current execution context."""
 
 
 class singleton(Scope):
     """Share instances across application."""
 
     def __init__(self) -> None:
-        self._store: dict[t.Hashable, t.Any] = {}
+        self._store: dict[Hashable, Any] = {}
 
-    def set(self, key: t.Hashable, value: t.Any) -> None:
+    def set(self, key: Hashable, value: Any) -> None:
         self._store[key] = value
 
-    def get(self, key: t.Hashable) -> t.Any:
+    def get(self, key: Hashable) -> Any:
         return self._store[key]
 
 
@@ -52,14 +56,14 @@ class threadlocal(Scope):
     def __init__(self) -> None:
         self._local = threading.local()
 
-    def set(self, key: t.Hashable, value: t.Any) -> None:
+    def set(self, key: Hashable, value: Any) -> None:
         try:
             store = self._local.store
         except AttributeError:
             store = self._local.store = {}
         store[key] = value
 
-    def get(self, key: t.Hashable) -> t.Any:
+    def get(self, key: Hashable) -> Any:
         try:
             rv = self._local.store[key]
         except AttributeError:
@@ -79,22 +83,22 @@ class contextvars(Scope):
     .. versionadded:: 2.1
     """
 
-    _store_obj: weakref.WeakKeyDictionary[Scope, dict[t.Hashable, _contextvars.ContextVar[t.Any]]]
+    _store_obj: weakref.WeakKeyDictionary[Scope, dict[Hashable, _contextvars.ContextVar[Any]]]
     _store_obj = weakref.WeakKeyDictionary()
 
     @property
-    def _store(self) -> dict[t.Hashable, _contextvars.ContextVar[t.Any]]:
+    def _store(self) -> dict[Hashable, _contextvars.ContextVar[Any]]:
         try:
             scope_store = self._store_obj[self]
         except KeyError:
             scope_store = self._store_obj[self] = {}
         return scope_store
 
-    def set(self, key: t.Hashable, value: t.Any) -> None:
+    def set(self, key: Hashable, value: Any) -> None:
         self._store[key] = _contextvars.ContextVar(str(key))
         self._store[key].set(value)
 
-    def get(self, key: t.Hashable) -> t.Any:
+    def get(self, key: Hashable) -> Any:
         try:
             return self._store[key].get()
         except LookupError:
@@ -104,8 +108,8 @@ def get(self, key: t.Hashable) -> t.Any:
 class noscope(Scope):
     """Do not share instances, create them each time on demand."""
 
-    def set(self, key: t.Hashable, value: t.Any) -> None:
+    def set(self, key: Hashable, value: Any) -> None:
         pass
 
-    def get(self, key: t.Hashable) -> t.Any:
+    def get(self, key: Hashable) -> Any:
         raise KeyError(key)

src/picobox/_stack.py

@@ -4,27 +4,29 @@
 
 import contextlib
 import threading
-import typing as t
+import typing
 
 from ._box import Box, ChainBox, _unset
 
-if t.TYPE_CHECKING:
-    import typing_extensions
+if typing.TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable, Generator, Hashable
+    from contextlib import AbstractContextManager
+    from typing import Any, ParamSpec, TypeVar, Union
 
     from ._scopes import Scope
 
-    P = typing_extensions.ParamSpec("P")
-    T = typing_extensions.TypeVar("T")
-    R = t.Union[T, t.Awaitable[T]]
+    P = ParamSpec("P")
+    T = TypeVar("T")
+    R = Union[T | Awaitable[T]]
 
 _ERROR_MESSAGE_EMPTY_STACK = "No boxes found on the stack, please `.push()` a box first."
 
 
 @contextlib.contextmanager
 def _create_push_context_manager(
     box: Box,
-    pop_callback: t.Callable[[], Box],
-) -> t.Generator[Box, None, None]:
+    pop_callback: Callable[[], Box],
+) -> Generator[Box, None, None]:
     """Create a context manager that calls something on exit."""
     try:
         yield box
@@ -45,7 +47,7 @@ class _CurrentBoxProxy(Box):
     def __init__(self, stack: list[Box]) -> None:
         self._stack = stack
 
-    def __getattribute__(self, name: str) -> t.Any:
+    def __getattribute__(self, name: str) -> Any:
         if name == "_stack":
             return super().__getattribute__(name)
 
@@ -100,7 +102,7 @@ def do(magic):
     """
 
     def __init__(self, name: str | None = None) -> None:
-        self._name = name or f"0x{id(t):x}"
+        self._name = name or f"0x{id(self):x}"
         self._stack: list[Box] = []
         self._lock = threading.Lock()
 
@@ -114,7 +116,7 @@ def __init__(self, name: str | None = None) -> None:
     def __repr__(self) -> str:
         return f"<Stack ({self._name})>"
 
-    def push(self, box: Box, *, chain: bool = False) -> t.ContextManager[Box]:
+    def push(self, box: Box, *, chain: bool = False) -> AbstractContextManager[Box]:
         """Push a :class:`Box` instance to the top of the stack.
 
         Returns a context manager, that will automatically pop the box from the
@@ -165,33 +167,33 @@ def pop(self) -> Box:
 
     def put(
         self,
-        key: t.Hashable,
-        value: t.Any = _unset,
+        key: Hashable,
+        value: Any = _unset,
         *,
-        factory: t.Callable[[], t.Any] | None = None,
+        factory: Callable[[], Any] | None = None,
         scope: type[Scope] | None = 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:
+    def get(self, key: Hashable, default: Any = _unset) -> Any:
         """The same as :meth:`Box.get` but for a box at the top."""
         return self._current_box.get(key, default=default)
 
     def pass_(
         self,
-        key: t.Hashable,
+        key: Hashable,
         *,
         as_: str | None = None,
-    ) -> t.Callable[[t.Callable[P, R[T]]], t.Callable[P, R[T]]]:
+    ) -> Callable[[Callable[P, R[T]]], Callable[P, R[T]]]:
         """The same as :meth:`Box.pass_` but for a box at the top."""
         return Box.pass_(self._current_box, key, as_=as_)
 
 
 _instance = Stack("shared")
 
 
-def push(box: Box, *, chain: bool = False) -> t.ContextManager[Box]:
+def push(box: Box, *, chain: bool = False) -> AbstractContextManager[Box]:
     """The same as :meth:`Stack.push` but for a shared stack instance.
 
     .. versionadded:: 1.1 ``chain`` parameter
@@ -208,25 +210,25 @@ def pop() -> Box:
 
 
 def put(
-    key: t.Hashable,
-    value: t.Any = _unset,
+    key: Hashable,
+    value: Any = _unset,
     *,
-    factory: t.Callable[[], t.Any] | None = None,
+    factory: Callable[[], Any] | None = None,
     scope: type[Scope] | None = None,
 ) -> None:
     """The same as :meth:`Stack.put` but for a shared stack instance."""
     return _instance.put(key, value, factory=factory, scope=scope)
 
 
-def get(key: t.Hashable, default: t.Any = _unset) -> t.Any:
+def get(key: Hashable, default: Any = _unset) -> Any:
     """The same as :meth:`Stack.get` but for a shared stack instance."""
     return _instance.get(key, default=default)
 
 
 def pass_(
-    key: t.Hashable,
+    key: Hashable,
     *,
     as_: str | None = None,
-) -> t.Callable[[t.Callable[P, R[T]]], t.Callable[P, R[T]]]:
+) -> Callable[[Callable[P, R[T]]], Callable[P, R[T]]]:
     """The same as :meth:`Stack.pass_` but for a shared stack instance."""
     return _instance.pass_(key, as_=as_)