Async support

[tiangolo]

Privileged issue

• I'm @tiangolo or he asked me directly to create an issue here.

Issue Content

I want Typer to have optional support for async functions.

It would depend on having AnyIO (or maybe Asyncer 🤔) installed. If the command function is async, then it would run it with anyio.run().

Maybe autocompletion functions could also be async, so they would have to be checked to see if they need to be called directly or awaited.

[gaby]

Why not use native asyncio.run() ?

[Deltik]

I'm currently using this type-annotated wrapper to make the Typer.callback() and Typer.command() decorators compatible with both async and non-async functions:

import asyncio
import inspect
from collections.abc import Callable, Coroutine
from functools import wraps
from typing import Any, ParamSpec, TypeVar, cast

from typer import Typer
from typer.core import TyperCommand, TyperGroup
from typer.models import CommandFunctionType

P = ParamSpec("P")
R = TypeVar("R")


class AsyncTyper(Typer):
    @staticmethod
    def maybe_run_async(
        decorator: Callable[[CommandFunctionType], CommandFunctionType],
        f: CommandFunctionType,
    ) -> CommandFunctionType:
        if inspect.iscoroutinefunction(f):

            @wraps(f)
            def runner(*args: Any, **kwargs: Any) -> Any:
                return asyncio.run(cast(Callable[..., Coroutine[Any, Any, Any]], f)(*args, **kwargs))

            return decorator(cast(CommandFunctionType, runner))
        return decorator(f)

    # noinspection PyShadowingBuiltins
    def callback(
        self,
        name: str | None = None,
        *,
        cls: type[TyperGroup] | None = None,
        invoke_without_command: bool = False,
        no_args_is_help: bool = False,
        subcommand_metavar: str | None = None,
        chain: bool = False,
        result_callback: Callable[..., Any] | None = None,
        context_settings: dict[Any, Any] | None = None,
        help: str | None = None,  # noqa: A002
        epilog: str | None = None,
        short_help: str | None = None,
        options_metavar: str = "[OPTIONS]",
        add_help_option: bool = True,
        hidden: bool = False,
        deprecated: bool = False,
        rich_help_panel: str | None = None,
    ) -> Callable[[CommandFunctionType], CommandFunctionType]:
        decorator = super().callback(
            name=name,
            cls=cls,
            invoke_without_command=invoke_without_command,
            no_args_is_help=no_args_is_help,
            subcommand_metavar=subcommand_metavar,
            chain=chain,
            result_callback=result_callback,
            context_settings=context_settings,
            help=help,
            epilog=epilog,
            short_help=short_help,
            options_metavar=options_metavar,
            add_help_option=add_help_option,
            hidden=hidden,
            deprecated=deprecated,
            rich_help_panel=rich_help_panel,
        )
        return lambda f: self.maybe_run_async(decorator, f)

    # noinspection PyShadowingBuiltins
    def command(
        self,
        name: str | None = None,
        *,
        cls: type[TyperCommand] | None = None,
        context_settings: dict[Any, Any] | None = None,
        help: str | None = None,  # noqa: A002
        epilog: str | None = None,
        short_help: str | None = None,
        options_metavar: str = "[OPTIONS]",
        add_help_option: bool = True,
        no_args_is_help: bool = False,
        hidden: bool = False,
        deprecated: bool = False,
        rich_help_panel: str | None = None,
    ) -> Callable[[CommandFunctionType], CommandFunctionType]:
        decorator = super().command(
            name=name,
            cls=cls,
            context_settings=context_settings,
            help=help,
            epilog=epilog,
            short_help=short_help,
            options_metavar=options_metavar,
            add_help_option=add_help_option,
            no_args_is_help=no_args_is_help,
            hidden=hidden,
            deprecated=deprecated,
            rich_help_panel=rich_help_panel,
        )
        return lambda f: self.maybe_run_async(decorator, f)

There is no dependency on AnyIO or Asyncer.

[albireox]

This would be a useful addition. Currently I'm using a decorator to run the async callback

def cli_coro(
    signals=(signal.SIGHUP, signal.SIGTERM, signal.SIGINT),
    shutdown_func=None,
):
    """Decorator function that allows defining coroutines with click."""

    def decorator_cli_coro(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            loop = asyncio.get_event_loop()
            if shutdown_func:
                for ss in signals:
                    loop.add_signal_handler(ss, shutdown_func, ss, loop)
            return loop.run_until_complete(f(*args, **kwargs))

        return wrapper

    return decorator_cli_coro

@cli_coro()
async def cli_command(...):

but it would be useful to have an option directly in Typer.

[bckohan]

Strong support for async! Especially for chained sub commands.

I'm not sure it makes sense to support async shell completions though. There's never an instance where more than one parameter needs completion simultaneously and if you need to do multiple I/O bound things at once to complete a parameter you can just start an event loop in your completer function.

[dgarros]

I'd love to see some support for Async in Typer as well

we've been using a simpler version of AsyncTyper successfully for a while now

from __future__ import annotations

import asyncio
import inspect
from functools import partial, wraps
from typing import Any, Callable

from typer import Typer


class AsyncTyper(Typer):
    @staticmethod
    def maybe_run_async(decorator: Callable, func: Callable) -> Any:
        if inspect.iscoroutinefunction(func):

            @wraps(func)
            def runner(*args: Any, **kwargs: Any) -> Any:
                return asyncio.run(func(*args, **kwargs))

            decorator(runner)
        else:
            decorator(func)
        return func

    def callback(self, *args: Any, **kwargs: Any) -> Any:
        decorator = super().callback(*args, **kwargs)
        return partial(self.maybe_run_async, decorator)

    def command(self, *args: Any, **kwargs: Any) -> Any:
        decorator = super().command(*args, **kwargs)
        return partial(self.maybe_run_async, decorator)

[bckohan]

Here's the upstream issue: pallets/click#2033

Seems like true async support is best handled upstream in click? Reading through the discussion here, there are some not insignificant hurdles to bringing async into click - mostly because of context globals.

I also very much want async so I plan to dig into this at the click level when time allows.

[tarruda]

FWIW this is what I'm currently using as a workaround:

import typer
import asyncio
import functools
from typing import Any, Callable, Coroutine, ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")

def syncify(
    func: Callable[P, Coroutine[Any, Any, R]]
) -> Callable[P, R]:

    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        loop = asyncio.get_event_loop()
        return loop.run_until_complete(func(*args, **kwargs))

    return wrapper


@syncify
async def some_async_command(...):
   ...


if __name__ == '__main__':
    typer.run(some_async_command)  # also works with subcommands

[danmichaeljones]

You can also use the syncify function from asyncer for a pretty clean workaround:

from functools import partial

import typer
from asyncer import syncify


@app.command()
@partial(syncify, raise_sync_error=False)
async def main(...):
    ...

if __name__ == "__main__":
    app()