Replies: 79 comments
-
Just found pallets/click#85 (comment) which may help me |
Beta Was this translation helpful? Give feedback.
-
Seeing as how FastAPI is an async framework, having an async CLI seems logical. The main reason being sharing code from the CLI and Web entry points. You can of course use the asgiref.sync.async_to_sync converter helpers to call existing async methods from the CLI but there are complications here and it makes your cli code clunky. I replaced typer with smurfix/trio-click (which is asyncclick on Pypi) and it works great, but of course this is just async click, not the cool typer implementation. Forking typer and replaceing all |
Beta Was this translation helpful? Give feedback.
-
I will update asyncclick to the latest |
Beta Was this translation helpful? Give feedback.
-
Thanks @smurfix. Once you have asyncclick updated, if @tiangolo doesn't have a nice async typer by then, perhaps ill make a good |
Beta Was this translation helpful? Give feedback.
-
@mreschke I made a pull request to this repo that gets you most of the way to async. #128 |
Beta Was this translation helpful? Give feedback.
-
@mreschke I've updated @jessekrubin's PR to remove the conflicts with master, in case you find it useful. |
Beta Was this translation helpful? Give feedback.
-
Thanks guys. Ill need some time to pull it all in and prototype this instead of asyncclick. If this all works out what is the probability of merging this request and making it a part of this official typer repo. Optional async would be perfect. I really hate to fork permanently. |
Beta Was this translation helpful? Give feedback.
-
We all need this |
Beta Was this translation helpful? Give feedback.
-
I agree with @mreschke, we tightly couple all of our code and actually use Type CLI to call our uvicorn/guinicorn using various "management" commands. Ran into this once we wanted to use some of the async calls we have. |
Beta Was this translation helpful? Give feedback.
-
Hi :) |
Beta Was this translation helpful? Give feedback.
-
@neimad1985 I don't think async is PR-ed in yet, but I use async with typer all the time by just running the async processes from within my sync functions once the parsing is done. It works for most basic things. |
Beta Was this translation helpful? Give feedback.
-
Thanks for the quick answer @jessekrubin |
Beta Was this translation helpful? Give feedback.
-
from asyncio import run as aiorun
import typer
async def _main(name: str):
typer.echo(f"Hello {name}")
def main(name: str = typer.Argument("Wade Wilson")):
aiorun(_main(name=name))
if __name__ == "__main__":
typer.run(main) |
Beta Was this translation helpful? Give feedback.
-
Ok thanks, that's exactly what I was thinking. |
Beta Was this translation helpful? Give feedback.
-
@neimad1985 A decorator might help you: from functools import wraps
import anyio
def run_async(func):
@wraps(func)
def wrapper(*args, **kwargs):
async def coro_wrapper():
return await func(*args, **kwargs)
return anyio.run(coro_wrapper)
return wrapper
@run_async
async def main(name: str = typer.Argument("Wade Wilson")):
typer.echo(f"Hello {name}") You can even have async completions: import click
def async_completion(func):
func = run_async(func)
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except (click.exceptions.Abort, click.exceptions.Exit):
return []
return wrapper
async def list_users() -> List[str]:
...
@run_async
async def main(
name: str = typer.Argument("Wade Wilson", autocompletion=async_completion(list_users))
):
typer.echo(f"Hello {name}") |
Beta Was this translation helpful? Give feedback.
-
@rafalkrupinski, that person has nothing to do with this project he is just there to waste everyone's time, because he only uses async when using "web framework" and nothing else. |
Beta Was this translation helpful? Give feedback.
-
Hmm you seem to lack context or interpretation. I wasn't even responding to just you, but also to someone else who was comparing Typer with FastAPI, hence my response was focused on the comparison between the two. But interesting that you are able to instantly infer my whole dev experience from this brief conversation. This issue has been open for 3 years with no traction forward so I'm just trying to figure out why that is the case :) since I don't think there's been any official positioning about it. I'm pretty sure my questions won't change the maintainers' mind if they are leaning towards incorporating this into the package, when they get time to come here it should actually help if it doesn't get derailed by this kind of comment. Also, I've even helped formulate a solution for it that could be incorporated in the official implementation here... In any case, just to address your "answer", it doesn't matter if there are other async-only libraries and/or if they have to do with web. The point is that adding async capabilities to Typer means there should be an appeal for it to handle concurrency. This means now Typer will have to deal with the async loop in some form. If this hasn't been added yet, maybe the maintainer's philosophy is that this isn't the case and that concurrency should be handled by the user. |
Beta Was this translation helpful? Give feedback.
-
Hello all! Thanks for all the discussion and workarounds. I want to add support for async in Typer, without making it required or default, using AnyIO as an optional dependency. In FastAPI/Starlette, AnyIO is required, as everything is just async underneath. But in Typer I don't want to force people to use AnyIO when they don't really need it. That makes the whole thing a bit more complex as I need to add all the logic to conditionally use or not AnyIO and do async or not. This is one of the top priorities for Typer, along with a code reference (API reference). I'm finishing some things in FastAPI, I have some things in SQLModel too, and then I'll continue with Typer stuff. |
Beta Was this translation helpful? Give feedback.
-
AsyncTyper for anyio (Still with typing problems) import inspect
from functools import partial, wraps
import anyio
import asyncer
import typer
from typer import Typer
class AsyncTyper(Typer):
@staticmethod
def maybe_run_async(decorator, f):
if inspect.iscoroutinefunction(f):
@wraps(f)
def runner(*args, **kwargs):
return asyncer.runnify(f)(*args, **kwargs)
decorator(runner)
else:
decorator(f)
return f
def callback(self, *args, **kwargs):
decorator = super().callback(*args, **kwargs)
return partial(self.maybe_run_async, decorator)
def command(self, *args, **kwargs):
decorator = super().command(*args, **kwargs)
return partial(self.maybe_run_async, decorator)
app = AsyncTyper()
@app.command()
async def async_hello(name: str, last_name: str = "") -> None:
await anyio.sleep(1)
typer.echo(f"Hello World {name} {last_name}")
@app.command()
def hello() -> None:
print("Hello World")
if __name__ == "__main__":
app() |
Beta Was this translation helpful? Give feedback.
-
I ran into a related problem when running async unit tests with pytest-asyncio because by the time the test calls Would love your feedback on whether there is a better way to go about this. # cli/commands/generate.py
...
@app.command("profiles")
@embed_event_loop
async def generate_profiles():
"""
Generates profile data using Random User Generator API (https://randomuser.me/) and
adds them to the database.
"""
...
# test/integration/cli/test_generate.py
...
def test_generate_profiles_happy_path(runner: CliRunner):
result = runner.invoke(["generate", "profiles"])
assert result.exception is None
...
@embed_event_loop
async def verify():
# Verify that the profiles were saved correctly to the database.
async with db_service.session() as session:
...
verify()
... where |
Beta Was this translation helpful? Give feedback.
-
Hello @tiangolo, At least for me, by putting AnyIO as a dependency is not a problem. Actually, an async variant of the typer with specialyzed algorythm to run async functions could be enough. In this case a check if AnyIO is importable could be put in the AsyncTyper constructor. Thank you for these libraries you develop. |
Beta Was this translation helpful? Give feedback.
-
@borissmidt's PR is still open. Not sure about 'hostility' |
Beta Was this translation helpful? Give feedback.
-
Any news on the async support? Any way we can help? |
Beta Was this translation helpful? Give feedback.
-
with syncer:
pip install syncer usage:import os
import typer
from faststream.nats import NatsBroker
from loguru import logger
from syncer import sync
app = typer.Typer()
HOST = os.getenv("NATS_HOST", default="nats://localhost:4222")
async def aio_pub(host: str, msg: str, subject: str):
async with NatsBroker(host) as broker:
logger.debug(f"publish message: {msg}")
await broker.publish(msg, subject=subject)
@app.command("pub")
def pub_task(host: str = HOST):
#
# here! converter: async -> sync
#
f_pub = sync(aio_pub)
# sync call:
f_pub(host, "Hi async wrap!", "test")
logger.debug("pub task done")
if __name__ == "__main__":
app() |
Beta Was this translation helpful? Give feedback.
-
any news on this? |
Beta Was this translation helpful? Give feedback.
-
There is a new issue, #950, meant to introduce asyncio support to Typer. Besides that, I made a new type-annotated wrapper around |
Beta Was this translation helpful? Give feedback.
-
In case you need full typing solution from snippet above import inspect
import typing
from functools import partial, wraps
import asyncer
from typer import Typer
from typer.core import TyperCommand
from typer.models import CommandFunctionType, Default
class AsyncTyper(Typer):
@staticmethod
def maybe_run_async(decorator, f):
if inspect.iscoroutinefunction(f):
@wraps(f)
def runner(*args, **kwargs):
return asyncer.runnify(f)(*args, **kwargs)
decorator(runner)
else:
decorator(f)
return f
def callback(self, *args, **kwargs):
decorator = super().callback(*args, **kwargs)
return partial(self.maybe_run_async, decorator)
def command(
self,
name: str | None = None,
*,
cls: type[TyperCommand] | None = None,
context_settings: dict | None = None,
help: str | None = None,
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 settings
rich_help_panel: str | None = Default(None),
) -> typing.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 partial(self.maybe_run_async, decorator) |
Beta Was this translation helpful? Give feedback.
-
I switched to golang and didn't really need this anymore. But recently i made a python CLI using cyclopts which is almost the same as Typer but comes with async support out of the box and an easier way to add descriptions to your commands using the Doc strings. I hope this will help others. |
Beta Was this translation helpful? Give feedback.
-
Not to mention it uses |
Beta Was this translation helpful? Give feedback.
-
Let's convert this thread into discussion and use #950 to track the "async support" feature |
Beta Was this translation helpful? Give feedback.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
-
First check
Description
I have existing methods/model functions that use async functions with encode/databases to load data but I'm unable to use these within commands without getting errors such as
RuntimeWarning: coroutine 'something' was never awaited
How can I make make my
@app.command()
functions async friendly?Beta Was this translation helpful? Give feedback.
All reactions