Added async_command to typer.Typer

typer/main.py

@@ -1,10 +1,11 @@
+import asyncio
 import inspect
 import os
 import sys
 import traceback
 from datetime import datetime
 from enum import Enum
-from functools import update_wrapper
+from functools import update_wrapper, wraps
 from pathlib import Path
 from traceback import FrameSummary, StackSummary
 from types import TracebackType
@@ -18,6 +19,7 @@
 from .models import (
     AnyType,
     ArgumentInfo,
+    AsyncCommandFunctionType,
     CommandFunctionType,
     CommandInfo,
     Default,
@@ -32,6 +34,7 @@
     ParameterInfo,
     ParamMeta,
     Required,
+    RunFunction,
     TyperInfo,
 )
 from .utils import get_params_from_function
@@ -257,6 +260,60 @@ def decorator(f: CommandFunctionType) -> CommandFunctionType:
 
         return decorator
 
+    def async_command(
+        self,
+        name: Optional[str] = None,
+        *,
+        run_func: RunFunction = lambda f, *args, **kwargs: asyncio.run(
+            f(*args, **kwargs)
+        ),
+        cls: Optional[Type[click.Command]] = None,
+        context_settings: Optional[Dict[Any, Any]] = None,
+        help: Optional[str] = None,
+        epilog: Optional[str] = None,
+        short_help: Optional[str] = None,
+        options_metavar: str = "[OPTIONS]",
+        add_help_option: bool = True,
+        no_args_is_help: bool = False,
+        hidden: bool = False,
+        deprecated: bool = False,
+    ) -> Callable[[AsyncCommandFunctionType], AsyncCommandFunctionType]:
+        """Same arguments as command but works with async functions."""
+        # Dynamically import either anyio or asyncio
+
+        def decorator(async_func: AsyncCommandFunctionType) -> AsyncCommandFunctionType:
+            # Now we make a function that turns the async
+            # function into a synchronous function.
+            # By wrapping async_func we preserve the
+            # meta characteristics typer needs to create
+            # a good interface, such as the description and
+            # argument type hints
+            @wraps(async_func)
+            def sync_func(*args: Any, **kwargs: Any) -> Any:
+                return run_func(async_func, *args, **kwargs)
+
+            # Now use self.command as normal to register the
+            # synchronous function
+            self.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,
+            )(sync_func)
+
+            # We return the async function unmodifed,
+            # so its library functionality is preserved
+            return async_func
+
+        return decorator
+
     def add_typer(
         self,
         typer_instance: "Typer",

typer/models.py

@@ -4,9 +4,11 @@
     TYPE_CHECKING,
     Any,
     Callable,
+    Coroutine,
     Dict,
     List,
     Optional,
+    Protocol,
     Sequence,
     Type,
     TypeVar,
@@ -74,6 +76,23 @@ def __bool__(self) -> bool:
 DefaultType = TypeVar("DefaultType")
 
 CommandFunctionType = TypeVar("CommandFunctionType", bound=Callable[..., Any])
+AsyncCommandFunctionType = TypeVar(
+    "AsyncCommandFunctionType", bound=Callable[..., Coroutine[None, None, Any]]
+)
+
+
+class RunFunction(Protocol):
+    """
+    Defines a run function as the following.
+
+    Examples:
+        asyncio: `run_func = lambda f, *args, **kwargs: asyncio.run(f(*args, **kwargs))`
+        anyio: `run_func = lambda f, *args, **kwargs: anyio.run(f, *args, **kwargs)`
+        trio: `run_func = lambda f, *args, **kwargs: trio.run(f, *args, **kwargs)`
+    """
+
+    def __call__(self, f: AsyncCommandFunctionType, *args: Any, **kwargs: Any) -> Any:
+        ...
 
 
 def Default(value: DefaultType) -> DefaultType: