Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Drop 'typing_extensions', use modern annotations #100

Merged
merged 1 commit into from
Jun 29, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand All @@ -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
42 changes: 22 additions & 20 deletions src/picobox/_box.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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__"):
Expand Down Expand Up @@ -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]

Expand Down Expand Up @@ -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:
Expand Down
32 changes: 18 additions & 14 deletions src/picobox/_scopes.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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]


Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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)
50 changes: 26 additions & 24 deletions src/picobox/_stack.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)

Expand Down Expand Up @@ -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()

Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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_)
Loading
Loading