feat: typer integration

docs/integrations/typer.rst

@@ -4,9 +4,12 @@ Typer
 =================================
 
 
-Though it is not required, you can use dishka-click integration. It features automatic injection to command handlers
-In contrast with other integrations there is no scope management.
+Though it is not required, you can use dishka-click integration. It features:
 
+* automatic APP and REQUEST scope management
+* automatic injection of dependencies into handler function
+* passing ``typer.Context`` object as a context data to providers
+* you can still request ``typer.Context`` as with usual typer commands
 
 
 How to use
@@ -16,34 +19,48 @@ How to use
 
 .. code-block:: python
 
-    from dishka.integrations.typer import setup_dishka, inject
+    from dishka.integrations.typer import setup_dishka, inject, TyperProvider
 
 
-2. Create a container and set it up with the typer app. Pass ``auto_inject=True`` if you do not want to use the ``@inject`` decorator explicitly.
+2. Create provider. You can use ``typer.Context`` as a factory parameter to access on REQUEST-scope.
 
 .. code-block:: python
 
-    app = typer.Typer()
+    class YourProvider(Provider):
+        @provide(scope=Scope.REQUEST)
+        def command_name(self, context: typer.Context) -> str | None:
+            return context.command.name
 
-    container = make_container(MyProvider())
-    setup_dishka(container=container, app=app, auto_inject=True)
 
+3. *(optional)* Use ``TyperProvider()`` when creating your container if you are using ``typer.Context`` in providers.
+
+.. code-block:: python
 
-3. Mark those of your command handlers parameters which are to be injected with ``FromDishka[]``
+    container = make_async_container(YourProvider(), Typerprovider())
+
+
+4. Mark those of your command handlers parameters which are to be injected with ``FromDishka[]``. You can use ``typer.Context`` in the command as usual.
 
 .. code-block:: python
 
     app = typer.Typer()
 
     @app.command(name="greet")
-    def greet_user(greeter: FromDishka[Greeter], user: Annotated[str, typer.Argument()]) -> None:
+    def greet_user(ctx: typer.Context, greeter: FromDishka[Greeter], user: Annotated[str, typer.Argument()]) -> None:
         ...
 
-3a. *(optional)* decorate them using ``@inject`` if you want to mark commands explicitly
+4a. *(optional)* decorate commands using ``@inject`` if you want to mark them explicitly
 
 .. code-block:: python
 
     @app.command(name="greet")
     @inject  # Use this decorator *before* the command decorator
     def greet_user(greeter: FromDishka[Greeter], user: Annotated[str, typer.Argument()]) -> None:
         ...
+
+
+5. *(optional)* Use ``auto_inject=True`` when setting up dishka to automatically inject dependencies into your command handlers. When doing this, ensure all commands have already been created when you call setup. This limitation is not required when using ``@inject`` manually.
+
+.. code-block:: python
+
+    setup_dishka(container=container, app=app, auto_inject=True)

examples/integrations/typer_app/with_auto_inject.py

@@ -5,21 +5,28 @@
 from typing import Annotated, Protocol
 import typer
 from functools import partial
-from dishka import make_container, Provider
+from dishka import make_container, Provider, provide
 from dishka.entities.scope import Scope
-from dishka.integrations.typer import FromDishka, inject, setup_dishka
+from dishka.integrations.typer import FromDishka, TyperProvider, inject, setup_dishka
 
 
 class Greeter(Protocol):
     """Protocol to be extra generic on our greeting infrastructure."""
     def __call__(self, text: str) -> None: ...
 
 
-provider = Provider(scope=Scope.APP)
+class ColorfulProvider(Provider):
 
-# We provide an advanced greeting experience with `typer.secho`
-# For a less advanced implementation, we could use `print`
-provider.provide(lambda: partial(typer.secho, fg="blue"), provides=Greeter)
+    @provide(scope=Scope.REQUEST)  # We need Scope.REQUEST for the context
+    def greeter(self, context: typer.Context) -> Greeter:
+        if context.command.name == "hello":
+            # Hello should most certainly be blue
+            return partial(typer.secho, fg="blue")
+        if context.command.name == "goodbye":
+            # Goodbye should be red
+            return partial(typer.secho, fg="red")
+        # Unexpected commands can be yellow
+        return partial(typer.secho, fg="yellow")
 
 
 app = typer.Typer()
@@ -41,9 +48,20 @@ def goodbye(greeter: FromDishka[Greeter], name: str, formal: bool = False) -> No
         greeter(f"Bye {name}!")
 
 
+@app.command()
+def hi(
+    greeter: FromDishka[Greeter],
+    name: Annotated[str, typer.Argument(..., help="The name to greet")],
+) -> None:
+    greeter(f"Hi {name}")
+
+
+# Build the container with the `TyperProvider` to get the `typer.Context`
+# parameter in REQUEST providers
+container = make_container(ColorfulProvider(scope=Scope.REQUEST), TyperProvider())
+
 # Setup dishka to inject the dependency container
 # *Must* be after defining the commands when using auto_inject
-container = make_container(provider)
 setup_dishka(container=container, app=app, auto_inject=True)
 
 

src/dishka/integrations/typer.py

@@ -7,27 +7,109 @@
 ]
 
 from collections.abc import Callable
-from typing import Final, TypeVar
+from inspect import Parameter
+from typing import Final, ParamSpec, TypeVar, cast, get_type_hints
 
+import click
 import typer
-from click import get_current_context
 
-from dishka import Container, FromDishka
+from dishka import Container, FromDishka, Scope
+from dishka.dependency_source.make_context_var import from_context
+from dishka.provider import Provider
 from .base import is_dishka_injected, wrap_injection
 
 T = TypeVar("T")
+P = ParamSpec("P")
 CONTAINER_NAME: Final = "dishka_container"
 
 
-def inject(func: Callable[..., T]) -> Callable[..., T]:
-    return wrap_injection(
-        func=func,
-        container_getter=lambda _, __: get_current_context().meta[
-            CONTAINER_NAME
-        ],
-        remove_depends=True,
-        is_async=False,
+def inject(func: Callable[P, T]) -> Callable[P, T]:
+    # Try to isolate a parameter in the function signature requesting a
+    # typer.Context
+    hints = get_type_hints(func)
+    param_name = next(
+        (name for name, hint in hints.items() if hint is typer.Context),
+        None,
     )
+    if param_name is None:
+        # When the handler does not request a typer.Context, we need to add it
+        # in our wrapper to be able to inject it in into the container
+        def wrapper(context: typer.Context, *args: P.args, **kwargs: P.kwargs) -> T:
+            # Inject the typer context into the container
+            container: Container = context.meta[CONTAINER_NAME]
+            with container({typer.Context: context}, scope=Scope.REQUEST) as new_container:
+                context.meta[CONTAINER_NAME] = new_container
+
+                # Then proceed with the regular injection logic
+                injected_func = wrap_injection(
+                    func=func,
+                    container_getter=lambda _, __: click.get_current_context().meta[CONTAINER_NAME],
+                    remove_depends=True,
+                    is_async=False,
+                )
+                return injected_func(*args, **kwargs)
+
+        # We reuse the logic of `wrap_injection`, but only to build the expected
+        # signature (removing dishka dependencies, adding the typer.Context
+        # parameter)
+        expected_signature = wrap_injection(
+            func=func,
+            container_getter=lambda _, __: click.get_current_context().meta[CONTAINER_NAME],
+            additional_params=[Parameter(name="context", kind=Parameter.POSITIONAL_ONLY, annotation=typer.Context)],
+            remove_depends=True,
+            is_async=False,
+        )
+
+    else:
+        # When the handler requests a typer.Context, we just need to find it and
+        # inject
+        def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
+            # Get the context from the existing argument
+            if param_name in kwargs:
+                context: typer.Context = kwargs[param_name]  # type: ignore[assignment]
+            else:
+                maybe_context = next(
+                    # Even though we type `typer.Context`, we get a
+                    # `click.Context` instance
+                    (arg for arg in args if isinstance(arg, click.Context)), None,
+                )
+                if maybe_context is None:
+                    raise RuntimeError(f"Context argument {param_name} not provided at runtime.")
+                context = maybe_context
+
+            # Inject the typer context into the container
+            container: Container = context.meta[CONTAINER_NAME]
+            with container({typer.Context: context}, scope=Scope.REQUEST) as new_container:
+                context.meta[CONTAINER_NAME] = new_container
+
+                # Then proceed with the regular injection logic
+                injected_func = wrap_injection(
+                    func=func,
+                    container_getter=lambda _, __: click.get_current_context().meta[CONTAINER_NAME],
+                    remove_depends=True,
+                    is_async=False,
+                )
+                return injected_func(*args, **kwargs)
+
+        # This time, no need to add a parameter to the signature
+        expected_signature = wrap_injection(
+            func=func,
+            container_getter=lambda _, __: get_current_context().meta[CONTAINER_NAME],
+            remove_depends=True,
+            is_async=False,
+        )
+
+    # Copy over all metadata from the expected injected function's signature to
+    # our wrapper
+    wrapper.__dishka_injected__ = True  # type: ignore[attr-defined]
+    wrapper.__name__ = expected_signature.__name__
+    wrapper.__qualname__ = expected_signature.__qualname__
+    wrapper.__doc__ = expected_signature.__doc__
+    wrapper.__module__ = expected_signature.__module__
+    wrapper.__annotations__ = expected_signature.__annotations__
+    wrapper.__signature__ = expected_signature.__signature__  # type: ignore[attr-defined]
+
+    return cast(Callable[P, T], wrapper)
 
 
 def _inject_commands(app: typer.Typer) -> None:
@@ -42,6 +124,10 @@ def _inject_commands(app: typer.Typer) -> None:
             _inject_commands(group.typer_instance)
 
 
+class TyperProvider(Provider):
+    context = from_context(provides=typer.Context, scope=Scope.APP)
+
+
 def setup_dishka(
     container: Container,
     app: typer.Typer,

tests/integrations/typer/test_typer.py

@@ -6,29 +6,39 @@
 import typer
 from typer.testing import CliRunner
 
-from dishka import FromDishka, make_container
-from dishka.integrations.typer import inject, setup_dishka
+from dishka import FromDishka, Scope, make_container
+from dishka.dependency_source.make_factory import provide
+from dishka.integrations.typer import TyperProvider, inject, setup_dishka
 from dishka.provider import Provider
 from ..common import (
     APP_DEP_VALUE,
+    REQUEST_DEP_VALUE,
     AppDep,
     AppMock,
     AppProvider,
+    RequestDep,
 )
 
 AppFactory = Callable[
     [Callable[..., Any], Provider], AbstractContextManager[typer.Typer],
 ]
 
 
+class SampleProvider(Provider):
+
+    @provide(scope=Scope.REQUEST)
+    def invoked_subcommand(self, context: typer.Context) -> str | None:
+        return context.command.name
+
+
 @contextmanager
 def dishka_app(
     handler: Callable[..., Any], provider: Provider,
 ) -> Iterator[typer.Typer]:
     app = typer.Typer()
     app.command(name="test")(inject(handler))
 
-    container = make_container(provider)
+    container = make_container(provider, SampleProvider(), TyperProvider())
     setup_dishka(container=container, app=app, finalize_container=False)
 
     yield app
@@ -42,7 +52,7 @@ def dishka_auto_app(
     app = typer.Typer()
     app.command(name="test")(handler)
 
-    container = make_container(provider)
+    container = make_container(provider, SampleProvider(), TyperProvider())
     setup_dishka(
         container=container,
         app=app,
@@ -63,7 +73,7 @@ def dishka_nested_group_app(
     group.command(name="sub")(handler)
     app.add_typer(group, name="test")
 
-    container = make_container(provider)
+    container = make_container(provider, SampleProvider(), TyperProvider())
     setup_dishka(
         container=container,
         app=app,
@@ -153,3 +163,57 @@ def test_app_dependency_with_nested_groups_and_option(
             APP_DEP_VALUE, "Wade", "Wilson",
         )
         app_provider.request_released.assert_not_called()
+
+
+def handle_with_context_dependency(
+    a: FromDishka[RequestDep],
+    mock: FromDishka[AppMock],
+    command_name: FromDishka[str | None],
+) -> None:
+    """Function using a dependency """
+    mock(a)
+    assert command_name == "test"
+
+
+def handle_with_context_dependency_and_context_arg(
+    ctx_arg: typer.Context,
+    a: FromDishka[RequestDep],
+    mock: FromDishka[AppMock],
+    command_name: FromDishka[str | None],
+) -> None:
+    mock(a)
+    assert command_name == "test"
+    assert ctx_arg.command.name == "test"
+
+
+def handle_with_context_dependency_and_context_kwarg(
+    a: FromDishka[RequestDep],
+    mock: FromDishka[AppMock],
+    command_name: FromDishka[str | None],
+    *,
+    ctx_kwarg: typer.Context,  # Force as kwargs
+) -> None:
+    mock(a)
+    assert command_name == "test"
+    assert ctx_kwarg.command.name == "test"
+
+
+@pytest.mark.parametrize("app_factory", [dishka_app, dishka_auto_app])
+@pytest.mark.parametrize(
+    "handler_function", [
+        handle_with_context_dependency_and_context_arg,
+        handle_with_context_dependency_and_context_kwarg,
+        handle_with_context_dependency,
+    ],
+)
+def test_request_dependency_with_context_command(
+    app_provider: AppProvider,
+    app_factory: AppFactory,
+    handler_function: Callable[..., Any],
+) -> None:
+    runner = CliRunner()
+    with app_factory(handler_function, app_provider) as command:
+        result = runner.invoke(command, ["test"])
+        assert result.exit_code == 0, result.stdout
+        app_provider.app_mock.assert_called_with(REQUEST_DEP_VALUE)
+        app_provider.request_released.assert_called_once()