Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Async support #950

Open
1 task done
tiangolo opened this issue Aug 23, 2024 · 3 comments
Open
1 task done

Async support #950

tiangolo opened this issue Aug 23, 2024 · 3 comments
Assignees
Labels
feature New feature, enhancement or request

Comments

@tiangolo
Copy link
Member

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
Copy link

gaby commented Aug 25, 2024

Why not use native asyncio.run() ?

@svlandeg svlandeg added the feature New feature, enhancement or request label Sep 4, 2024
@Deltik
Copy link

Deltik commented Sep 9, 2024

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
Copy link

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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
feature New feature, enhancement or request
Projects
None yet
Development

No branches or pull requests

5 participants