feat: typer integration

docs/integrations/index.rst

@@ -25,6 +25,7 @@ You can create custom integrations for your framework of choice.
    starlette
    taskiq
    telebot
+   typer
    adding_new
 
 .. list-table:: Built-in frameworks integrations
@@ -43,7 +44,7 @@ You can create custom integrations for your framework of choice.
    * - :ref:`grpcio`
      - :ref:`Aiogram_dialog`
      - :ref:`FastStream`
-     -
+     - :ref:`Typer`
 
    * - :ref:`Fastapi`
      - :ref:`pyTelegramBotAPI<telebot>`

docs/integrations/typer.rst

@@ -0,0 +1,66 @@
+.. _typer:
+
+Typer
+=================================
+
+
+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
+****************
+
+1. Import
+
+.. code-block:: python
+
+    from dishka.integrations.typer import setup_dishka, inject, TyperProvider
+
+
+2. Create provider. You can use ``typer.Context`` as a factory parameter to access on REQUEST-scope.
+
+.. code-block:: python
+
+    class YourProvider(Provider):
+        @provide(scope=Scope.REQUEST)
+        def command_name(self, context: typer.Context) -> str | None:
+            return context.command.name
+
+
+3. *(optional)* Use ``TyperProvider()`` when creating your container if you are using ``typer.Context`` in providers.
+
+.. code-block:: python
+
+    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(ctx: typer.Context, greeter: FromDishka[Greeter], user: Annotated[str, typer.Argument()]) -> None:
+        ...
+
+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

@@ -0,0 +1,70 @@
+"""Example typer app greeting a user, taken from
+https://typer.tiangolo.com/#example-upgrade with small modifications.
+"""
+
+from typing import Annotated, Protocol
+import typer
+from functools import partial
+from dishka import make_container, Provider, provide
+from dishka.entities.scope import Scope
+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: ...
+
+
+class ColorfulProvider(Provider):
+
+    @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()
+
+
+@app.command()
+def hello(
+    greeter: FromDishka[Greeter],
+    name: Annotated[str, typer.Argument(..., help="The name to greet")],
+) -> None:
+    greeter(f"Hello {name}")
+
+
+@app.command()
+def goodbye(greeter: FromDishka[Greeter], name: str, formal: bool = False) -> None:
+    if formal:
+        greeter(f"Goodbye Ms. {name}. Have a good day.")
+    else:
+        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
+setup_dishka(container=container, app=app, auto_inject=True)
+
+
+if __name__ == "__main__":
+    app()
+

examples/integrations/typer_app/with_inject.py

@@ -0,0 +1,53 @@
+"""Example typer app greeting a user, taken from
+https://typer.tiangolo.com/#example-upgrade with small modifications.
+"""
+
+from typing import Annotated, Protocol
+import typer
+from functools import partial
+from dishka import make_container, Provider
+from dishka.entities.scope import Scope
+from dishka.integrations.typer import FromDishka, 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)
+
+# 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)
+
+
+app = typer.Typer()
+
+
+# Setup dishka to inject the dependency container
+# Can be done before or after defining the commands when using @inject manually
+container = make_container(provider)
+setup_dishka(container=container, app=app, auto_inject=False)
+
+
+@app.command()
+@inject
+def hello(
+    greeter: FromDishka[Greeter],
+    name: Annotated[str, typer.Argument(..., help="The name to greet")],
+) -> None:
+    greeter(f"Hello {name}")
+
+
+@app.command()
+@inject
+def goodbye(greeter: FromDishka[Greeter], name: str, formal: bool = False) -> None:
+    if formal:
+        greeter(f"Goodbye Ms. {name}. Have a good day.")
+    else:
+        greeter(f"Bye {name}!")
+
+
+if __name__ == "__main__":
+    app()

noxfile.py

@@ -58,6 +58,8 @@ def get_tests(self) -> str:
     IntegrationEnv("taskiq", "latest"),
     IntegrationEnv("telebot", "415"),
     IntegrationEnv("telebot", "latest"),
+    IntegrationEnv("typer", "0151"),
+    IntegrationEnv("typer", "latest"),
 ]
 
 

requirements/typer-0151.txt

@@ -0,0 +1,2 @@
+-r test.txt
+typer==0.15.1

requirements/typer-latest.txt

@@ -0,0 +1,2 @@
+-r test.txt
+typer

src/dishka/integrations/typer.py

@@ -0,0 +1,146 @@
+"""Integration for Typer https://typer.tiangolo.com"""
+
+__all__ = [
+    "FromDishka",
+    "inject",
+    "setup_dishka",
+]
+
+from collections.abc import Callable
+from inspect import Parameter
+from typing import Final, ParamSpec, TypeVar, cast, get_type_hints
+
+import click
+import typer
+
+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[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:
+    for command in app.registered_commands:
+        if command.callback is not None and not is_dishka_injected(
+            command.callback,
+        ):
+            command.callback = inject(command.callback)
+
+    for group in app.registered_groups:
+        if group.typer_instance is not 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,
+    *,
+    finalize_container: bool = True,
+    auto_inject: bool = False,
+) -> None:
+    @app.callback()
+    def inject_dishka_container(context: typer.Context) -> None:
+        context.meta[CONTAINER_NAME] = container
+
+        if finalize_container:
+            context.call_on_close(container.close)
+
+    if auto_inject:
+        _inject_commands(app)

tests/integrations/typer/__init__.py

@@ -0,0 +1,3 @@
+import pytest
+
+pytest.importorskip("typer")