refactoring: ♻️ AsyncSemaphore

Author: remimd

injection/__init__.pyi

@@ -149,7 +149,7 @@ class Module:
         always be the same.
         """
 
-    def scoped[**P, T](
+    def scoped[T](
         self,
         scope_name: str,
         /,

injection/_core/common/asynchronous.py

@@ -1,7 +1,26 @@
 from abc import abstractmethod
 from collections.abc import Awaitable, Callable, Generator
 from dataclasses import dataclass
-from typing import Any, AsyncContextManager, NoReturn, Protocol, runtime_checkable
+from typing import (
+    Any,
+    AsyncContextManager,
+    NoReturn,
+    Protocol,
+    runtime_checkable,
+)
+
+AsyncSemaphore: Callable[[int], AsyncContextManager[Any]]
+
+try:
+    import anyio
+
+except ImportError:  # pragma: no cover
+    import asyncio
+
+    AsyncSemaphore = asyncio.Semaphore
+
+else:
+    AsyncSemaphore = anyio.Semaphore
 
 
 @dataclass(repr=False, eq=False, frozen=True, slots=True)
@@ -45,16 +64,3 @@ async def acall(self, /, *args: P.args, **kwargs: P.kwargs) -> T:
 
     def call(self, /, *args: P.args, **kwargs: P.kwargs) -> T:
         return self.callable(*args, **kwargs)
-
-
-try:
-    import anyio
-
-    def create_semaphore(value: int) -> AsyncContextManager[Any]:
-        return anyio.Semaphore(value)
-
-except ImportError:  # pragma: no cover
-    import asyncio
-
-    def create_semaphore(value: int) -> AsyncContextManager[Any]:
-        return asyncio.Semaphore(value)

injection/_core/injectables.py

@@ -13,10 +13,7 @@
     runtime_checkable,
 )
 
-from injection._core.common.asynchronous import Caller
-from injection._core.common.asynchronous import (
-    create_semaphore as _create_async_semaphore,
-)
+from injection._core.common.asynchronous import AsyncSemaphore, Caller
 from injection._core.scope import (
     Scope,
     get_scope,
@@ -64,7 +61,7 @@ class CacheLogic[T]:
     __semaphore: AsyncContextManager[Any]
 
     def __init__(self) -> None:
-        self.__semaphore = _create_async_semaphore(1)
+        self.__semaphore = AsyncSemaphore(1)
 
     async def aget_or_create[K](
         self,

injection/ext/fastapi.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass, field
 from types import GenericAlias
-from typing import Annotated, Any, TypeAliasType
+from typing import Annotated, Any, TypeAlias, TypeAliasType
 
 from fastapi import Depends
 
@@ -25,16 +25,16 @@ def __call__[T](
     ) -> Any:
         module = module or self.module
         threadsafe = self.threadsafe if threadsafe is None else threadsafe
-        ainstance = module.aget_lazy_instance(cls, default, threadsafe=threadsafe)
+        lazy_instance = module.aget_lazy_instance(cls, default, threadsafe=threadsafe)
 
         async def dependency() -> T:
-            return await ainstance
+            return await lazy_instance
 
         class_name = getattr(cls, "__name__", str(cls))
         dependency.__name__ = f"inject({class_name})"
         return Depends(dependency, use_cache=False)
 
-    def __getitem__(self, params: Any, /) -> Any:
+    def __getitem__[T, *Ts](self, params: T | tuple[T, *Ts], /) -> TypeAlias:
         iter_params = iter(params if isinstance(params, tuple) else (params,))
         cls = next(iter_params)
         return Annotated[cls, self(cls), *iter_params]