From 9899dd8f924fb7db1a8d206085dc049cd8e76324 Mon Sep 17 00:00:00 2001 From: Ihor Kalnytskyi Date: Fri, 28 Jun 2024 20:37:30 +0300 Subject: [PATCH] Drop 'typing_extensions', use modern annotations Since introduction of `from __future__ import annotations` in the project sources, the annotations evaluations aren't happening at runtime, which means we can start using modern type annotation techniques, such as | operand instead of Union. It also means we can completely rely on the `typing` module and drop `typing_extensions` because all imports are now under TYPE_CHECKING, and their usage in signatures are not evaluated at runtime. --- pyproject.toml | 3 +- src/picobox/_box.py | 43 +++++++++++++++-------------- src/picobox/_scopes.py | 32 ++++++++++++---------- src/picobox/_stack.py | 50 ++++++++++++++++++---------------- src/picobox/ext/asgiscopes.py | 25 +++++++++-------- src/picobox/ext/flaskscopes.py | 18 ++++++------ src/picobox/ext/wsgiscopes.py | 17 +++++++----- 7 files changed, 102 insertions(+), 86 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 1d5aaef..f0498f8 100644 --- a/pyproject.toml +++ b/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 diff --git a/src/picobox/_box.py b/src/picobox/_box.py index 31994b3..c6164a4 100644 --- a/src/picobox/_box.py +++ b/src/picobox/_box.py @@ -2,19 +2,21 @@ from __future__ import annotations +import collections.abc 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 +52,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 +95,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 +131,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 +173,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 +192,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 +220,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(collections.abc.Awaitable[T], coroutine) else: wrapper = fn_with_dependencies # type: ignore[assignment] @@ -266,16 +269,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: diff --git a/src/picobox/_scopes.py b/src/picobox/_scopes.py index 7c3fdf8..775352e 100644 --- a/src/picobox/_scopes.py +++ b/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,11 +29,11 @@ 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.""" @@ -37,12 +41,12 @@ 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) diff --git a/src/picobox/_stack.py b/src/picobox/_stack.py index ec07c86..934ebe7 100644 --- a/src/picobox/_stack.py +++ b/src/picobox/_stack.py @@ -4,18 +4,20 @@ 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." @@ -23,8 +25,8 @@ @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"" - 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,25 +167,25 @@ 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_) @@ -191,7 +193,7 @@ def pass_( _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_) diff --git a/src/picobox/ext/asgiscopes.py b/src/picobox/ext/asgiscopes.py index 74b1211..5bc0e29 100644 --- a/src/picobox/ext/asgiscopes.py +++ b/src/picobox/ext/asgiscopes.py @@ -3,19 +3,22 @@ from __future__ import annotations import contextvars -import typing as t +import typing import weakref import picobox -if t.TYPE_CHECKING: - Store = weakref.WeakKeyDictionary[picobox.Scope, t.Dict[t.Hashable, t.Any]] +if typing.TYPE_CHECKING: + from collections.abc import Awaitable, Callable, Hashable, MutableMapping + from typing import Any + + Store = weakref.WeakKeyDictionary[picobox.Scope, dict[Hashable, 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]] + ASGIScope = MutableMapping[str, Any] + ASGIMessage = MutableMapping[str, Any] + ASGIReceive = Callable[[], Awaitable[ASGIMessage]] + ASGISend = Callable[[ASGIMessage], Awaitable[None]] + ASGIApplication = Callable[[ASGIScope, ASGIReceive, ASGISend], Awaitable[None]] _current_app_store: StoreCtxVar = contextvars.ContextVar(f"{__name__}.current-app-store") @@ -67,7 +70,7 @@ class _asgiscope(picobox.Scope): _store_cvar: StoreCtxVar @property - def _store(self) -> dict[t.Hashable, t.Any]: + def _store(self) -> dict[Hashable, Any]: try: store = self._store_cvar.get() except LookupError: @@ -86,10 +89,10 @@ def _store(self) -> dict[t.Hashable, t.Any]: scope_store = store.setdefault(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] = value - def get(self, key: t.Hashable) -> t.Any: + def get(self, key: Hashable) -> Any: return self._store[key] diff --git a/src/picobox/ext/flaskscopes.py b/src/picobox/ext/flaskscopes.py index 1c093d5..8b1e2ea 100644 --- a/src/picobox/ext/flaskscopes.py +++ b/src/picobox/ext/flaskscopes.py @@ -2,27 +2,29 @@ from __future__ import annotations -import typing as t +import typing import weakref import flask import picobox -if t.TYPE_CHECKING: +if typing.TYPE_CHECKING: + from collections.abc import Hashable + from typing import Any - class _flask_store_obj(t.Protocol): - __dependencies__: weakref.WeakKeyDictionary[picobox.Scope, dict[t.Hashable, t.Any]] + class _flask_store_obj(typing.Protocol): + __dependencies__: weakref.WeakKeyDictionary[picobox.Scope, dict[Hashable, Any]] class _flaskscope(picobox.Scope): """A base class for Flask scopes.""" def __init__(self, store_obj: object) -> None: - self._store_obj = t.cast("_flask_store_obj", store_obj) + self._store_obj = typing.cast("_flask_store_obj", store_obj) @property - def _store(self) -> dict[t.Hashable, t.Any]: + def _store(self) -> dict[Hashable, Any]: try: store = self._store_obj.__dependencies__ except AttributeError: @@ -34,10 +36,10 @@ def _store(self) -> dict[t.Hashable, t.Any]: scope_store = store.setdefault(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] = value - def get(self, key: t.Hashable) -> t.Any: + def get(self, key: Hashable) -> Any: return self._store[key] diff --git a/src/picobox/ext/wsgiscopes.py b/src/picobox/ext/wsgiscopes.py index ea952f0..7394caa 100644 --- a/src/picobox/ext/wsgiscopes.py +++ b/src/picobox/ext/wsgiscopes.py @@ -3,15 +3,18 @@ from __future__ import annotations import contextvars -import typing as t +import typing import weakref import picobox -if t.TYPE_CHECKING: +if typing.TYPE_CHECKING: + from collections.abc import Hashable, Iterable + from typing import Any + from _typeshed.wsgi import StartResponse, WSGIApplication, WSGIEnvironment - Store = weakref.WeakKeyDictionary[picobox.Scope, t.Dict[t.Hashable, t.Any]] + Store = weakref.WeakKeyDictionary[picobox.Scope, dict[Hashable, Any]] StoreCtxVar = contextvars.ContextVar[Store] @@ -45,7 +48,7 @@ def __call__( self, environ: WSGIEnvironment, start_response: StartResponse, - ) -> t.Iterable[bytes]: + ) -> Iterable[bytes]: """Define scopes and invoke the WSGI application.""" # Storing the WSGI application's scope state within a ScopeMiddleware # instance because it's assumed that each WSGI middleware is typically @@ -69,7 +72,7 @@ class _wsgiscope(picobox.Scope): _store_cvar: StoreCtxVar @property - def _store(self) -> dict[t.Hashable, t.Any]: + def _store(self) -> dict[Hashable, Any]: try: store = self._store_cvar.get() except LookupError: @@ -88,10 +91,10 @@ def _store(self) -> dict[t.Hashable, t.Any]: scope_store = store.setdefault(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] = value - def get(self, key: t.Hashable) -> t.Any: + def get(self, key: Hashable) -> Any: return self._store[key]