Comments (74)
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.
from typer.
I like @ryanpeach solution. I've made an alternative that doesn't require patching the Typer class directly. You can just inherit from it normally:
from asyncio import run
from functools import wraps
import typer
class AsyncTyper(typer.Typer):
def async_command(self, *args, **kwargs):
def decorator(async_func):
@wraps(async_func)
def sync_func(*_args, **_kwargs):
return run(async_func(*_args, **_kwargs))
self.command(*args, **kwargs)(sync_func)
return async_func
return decorator
app = AsyncTyper()
@app.async_command()
async def my_async_command():
...
@app.command()
async def my_normal_command():
...
if __name__ == "__main__":
app()
from typer.
Actually the decorator @cauebs wrote doesn't make sense to me (maybe I just misunderstand click and anyio). The point is to support running the asynchronous function in two modes:
- Called directly from the cli
- Called as an asynchronous function from another function as a library function
If you just decorate the function with a function that makes it synchronous, you've ruined it.
But also we need argument information preserved.
So I propose the following:
# file: root/__init__.py
from functools import wraps
from asyncio import sleep, run
import typer
# This is a standard decorator that takes arguments
# the same way app.command does but with
# app as the first parameter
def async_command(app, *args, **kwargs):
def decorator(async_func):
# 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, **_kwargs):
return run(async_func(*_args, **_kwargs))
# Now use app.command as normal to register the
# synchronous function
app.command(*args, **kwargs)(sync_func)
# We return the async function unmodifed,
# so its library functionality is preserved
return async_func
return decorator
# as a method injection, app will be replaced as self
# making the syntax exactly the same as it used to be.
# put this all in __init__.py and it will be injected into
# the library project wide
typer.Typer.async_command = async_command
# file: root/some/code.py
import typer
from asyncio import sleep
app=typer.Typer()
# The command we want to be accessible by both
# the async library and the CLI
@app.async_command()
async def foo(bar: str = typer.Argument(..., help="foo bar")):
"""Foo bar"""
return await sleep(5)
if __name__=="__main__":
app()
This is written in such a way it could be literally written as a PR and put as a method into typer.main.Typer.
Thoughts?
from typer.
@neimad1985 Easier, but less fancy than the decorator solution is to just nest your async func:
from asyncio import run as aiorun
import typer
def main(name: str = typer.Argument("Wade Wilson")):
async def _main():
typer.echo(f"Hello {name}")
aiorun(_main())
if __name__ == "__main__":
typer.run(main)
from typer.
Noted but I don't think I've been impolite in this thread... And I believe I've added a relevant feature (the ability to wrap async functions as cli commands). Correct me if I'm wrong.
from typer.
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)
from typer.
i'm not able to be enthusiast about your attitude in this PR, this is where my emoji stem from. Given the irrelevant value your little boilerplate adds upstream this will neither add nor remove value to this library in my (irrelevant) opinion.
from typer.
Any news on the async support? Any way we can help?
from typer.
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()
from typer.
Just found pallets/click#85 (comment) which may help me
from typer.
We all need this
from typer.
@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}")
from typer.
The workaround here are very clever! In my codebases I'm trying to use type annotations wherever possible, so I annotated the above to the point where I think everything works fine. If there are any type annotation aficionados here, feel free to rip it apart:
import asyncio
from collections.abc import Callable, Coroutine
from functools import wraps
from typing import Any, ParamSpec, TypeVar
import typer
P = ParamSpec("P")
R = TypeVar("R")
class AsyncTyper(typer.Typer):
"""Asyncronous Typer that derives from Typer.
Use this when you have an asynchronous command you want to build, otherwise, just use Typer.
"""
def async_command( # type: ignore # Because we're being generic in this decorator, 'Any' is fine for the args.
self,
*args: Any,
**kwargs: Any,
) -> Callable[
[Callable[P, Coroutine[Any, Any, R]]],
Callable[P, Coroutine[Any, Any, R]],
]:
"""An async decorator for Typer commands that are asynchronous."""
def decorator( # type: ignore # Because we're being generic in this decorator, 'Any' is fine for the args.
async_func: Callable[P, Coroutine[Any, Any, R]],
) -> Callable[P, Coroutine[Any, Any, R]]:
@wraps(async_func)
def sync_func(*_args: P.args, **_kwargs: P.kwargs) -> R:
return asyncio.run(async_func(*_args, **_kwargs))
# Now use app.command as normal to register the synchronous function
self.command(*args, **kwargs)(sync_func)
# We return the async function unmodified, so its library functionality is preserved.
return async_func
return decorator
from typer.
It'd be nice if we could use Typer with the async-click fork.
from typer.
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 import click
with import asyncclick as click
works like a charm but it means maintenance of the fork yourself. If @smurfix could keep his async fork maintained and up to date with click upstream, and if typer was based on asyncclick then we would really have something great here.
from typer.
Thanks @smurfix. Once you have asyncclick updated, if @tiangolo doesn't have a nice async typer by then, perhaps ill make a good typer-async
fork of typer and use asyncclick and add to pypi for us all to use.
from typer.
Also interested in a solution to this.
from typer.
I think the solutions above by @macintacos @rafalkrupinski @gpkc are close, but they miss the magic of typer, which is no instructions needed. using inspect.iscoroutinefunction() (in python since 3.5) we can just do the code automatically so users can just do @app.command() as normal.
This implementation also allows to switch event loop (example show for uvloop) using the asyncio.Runner available in python >= 3.11.
I think something like this would be good to have in typer as it fits the simple design of typer.
#!/usr/bin/python
import typer
import asyncio
import inspect
import uvloop
import sys
from functools import wraps
class UTyper(typer.Typer):
def __init__(self, *args, loop_factory=None, **kwargs):
super().__init__(*args, **kwargs)
self.loop_factory = loop_factory
def command(self, *args, **kwargs):
decorator = super().command(*args, **kwargs)
def add_runner(f):
@wraps(f)
def runner(*args, **kwargs):
if sys.version_info >= (3, 11) and self.loop_factory:
with asyncio.Runner(loop_factory=self.loop_factory) as runner:
runner.run(f(*args,**kwargs))
else:
asyncio.run(f(*args,**kwargs))
if inspect.iscoroutinefunction(f):
return decorator(runner)
return decorator(f)
return add_runner
app = UTyper(loop_factory=uvloop.new_event_loop)
@app.command()
def hello(name: str):
print(f"Hello {name}")
@app.command()
async def goodbye(name: str, formal: bool = False):
if formal:
print(f"Goodbye Ms. {name}. Have a good day.")
else:
print(f"Bye {name}!")
await asyncio.sleep(1)
if __name__ == "__main__":
app()
prints:
$ ./test.py goodbye asdf --formal
Goodbye Ms. asdf. Have a good day.
$ ./test.py hello asdf
Hello asdf
and the 'goodbye' test pauses for a second, as expected.
from typer.
@cauebs I suppose jumping to a PR is a bit of a jump, I wasn't really aware the project was big enough to have a lot of maintainers or QA. We are planning on using the code I just wrote on a rather big company cli, so the feature is required for us, and is complete for our purposes. I just demo'd how to add basic anyio support, and I'll help out as best I can. I'm sure the PR can provide good scaffolding for a future solution.
from typer.
You could possibly implement Tiangolo's new little reop Asyncer, it has Aynio support for making functions async. It is very similar to @ryanpeach's implementation of anyio :)
https://github.com/tiangolo/asyncer
from typer.
What is the final solution for this? I'm a bit confused.
from typer.
is this project deprecated? There is no answer to the merge request for 1 year.
from typer.
@borissmidt good find. yes, the entire loop factory/runner thing is not necessary. i've posted a concise version below.
#!/usr/bin/python
import typer
import asyncio
import inspect
import uvloop
from functools import wraps
class UTyper(typer.Typer):
def command(self, *args, **kwargs):
decorator = super().command(*args, **kwargs)
def add_runner(f):
@wraps(f)
def runner(*args, **kwargs):
asyncio.run(f(*args,**kwargs))
if inspect.iscoroutinefunction(f):
return decorator(runner)
return decorator(f)
return add_runner
app = UTyper()
app2 = UTyper()
@app2.command()
async def hello(name: str):
print(f"Hello to subtest {name}")
print(f"{asyncio.get_running_loop()=}")
app.add_typer(app2, name="sub")
@app.command()
def hello(name: str):
print(f"Hello {name}")
@app.command()
async def goodbye(name: str, formal: bool = False):
if formal:
print(f"Goodbye Ms. {name}. Have a good day.")
else:
print(f"Bye {name}!")
await asyncio.sleep(1)
print(f"{asyncio.get_running_loop()=}")
if __name__ == "__main__":
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
app()
from typer.
The UTyper
workaround above works well, but when used as a decorator it converts your coroutine functions into sync functions. This breaks calling a command from another command. I also think it is simply more appropriate to leave the function functionally unchanged, as typer already does for sync functions. The below adjustment should fix this, and will work even when calling the outer command:
class UTyper(typer.Typer):
# https://github.com/tiangolo/typer/issues/88
@override
def command(self, *args, **kwargs):
decorator = super().command(*args, **kwargs)
def add_runner(f):
if inspect.iscoroutinefunction(f):
@wraps(f)
def runner(*args, **kwargs):
return asyncio.run(f(*args,**kwargs))
decorator(runner)
else:
decorator(f)
return f
return add_runner
app = UTyper()
@app.command()
async def inner():
await asyncio.sleep(1)
print("inner")
@app.command()
async def outer():
await inner()
if __name__ == "__main__":
app()
from typer.
I am working on a cli part of an async app with async DB connection, so needed the async cli as well, thanks for all contributors to the AsyncTyper / UTyper workaround, here is my working version:
import asyncio
import inspect
from functools import wraps, partial
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 asyncio.run(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)
I needed the callback to be async as well to handle database connection init instead of repeating it
from typer.
pr #340 test ok. it is better.
from https://github.com/skeletorXVI/typer/tree/feature/anyio-support
not merge yet
import asyncio
import typer
app = typer.Typer()
@app.command()
async def wait(seconds: int):
await asyncio.sleep(seconds)
typer.echo(f"Waited for {seconds} seconds")
if __name__ == "__main__":
app()
from typer.
class AsyncTyper(typer.Typer):
def async_command(self, _func: Callable = None, *args, **kwargs):
def decorator(async_func):
@wraps(async_func)
def sync_func(*_args, **_kwargs):
return run(async_func(*_args, **_kwargs))
self.command(*args, **kwargs)(sync_func)
return async_func
if _func:
return decorator(_func)
return decorator
Then it also works without parenthesis
@app.async_command
async def my_async_command():
pass
edit: I just realised that original typer doesn't work without them :D
from typer.
Calling asyncio.run in the first line of the command does the job.
Of course you actually need to pass another function to asyncio.run, so for every command you need two functions, one of which is boilerplate.
So maybe you write a decorator for it... but it's such a common case, maybe someone wrote it already...
Wait, why it isn't a part of the library?
from typer.
I can't see any appeal for having it at the moment other than it being extra code to be maintained to be honest. What's the problem with just calling aiorun
?
from typer.
I can't see any appeal for having it at the moment other than it being extra code to be maintained to be honest. What's the problem with just calling
aiorun
?
Either change the project motto to "Another random CLI lib that works a little bit like FastAPI" or introduce async and keep the "little brother of FastAPI" headline.
FastAPI without async? Huh?
Jokes aside, is this really such a large maintenance effort?
from typer.
I will update asyncclick to the latest click
release as soon as anyio 2.0 is ready.
from typer.
@mreschke I've updated @jessekrubin's PR to remove the conflicts with master, in case you find it useful.
from typer.
Okay, we're all trying to help here. Let's not take anything personally.
@ryanpeach Your solution is in essence very similar to mine, but yours is one step ahead. One thing you missed and that I will insist on is that we should tackle not only commands but also things such as autocompletion functions (and others I might be missing).
And another matter we should discuss before jumping to a PR (and here I kind of understand the discomfort you might have caused to @aogier) is supporting different async runtimes other than asyncio
. I couldn't use the feature as it stands in your code, because I use trio
instead of asyncio
.
My proposal: add an "extra" on the package called anyio
, alt-async
or something else, that toggles an optional dependency on anyio=^3
. Then, in this decorators impl, we check if anyio
is available, and if it's not we fallback into asyncio
(your current impl). Otherwise it's just a matter of time until someone opens another issue here requesting support for trio
or something else.
On a final note, I usually wait for a maintainer to weigh in, so as not to waste any time on an unwelcome solution. I salute your initiative, but give people time to evaluate your proposal! π Cheers.
from typer.
I don't think it makes sense for typer to have an
async_command
built in.
Callingasyncio.run
in the first line of the command does the job.Can you please point to any examples?
Just call asyncio.run
inside the command
if you want to do it like that.
from asyncio import run as aiorun
@app.command()
def my_command(email: str) -> None:
async def do_my_command():
from <some_module> import awaitable
await awaitable()
aiorun(do_my_command())
from typer.
from typer.
Nobody would be complaining about missing async support if was just for compatibility (similarity actually) with FastAPI. There's a increasingly common case of calling async functions, IMO the main example would be remote requests with httpx.
I must admit I'm really surprised with the resistance over the 15 lines of code this change represents.
from typer.
I ran into a related problem when running async unit tests with pytest-asyncio because by the time the test calls runner.invoke(...)
there is already a running event loop. I'm not aware of a straightforward way to work around this issue, but I managed to find a slightly convoluted one, by effectively doing the same async-to-sync trick with the test π€
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 @embed_event_loop
is the "async-to-sync" decorator described in previous comments on this issue.
from typer.
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.
from typer.
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.
from typer.
Hi :)
Anything new on this ?
@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.
from typer.
Tested the code I posted above and it works. You could probably just add it as a method to typer.Typer. I'll make a pr.
from typer.
Unsure why @aogier downvoted. It runs and its integrating well into my repo.
from typer.
The workaround here are very clever! In my codebases I'm trying to use type annotations wherever possible, so I annotated the above to the point where I think everything works fine. If there are any type annotation aficionados here, feel free to rip it apart:
import asyncio from collections.abc import Callable, Coroutine from functools import wraps from typing import Any, ParamSpec, TypeVar import typer P = ParamSpec("P") R = TypeVar("R") class AsyncTyper(typer.Typer): """Asyncronous Typer that derives from Typer. Use this when you have an asynchronous command you want to build, otherwise, just use Typer. """ def async_command( # type: ignore # Because we're being generic in this decorator, 'Any' is fine for the args. self, *args: Any, **kwargs: Any, ) -> Callable[ [Callable[P, Coroutine[Any, Any, R]]], Callable[P, Coroutine[Any, Any, R]], ]: """An async decorator for Typer commands that are asynchronous.""" def decorator( # type: ignore # Because we're being generic in this decorator, 'Any' is fine for the args. async_func: Callable[P, Coroutine[Any, Any, R]], ) -> Callable[P, Coroutine[Any, Any, R]]: @wraps(async_func) def sync_func(*_args: P.args, **_kwargs: P.kwargs) -> R: return asyncio.run(async_func(*_args, **_kwargs)) # Now use app.command as normal to register the synchronous function self.command(*args, **kwargs)(sync_func) # We return the async function unmodified, so its library functionality is preserved. return async_func return decorator
Would you have any examples to share?
from typer.
I agree, in the pr i made i only wrap the callback registered to typer. Not the function itself. I'll see if there was any progress on getting it merged.
I'm just stuck with the pr because the mypy configurations is to strict and I can't make it happy for all python versions π
from typer.
from typer.
any news on this?
from typer.
Thanks @smurfix. Once you have asyncclick updated, if @tiangolo doesn't have a nice async typer by then, perhaps ill make a good
typer-async
fork of typer and use asyncclick and add to pypi for us all to use.
@mreschke I made a pull request to this repo that gets you most of the way to async. #128
from typer.
Hi :)
Anything new on this ?
from typer.
Thanks for the quick answer @jessekrubin
Would it be possible that you share a simple code example on how you do this please ?
from typer.
Ok thanks, that's exactly what I was thinking.
The problem is the duplication of functions main and _main and their arguments.
If you have multiple subcommands, which I have, for your program you have more and more duplication.
Anyway thanks for answering me.
from typer.
@cauebs
Thanks, that is actually a nice idea, I will try it !
from typer.
@jessekrubin
Very nice trick. I should have thought about this. Thank you.
from typer.
Just found pallets/click#85 (comment) which may help me
As that issue was closed a few years ago and is now locked I decided to open a new one containing a bit more information and addressing some comments in the previous issue. You can look at it yourself, leave some feedback (preferably typer agnostic) and upvote it to show interest in this feature: pallets/click#2033
from typer.
I think just adding those decorators to the library and having @app.command() auto detect if the function it's decorating is async or not and just pick the appropriate decoration. Not hard at all to implement. Thanks everyone for the suggestions
from typer.
Typer mentions being able to stack the FastAPI and Typer decorators on top of each other because they're "meant to work together". Without async support this actually isn't possible though. I hope this PR will be accepted to provide a path towards a solution, even if imperfect, since the stated design goals of Typer are to work with FastAPI which is async in nature.
from typer.
To the people scrolling down to find a solution. In my opinion, what @neimad1985 did is really the simplest and works fine.
@neimad1985 Easier, but less fancy than the decorator solution is to just nest your async func:
from asyncio import run as aiorun import typer def main(name: str = typer.Argument("Wade Wilson")): async def _main(): typer.echo(f"Hello {name}") aiorun(_main()) if __name__ == "__main__": typer.run(main)
If you don't want to nest functions you can also just:
import asyncio
import typer
def main(arg_1: str):
asyncio.run(my_async_func(arg_1))
async def my_async_func(arg_1: str):
await asyncio.sleep(5)
if __name__ == "__main__":
typer.run(main)
from typer.
@gpkc This is f*cking genius!
@tiangolo What do you think?
from typer.
I don't think it makes sense for typer to have an async_command
built in.
Calling asyncio.run
in the first line of the command does the job.
from typer.
I don't think it makes sense for typer to have an
async_command
built in.Calling
asyncio.run
in the first line of the command does the job.
Can you please point to any examples?
from typer.
This looks good, maybe this should transform into a PR?
from typer.
Looks like someone is working on a PR, maybe.
One issue I'll leave here for @borissmidt is that if one were to use
subapp=Typer(loop_factory=...)
app=Typer(loop_factory=...)
app.add_typer(subapp,"mysub"))
then it would be nice that the "subapp" app would use the loop_factory from the root Typer() object. I hadn't written that up yet, and currently it doesn't do that in my child class or your native implementation AFAICT
from typer.
from typer.
Here's my attempt at some asyncio background: there is only one call to run
during an event loop lifecycle. you can create tasks while an event loop is running, but if there isn't an event loop you need to start it up, and the run interface is the only game in town. The whole get_running_loop()
interfaces is when you know there is a running loop and you want to do something with it. That's not applicable for us (well it is in the fact that I use it to demonstrate different event loops in use).
Of course you can spin down the event loop and start up another one, but most of the time it's just one event loop per run.
What typer does is route to a method, and we're putting in a asyncio.run (well specifically the 3.11 version of that interface) into the decorator. Since typer will never call two methods, then we're only starting one event loop. Getting the loop factory option shared between Typer
instances linked via Typer.add_typer is probably a more obscure use case, because it's not clear how many people replace the event loop, and certainly how many people chain together Typer
instances. But imagine if you have Typer instances started in a dozen modules then you have to go to each one to set the loop_factory, that then the one place where it all joins together.
what I'm saying that it would be nice if you don't have to set the loop factory for every typer instance, just the root. This example below (that needs my UTyper class above, as well as uvloop package installed) shows that the loop_factory in use has to be set for each typer instance, and I think that's sub optimal, and unexpected behavior.
Since this the add_typer call is after all the decorations done, such a solution will require some minor surgery that occurs at the time of the add_typer call. Hopefully not too bad. But if this work is too much, I think we can get it into a second PR after this one goes it (which is what I hope you're looking to do soon). It might be better to defer this, as it will certainly complicate the implementation.
Demonstrating that the loop factor needs to be set to each Typer instance instead of just the root.
$ python test-utyper.py goodbye foo
Bye foo!
asyncio.get_running_loop()=<uvloop.Loop running=True closed=False debug=False>
$ python test-utyper.py sub hello foo
Hello to subtest foo
asyncio.get_running_loop()=<_UnixSelectorEventLoop running=True closed=False debug=False>
app = UTyper(loop_factory=uvloop.new_event_loop)
app2 = UTyper()
@app2.command()
async def hello(name: str):
print(f"Hello to subtest {name}")
print(f"{asyncio.get_running_loop()=}")
app.add_typer(app2, name="sub")
@app.command()
def hello(name: str):
print(f"Hello {name}")
@app.command()
async def goodbye(name: str, formal: bool = False):
if formal:
print(f"Goodbye Ms. {name}. Have a good day.")
else:
print(f"Bye {name}!")
await asyncio.sleep(1)
print(f"{asyncio.get_running_loop()=}")
if __name__ == "__main__":
app()
from typer.
Is there any good reason that async commands are not part of the library? Or is it just a missing feature?
from typer.
from typer.
I can't see any appeal for having it at the moment other than it being extra code to be maintained to be honest. What's the problem with just calling
aiorun
?
Why not use argparse
and not have extra dependencies? It's the same kind of argument. Ironically with how argparse
there's less work to do this, because you don't have to place asyncio.run()
for every subcommand. Given how typer
operates and interest in this ticket I believe this is necessary.
from typer.
Of course none of those are fair comparisons. It's still FastAPI's sibling because it follows the same design philosophy with types and clarity. Not to mention that async support is much more needed on a web framework, which allows it to handle requests concurrently. For a CLI, it's not the case that you really need concurrency in most scenarios.
So, in the end, the appeal to support async on Typer is purely from a compatibility perspective. For example, you might have a FastAPI project and your code will have many async parts, it can be handy to be able to call those directly from your Typer CLI.
In any case, it was just a devil's advocate positioning, no need to get emotional over it :) I'm assuming these are probably the reasons why the support for async hasn't been done yet, and of course if it's actually denied everyone can just fork the project and add it too, then use as a dependency, so you don't have to repeat anything.
from typer.
I didn't mean compatibility with FastAPI. I meant compatibility with async projects' codebase in general.
If you're making multiple remote requests with HTTPX, you could organize your code to, for example, chain aiorun
with asyncio.gather
. If you're making a single request, there's no advantage on using asyncio as httpx supports synchronous calls.
Again, this is just some basic questioning which I assume the package maintainers will think about and are probably the reason why thins hasn't been fixed yet.
from typer.
Of course none of those are fair comparisons. It's still FastAPI's sibling because it follows the same design philosophy with types and clarity. Not to mention that async support is much more needed on a web framework, which allows it to handle requests concurrently. For a CLI, it's not the case that you really need concurrency in most scenarios.
Just because you use async only for web doesn't mean everyone else is. There are many async-only libraries that have little to do with web.
from typer.
Nobody would be complaining about missing async support if was just for compatibility (similarity actually) with FastAPI. There's a increasingly common case of calling async functions, IMO the main example would be remote requests with httpx. I must admit I'm really surprised with the resistance over the 15 lines of code this change represents.
@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.
from typer.
Just because you use async only for web doesn't mean everyone else is. There are many async-only libraries that have little to do with web.
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.
from typer.
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.
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.
from typer.
@borissmidt's PR is still open. Not sure about 'hostility'
from typer.
Any news on the async support? Any way we can help?
with syncer:
- β try this lib: https://github.com/miyakogi/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()
from typer.
Related Issues (20)
- Documentation is misleading. new `typer.run` behavior HOT 2
- See if rich 13.x is compatible HOT 4
- Using `some_type | None` syntax for type annotations causes error in python 3.11 HOT 15
- unlimited argument for an option with comma spliter HOT 2
- TAB completion is giving local directory files where command is called. HOT 7
- Support for bytes in Options and Arguments HOT 3
- Is it possible to include a Prolog in `--help` HOT 2
- Get the typer output with html format to provide it to termynal HOT 5
- how to use typer on class method __init__ with self argument, got this error: Error: Missing argument 'SELF'. HOT 2
- Support for localization of messages HOT 1
- Auto-completion when application works in 2 modes (GUI, CLI) HOT 1
- DOC: Documentation of passing multiple values in "option" vs "argument" isn't sufficiently explicit HOT 2
- SIGINT from docker is ignored HOT 4
- π Roadmap HOT 3
- Source distribution of 0.11.0 is missing the `docs_src` folder HOT 1
- π Upgrading from `typer<0.12.0` to `typer==0.12.0` breaks the install by partially removing the package/module files HOT 4
- Rich markup in generated Markdown docs HOT 1
- Add CI configs to run tests on Windows and MacOS
- Rich markup in Zsh completion help lines HOT 1
- 22
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
π Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. πππ
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google β€οΈ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from typer.