From 356120dc46dcf29e50d2bb5c5f6121ef197df77d Mon Sep 17 00:00:00 2001 From: Ihor Kalnytskyi Date: Sat, 10 Feb 2024 13:20:31 +0200 Subject: [PATCH] Turn MyPy's no-untyped-def/no-untyped-call on The 'no-untyped-def' and 'no-untyped-call' error codes have been disabled before due to the source code not being ready for them. This patch delivers amends to the code base, and enables these rules. --- pyproject.toml | 4 +- src/picobox/_box.py | 42 ++++++---- src/picobox/_scopes.py | 10 +-- src/picobox/_stack.py | 138 ++++++++++++++++++--------------- src/picobox/ext/asgiscopes.py | 9 ++- src/picobox/ext/flaskscopes.py | 20 +++-- 6 files changed, 125 insertions(+), 98 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 78cca62..6ed659a 100644 --- a/pyproject.toml +++ b/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", ] diff --git a/src/picobox/_box.py b/src/picobox/_box.py index b9d0222..ef779a9 100644 --- a/src/picobox/_box.py +++ b/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,6 +88,11 @@ 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 @@ -88,11 +100,6 @@ def put( # 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,7 +185,7 @@ 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__"): @@ -181,7 +193,7 @@ def decorator(fn): 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( diff --git a/src/picobox/_scopes.py b/src/picobox/_scopes.py index 1fc8a73..bb619b0 100644 --- a/src/picobox/_scopes.py +++ b/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: diff --git a/src/picobox/_stack.py b/src/picobox/_stack.py index 7860421..ff61619 100644 --- a/src/picobox/_stack.py +++ b/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"" 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_) diff --git a/src/picobox/ext/asgiscopes.py b/src/picobox/ext/asgiscopes.py index 89138ff..19c53eb 100644 --- a/src/picobox/ext/asgiscopes.py +++ b/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 diff --git a/src/picobox/ext/flaskscopes.py b/src/picobox/ext/flaskscopes.py index 4fe9789..6c49925 100644 --- a/src/picobox/ext/flaskscopes.py +++ b/src/picobox/ext/flaskscopes.py @@ -1,5 +1,6 @@ """Scopes for Flask framework.""" +import typing as t import uuid import flask @@ -10,9 +11,8 @@ class _flaskscope(picobox.Scope): """A base class for Flask scopes.""" - _store = None - - def __init__(self): + def __init__(self, store: object) -> None: + self._store = store # Both application and request scopes are merely proxies to # corresponding storage objects in Flask. This means multiple # scope instances will share the same storage object under the @@ -21,7 +21,7 @@ def __init__(self): # distinguish dependencies stored by different scope instances. self._uuid = str(uuid.uuid4()) - def set(self, key, value): + def set(self, key: t.Hashable, value: t.Any) -> None: try: dependencies = self._store.__dependencies__ except AttributeError: @@ -34,7 +34,7 @@ def set(self, key, value): dependencies[key] = value - def get(self, key): + def get(self, key: t.Hashable) -> t.Any: try: rv = self._store.__dependencies__[self._uuid][key] except (AttributeError, KeyError): @@ -60,9 +60,8 @@ class application(_flaskscope): .. versionadded:: 2.2 """ - @property - def _store(self): - return flask.current_app + def __init__(self) -> None: + super().__init__(flask.current_app) class request(_flaskscope): @@ -78,6 +77,5 @@ class request(_flaskscope): .. versionadded:: 2.2 """ - @property - def _store(self): - return flask.g + def __init__(self) -> None: + super().__init__(flask.g)