Enter helper (#31)

* Add `enter` helper

* Update code after rebase

* Change `enter` realisation

* Add `helpers.enter` context manager for resolving dependencies in pytest fixtures

* Fix `enter` typehints

CHANGELOG.md

@@ -6,6 +6,10 @@ We follow [Semantic Versions](https://semver.org/).
 
 -
 
+## Version 0.18.0
+
+- Added `helpers.enter` context manager for resolving dependencies in pytest fixtures
+
 ## Version 0.17.1
 
 - Fixed typehints

README.md

@@ -161,6 +161,24 @@ if __name__ == "__main__":
 
 [Read on the documentation site](https://picodi.readthedocs.io/en/stable/knownissues.html)
 
+#### `helpers.enter(gen)`
+
+Helper to use generator dependencies in a context manager. May be useful in some cases.
+Note that this helper is not needed for regular usage because it ignores the scope of the dependency,
+so resources will not be cached and will be closed after each usage.
+
+```python
+from picodi import helpers
+
+
+def get_db():
+    yield "db connection"
+
+
+with helpers.enter(get_db()) as db:
+    print("processing data in db:", db)
+```
+
 ## License
 
 [MIT](https://github.com/yakimka/picodi/blob/main/LICENSE)

picodi/helpers.py

@@ -6,13 +6,23 @@
 
 import asyncio
 import contextlib
-from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, overload
+import inspect
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    AsyncContextManager,
+    ContextManager,
+    ParamSpec,
+    TypeVar,
+    overload,
+)
 
 import picodi
 from picodi import ManualScope
+from picodi.support import nullcontext
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncGenerator, Callable, Generator
+    from collections.abc import AsyncGenerator, Callable, Coroutine, Generator
     from types import TracebackType
 
 sentinel = object()
@@ -221,3 +231,76 @@ async def async_(
 
 
 lifespan = _Lifespan()
+
+
+@overload
+def enter(dependency: Callable[[], Generator[T, None, None]]) -> ContextManager[T]: ...
+
+
+@overload
+def enter(  # type: ignore[overload-overlap]
+    dependency: Callable[[], AsyncGenerator[T, None] | Coroutine[T, None, None]],
+) -> AsyncContextManager[T]: ...
+
+
+@overload
+def enter(dependency: Callable[[], T]) -> ContextManager[T]: ...
+
+
+def enter(
+    dependency: Callable[
+        [],
+        Coroutine[T, None, None]
+        | Generator[T, None, None]
+        | AsyncGenerator[T, None]
+        | T,
+    ],
+) -> AsyncContextManager[T] | ContextManager[T]:
+    """
+    Create a context manager from a dependency.
+
+    Don't use (or use carefully) in production code. This function is mostly for
+    cases when you can't use :func:`inject` decorator, for example in pytest fixtures.
+
+    :param dependency: dependency to create a context manager from.
+        Like with :func:`Provide` - don't call the dependency function here,
+        just pass it.
+    :return: sync or async context manager.
+
+    Example
+    -------
+
+    .. code-block:: python
+
+        from picodi.helpers import enter
+
+        def get_42():
+            yield 42
+
+        with enter(get_42) as val:
+            assert val == 42
+    """
+    result = dependency()
+
+    if inspect.isasyncgen(result):
+
+        @contextlib.asynccontextmanager
+        @picodi.inject
+        async def async_enter(
+            dep: Any = picodi.Provide(dependency),
+        ) -> AsyncGenerator[Any, None]:
+            yield dep
+
+        return async_enter()
+    if inspect.isgenerator(result):
+
+        @contextlib.contextmanager
+        @picodi.inject
+        def sync_enter(
+            dep: Any = picodi.Provide(dependency),
+        ) -> Generator[Any, None, None]:
+            yield dep
+
+        return sync_enter()
+
+    return nullcontext(result)

picodi/support.py

@@ -7,8 +7,10 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
 from contextlib import AsyncExitStack
 from contextlib import ExitStack as SyncExitStack
+from contextlib import nullcontext as stdlib_nullcontext
 from typing import TYPE_CHECKING, Any, AsyncContextManager, ContextManager
 
 if TYPE_CHECKING:
@@ -91,6 +93,20 @@ def close(self, exc: BaseException | None = None) -> Awaitable:
         return NullAwaitable()
 
 
+class nullcontext(stdlib_nullcontext):  # noqa: N801
+    """
+    A context manager that does nothing.
+
+    This is the :func:`python:contextlib.nullcontext` with the same functionality
+    but with the ability to await ``enter_result`` in async context.
+    """
+
+    async def __aenter__(self) -> Any:
+        if inspect.iscoroutine(self.enter_result):
+            self.enter_result = await self.enter_result
+        return self.enter_result
+
+
 def is_async_environment() -> bool:
     """
     Check if we are in async environment.

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry]
 name = "picodi"
 description = "Simple Dependency Injection for Python"
-version = "0.17.1"
+version = "0.18.0"
 license = "MIT"
 authors = [
   "yakimka"

tests/test_enter_helper.py

@@ -0,0 +1,85 @@
+from picodi import Provide, SingletonScope, dependency, inject
+from picodi.helpers import enter
+
+
+def get_42():
+    return 42
+
+
+def test_enter_sync_gen(closeable):
+    def dep():
+        yield 42
+        closeable.close()
+
+    with enter(dep) as val:
+        assert val == 42
+        assert closeable.is_closed is False
+
+    assert closeable.is_closed is True
+
+
+async def test_enter_async_gen(closeable):
+    async def dep():
+        yield 42
+        closeable.close()
+
+    async with enter(dep) as val:
+        assert val == 42
+        assert closeable.is_closed is False
+
+    assert closeable.is_closed is True
+
+
+def test_enter_injected_sync_gen(closeable):
+    @inject
+    def dep(num: int = Provide(get_42)):
+        yield num
+        closeable.close()
+
+    with enter(dep) as val:
+        assert val == 42
+        assert closeable.is_closed is False
+
+    assert closeable.is_closed is True
+
+
+async def test_enter_injected_async_gen(closeable):
+    @inject
+    async def dep(num: int = Provide(get_42)):
+        yield num
+        closeable.close()
+
+    async with enter(dep) as val:
+        assert val == 42
+        assert closeable.is_closed is False
+
+    assert closeable.is_closed is True
+
+
+def test_singleton_sync_gen_not_closed(closeable):
+    @dependency(scope_class=SingletonScope)
+    def dep():
+        yield 42
+        closeable.close()
+
+    with enter(dep) as val:
+        assert val == 42
+        assert closeable.is_closed is False
+
+    assert closeable.is_closed is False
+
+
+def test_enter_regular_dependency():
+    def dep():
+        return 42
+
+    with enter(dep) as val:
+        assert val == 42
+
+
+async def test_enter_regular_dependency_async():
+    async def dep():
+        return 42
+
+    async with enter(dep) as val:
+        assert val == 42