Async support
tiangolo opened this issue · 4 comments
tiangolo commented
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 commented
Why not use native asyncio.run()
?
Deltik commented
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)
albireox commented
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.