A "run" function for asyncio-based apps that does all the boilerplate.
License: Apache License 2.0
In past roles I needed to examine server logs extensively, and that experience was good inspiration for a bunch of ideas around making logs a little easier to produce and work with.
![dependabot-preview[bot] avatar](https://avatars.githubusercontent.com/in/2141?v=4 "dependabot-preview[bot]")
Would you be open to reducing the logging level for some components of the aiorun.run function from critical to info? In particular, it would be great if the "default" mode of an asyncio application, in which the main function run via aiorun.run(main()) cleanly exits via a call to loop.stop(), logged at the info level left the higher logging levels for (a) logging signal handlers and (b) logging if pending tasks are canceled.
I'd be happy to open a PR if this seems acceptable.
Verify that clean shutdown still happens if some tasks contain async generators anywhere in the execution context. (It should all work, just add a test case that verifies it.)
Hi! Thank you very much for this library, its very elegant, and to the point.
I have a suggestion regarding the executor_workers argument to run():
The ThreadPoolExecutor specifies the max_workers=None argument (same as max_workers: Optional[int] = None) because it auto-calculates the threadpool size to be os.cpu_count() * 5 (so on my PC, 16 * 5 = 80, a far cry from aiorun's default 10).
aiorun however specifies this arg as executor_workers: int = 10 which means I'll get type errors passing None to take advantage of ThreadPoolExecutor's default.
I could simply do this calculation myself, but it seems like it'd be useful for aiorun to mirror its default executor. What do you think?
If to try to create something(example: connection to rabbitmq) in a main function and it something compleat with an exception then would be nice to exit from an app. 袝his behaviour can be achieved through the use of run_until_complete method instead of run_forever. Propose to replace method.
Hi, I found the case when I want to cancel main starting coroutine (and await it's cancellation) before the cancellation of all other pending loop's tasks.
Perhaps, it's not the issue of aiorun package directly, but let me ask advice how to deal with such case (if it can be solved using aiorun).
The case can be reproduced by using code below:
import argparse
import asyncio
import logging
import signal
from aiorun import run
class SomeThirdPartyRunner:
def __init__(self):
self._queue = list(range(5))
# actually, I can't use `shutdown_waits_for` below because it's 3'rd party package
self._work_task = asyncio.create_task(self._work())
self._work_task_done = asyncio.Future()
async def _work(self):
try:
while self._queue:
logging.info(f'processing queue={self._queue}')
self._queue.pop()
await asyncio.sleep(1)
except asyncio.CancelledError:
logging.info(' *** we are here only when aiorun is used *** ')
finally:
self._work_task_done.set_result(None)
async def wait_done(self):
await self._work_task_done
async def corofn():
runner = SomeThirdPartyRunner()
try:
await asyncio.sleep(2)
except asyncio.CancelledError:
pass
finally:
logging.info('stopping runner...')
await runner.wait_done()
logging.info('runner stopped')
def run_through_aiorun():
run(_aiorun_main())
async def _aiorun_main():
await corofn()
def run_through_asyncio():
asyncio.run(_asyncio_main())
async def _asyncio_main():
loop = asyncio.get_event_loop()
task = loop.create_task(corofn())
task.add_done_callback(lambda _: loop.stop())
loop.add_signal_handler(signal.SIGINT, task.cancel)
loop.add_signal_handler(signal.SIGTERM, task.cancel)
await task
if __name__ == '__main__':
logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s - [%(levelname)s] - [%(name)s] "
"- %(filename)s:%(lineno)d - %(message)s",
)
parser = argparse.ArgumentParser()
parser.add_argument('--aiorun',
action='store_true',
help='Run script using aiorun package')
args = parser.parse_args()
if args.aiorun:
logging.info('Running using aiorun')
run_through_aiorun()
else:
logging.info('Running using asyncio')
run_through_asyncio()Starting script using aiorun and press Ctrl+C on 2'nd message processing queue:
(env_3_8) MacBook-Pro-2:~ fedir$ python script.py --aiorun
[INFO] - [root] - script.py:83 - Running using aiorun
[DEBUG] - [aiorun] - aiorun.py:155 - Entering run()
[DEBUG] - [asyncio] - selector_events.py:59 - Using selector: KqueueSelector
[DEBUG] - [aiorun] - aiorun.py:236 - Creating default executor
[INFO] - [root] - script.py:18 - processing queue=[0, 1, 2, 3, 4]
[INFO] - [root] - script.py:18 - processing queue=[0, 1, 2, 3]
^C[DEBUG] - [aiorun] - aiorun.py:304 - Entering shutdown handler
[CRITICAL] - [aiorun] - aiorun.py:317 - Stopping the loop
[INFO] - [aiorun] - aiorun.py:249 - Entering shutdown phase.
[INFO] - [aiorun] - aiorun.py:262 - Cancelling pending tasks.
[DEBUG] - [aiorun] - aiorun.py:264 - Cancelling task: <Task pending name='Task-1' coro=<run.<locals>.new_coro() running at /Users/fedir/env/env_3_8/lib/python3.8/site-packages/aiorun.py:206> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x10a00fdf0>()]>>
[DEBUG] - [aiorun] - aiorun.py:264 - Cancelling task: <Task pending name='Task-2' coro=<SomeThirdPartyRunner._work() running at script.py:20> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x10a00fe50>()]>>
[INFO] - [aiorun] - aiorun.py:276 - Running pending tasks till complete
[INFO] - [root] - script.py:40 - stopping runner...
[INFO] - [root] - script.py:22 - *** we are here only when aiorun is used ***
[INFO] - [root] - script.py:42 - runner stopped
[INFO] - [aiorun] - aiorun.py:281 - Waiting for executor shutdown.
[INFO] - [aiorun] - aiorun.py:286 - Shutting down async generators
[INFO] - [aiorun] - aiorun.py:288 - Closing the loop.
[INFO] - [aiorun] - aiorun.py:290 - Leaving. Bye!
Desired result for me in this case is do not cancel corofn() along with SomeThirdPartyRunner._work().
Starting script using asyncio and press Ctrl+C on the same log msg:
(env_3_8) MacBook-Pro-2:~ fedir$ python script.py
[INFO] - [root] - script.py:86 - Running using asyncio
[DEBUG] - [asyncio] - selector_events.py:59 - Using selector: KqueueSelector
[INFO] - [root] - script.py:18 - processing queue=[0, 1, 2, 3, 4]
[INFO] - [root] - script.py:18 - processing queue=[0, 1, 2, 3]
^C[INFO] - [root] - script.py:40 - stopping runner...
[INFO] - [root] - script.py:18 - processing queue=[0, 1, 2]
[INFO] - [root] - script.py:18 - processing queue=[0, 1]
[INFO] - [root] - script.py:18 - processing queue=[0]
[INFO] - [root] - script.py:42 - runner stopped
I've done some updates into aiorun to receive expected result:
https://github.com/cjrh/aiorun/blob/master/aiorun.py#L210
_origin_coro_task = loop.create_task(new_coro())and then on the start of "Entering the shutdown phase":
https://github.com/cjrh/aiorun/blob/master/aiorun.py#L249
if _origin_coro_task is not None:
logger.debug("Cancelling origin coro task: %s", _origin_coro_task)
_origin_coro_task.cancel()
loop.run_until_complete(_origin_coro_task)And received expected result below:
(env_3_8) MacBook-Pro-2:~ fedir$ python script.py --aiorun
[INFO] - [root] - script.py:83 - Running using aiorun
[DEBUG] - [aiorun] - aiorun.py:155 - Entering run()
[DEBUG] - [asyncio] - selector_events.py:59 - Using selector: KqueueSelector
[DEBUG] - [aiorun] - aiorun.py:237 - Creating default executor
[INFO] - [root] - script.py:18 - processing queue=[0, 1, 2, 3, 4]
[INFO] - [root] - script.py:18 - processing queue=[0, 1, 2, 3]
^C[DEBUG] - [aiorun] - aiorun.py:309 - Entering shutdown handler
[CRITICAL] - [aiorun] - aiorun.py:322 - Stopping the loop
[INFO] - [aiorun] - aiorun.py:250 - Entering shutdown phase.
[DEBUG] - [aiorun] - aiorun.py:252 - Cancelling origin coro task: <Task pending name='Task-1' coro=<run.<locals>.new_coro() running at /Users/fedir/env/env_3_8/lib/python3.8/site-packages/aiorun.py:207> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x105197d90>()]>>
[INFO] - [root] - script.py:40 - stopping runner...
[INFO] - [root] - script.py:18 - processing queue=[0, 1, 2]
[INFO] - [root] - script.py:18 - processing queue=[0, 1]
[INFO] - [root] - script.py:18 - processing queue=[0]
[INFO] - [root] - script.py:42 - runner stopped
[INFO] - [aiorun] - aiorun.py:267 - Cancelling pending tasks.
[INFO] - [aiorun] - aiorun.py:281 - Running pending tasks till complete
[INFO] - [aiorun] - aiorun.py:286 - Waiting for executor shutdown.
[INFO] - [aiorun] - aiorun.py:291 - Shutting down async generators
[INFO] - [aiorun] - aiorun.py:293 - Closing the loop.
[INFO] - [aiorun] - aiorun.py:295 - Leaving. Bye!
It would be great to receive expected result using the current aiorun functionality. Thx in advance for feedback.
(env versions)
(env_3_8) MacBook-Pro-2:~ fedir$ python -V
Python 3.8.1
(env_3_8) MacBook-Pro-2:~ fedir$ pip freeze | grep aiorun
aiorun==2020.1.3
P.S. I can provide more practical example of such behaviour if needed (using the combination of aiorun and aiokafka packages).
Hi! Thank you for your library 馃槂
I have a question about this line of code:
loop.create_task(new_coro())I am confused that the link to the result is not saved here.
Because there is a warning about this in the documentation.
Thank you in advance for your response.
If a loop parameter is provided to run(), don't do any of the cleanup steps. In this scenario, the caller is indicating that they need to do custom things with the loop, and the loop is created elsewhere. Therefore it should be cleaned up elsewhere as well. Also we should not shutdown the exectutor, nor cancel the currently pending tasks. Perhaps all we do is return after run_forever() unblocks.
Hey, so I have this issue where, for some reason when I stop the process with ctrl + c (when debugging in vs:code) or (sometimes) restart the service with systemctl the shutdown function isnt triggered at all.
When I stop the service with systemctl stop or using the UI button (break button) (when in vs:code) it works.
Logs dont show any errors, or anything.
Any idea?
Hi, great library, thanks a lot...
Documentation is good too! My issue is following:
I do want to start two separate rest api services via corutines in async way. I have no issues setting up a handler to catch exceptions in event_loop.
however I want to be able programmatically to stop the loop.
anything after run(loop=loop) is not going to work!
what is the way to actually stop the loop and consequently stop two rest api verticals....
Regards,
Ubuntu 16.04 by default ships with Python 3.5.1 that has a typing library that does not provide Coroutine typing definition. This was fixed in Python 3.5.2 (python/typing#119) but afaik Ubuntu does not ship the fixed version.
Running the first example in the readme gives this output:
Traceback (most recent call last):
File "./main.py", line 15, in <module>
from aiorun import run
File "/tmp/scratch/venv/lib/python3.5/site-packages/aiorun.py", line 14, in <module>
from typing import Optional, Coroutine, Callable
ImportError: cannot import name 'Coroutine'
Im currently using this ugly hack and typing library from pip but a better solution should be figgured out.
# Ubuntu 16.04 ships with broken typing lib
# change sys path to use typing lib in virtualenv
if sys.version_info[:3] < (3, 5, 2):
old_syspath = sys.path
sys.path.insert(0, sys.path[-1])
from aiorun import run
sys.path = old_syspath
else:
from aiorun import run
New version of aiorun breaks aiologfields, in that no log fields declared by aiologfields log anymore after updating aiorun. Nothing is logging outside of aiologfields or aiorun to say anything is wrong as well
Is there a way to figure out inside a task, if the loop runs under aiorun or pure asyncio?
Dependabot couldn't authenticate with https://pypi.python.org/simple/.
You can provide authentication details in your Dependabot dashboard by clicking into the account menu (in the top right) and selecting 'Config variables'.
if loop is provided, no reason to change default policies.
https://github.com/cjrh/aiorun/blob/master/aiorun.py#L168
is it feature or bug?
what about tiny pr which skips policies in case of provided loop?
Since #43, the event loop will be stopped whenever its exception handler is invoked, whether or not the context contains an exception key. But the exception handler can be invoked by anything with access to the event loop, and at least one third-party library invokes it without passing exception, namely aiohttp:
if self._loop.get_debug():
_warnings.warn(
f"Unclosed response {self!r}", ResourceWarning, source=self
)
context = {"client_response": self, "message": "Unclosed response"}
if self._source_traceback:
context["source_traceback"] = self._source_traceback
self._loop.call_exception_handler(context)Since this is just a warning, it feels like aiorun should not stop the loop when this happens. (Arguably aiohttp is at fault here for invoking the exception handler for an unexceptional event, but imo aiorun should deal with it gracefully anyway.)
If you agree, I can PR a fix for this?
Thank you for your library, it really enhances the user experience when working with asyncio. But in the course of exploitation we came across the fact that sometimes the program does not stop.
The program does not stop in such a situation
import asyncio
import logging
import aiorun
async def coro_with_exception():
await asyncio.sleep(0.1)
raise Exception
async def unstoppable_coro():
# For some reason one of the tasks does not want to stop
try:
while True:
await asyncio.sleep(0.1)
except asyncio.CancelledError:
while True:
await asyncio.sleep(0.1)
async def main():
asyncio.create_task(unstoppable_coro())
asyncio.create_task(coro_with_exception())
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
aiorun.run(
main(),
stop_on_unhandled_errors=True,
)We see the following logs
ERROR:aiorun:Unhandled exception; stopping loop: 'Task exception was never retrieved'
Traceback (most recent call last):
File "C:\git\aio_run_example.py", line 9, in coro_with_exception
raise Exception
Exception
INFO:aiorun:Entering shutdown phase.
INFO:aiorun:Cancelling pending tasks.
INFO:aiorun:Running pending tasks till complete
I propose to introduce a correction.
Replace
async def wait_for_cancelled_tasks():
return await gather(*tasks, *do_not_cancel, return_exceptions=True)To
async def wait_for_cancelled_tasks():
return await asyncio.wait([*tasks, *do_not_cancel], timeout=gracefull_shutdown_timeout)And add argument to run
def run(
coro: "Optional[Coroutine]" = None,
*,
loop: Optional[AbstractEventLoop] = None,
shutdown_handler: Optional[Callable[[AbstractEventLoop], None]] = None,
shutdown_callback: "ShutdownCallback" = None,
executor_workers: Optional[int] = None,
executor: Optional[Executor] = None,
use_uvloop: bool = False,
stop_on_unhandled_errors: bool = False,
gracefull_shutdown_timeout: float = 60
) -> None:
...If a correction is made, the program is guaranteed to stop.
the latest python version 3.7 brings asyncio.run() function,it is a great feature
now we cloud use it easily, example
import asyncio
async def main():
await asyncio.sleep(1.0)
print("hello cjrh!")
asyncio.run(main())And your aiorun is aslso an awesome library.
Lots of useful discussion at python/asyncio#465.
Can also add examples and scenarios from there.
In the readme the following example is given:
import asyncio
from aiorun import run, shutdown_waits_for
async def corofn():
await asyncio.sleep(60)
print('done!')
async def main():
try:
await shutdown_waits_for(corofn())
except asyncio.CancelledError
print('oh noes!')
run(main())
With the explanation:
If you hit CTRL-C before 60 seconds has passed, you will see oh noes! printed immediately, and then after 60 seconds (since start), done! is printed, and thereafter the program exits.
I'm finding that this isn't the case. The program never actually exits on its own.
(Also there's a missing : on the end of the asyncio.CancelledError)
the latest python version 3.7 brings asyncio.run() function,it is a great feature
now we cloud use it easily, example
路路路
import asyncio
async def main():
await asyncio.sleep(1.0)
print("hello cjrh!")
asyncio.run(main())
路路路
And your aiorun is aslso a awesome library.
I'm getting the "If you provide a loop instance, and you've configured a custom exception handler on it..." error, but I have not configured a custom exception handler.
Surprisingly, I'm getting this with and without the stop_on_unhandled_errors arg.
loop = asyncio.get_event_loop()
asyncio.set_event_loop(loop)
run(self.main(), loop=loop) # happens if this
# run(self.main(), loop=loop, stop_on_unhandled_errors=True) # and also thisThe only "funny" or additional thing is that this code is running in AWS Greengrass, but I don't think this should matter.
Thoughts, or additional info I can provide that might elucidate the problem?
Hello,
aiorun currently makes use of python's logging module, yet one call to print remains, would it be possible to have it removed or logged in the next version?
https://github.com/cjrh/aiorun/blob/master/aiorun.py#L200
Thanks very much.
The really massive problem is that Windows doesn't support signals in the same way that POSIX does, but even worse is that the signals API in asyncio doesn't even currently allow signals to be set for Windows.
Don't know what to do here. How do Windows services handle signals and cancellation?
We have several microservices running with aiorun
After update we get the following:
aiorun==2020.12.1 and aiohttp==3.6.2 -> works as before
aiorun==2019.4.1 and aiohttp==3.7.4 -> works as before
aiorun==2020.12.1 and aiohttp==3.7.4 -> outgoing requests do not work.
Unfortunately, I cannot get any reason more specific than this:
Traceback (most recent call last):
File "/home/pavka/Documents/console/slayer2/venv/lib64/python3.9/site-packages/aiohttp/connector.py", line 969, in _wrap_create_connection
return await self._loop.create_connection(*args, **kwargs) # type: ignore # noqa
File "/usr/lib64/python3.9/asyncio/base_events.py", line 1056, in create_connection
raise exceptions[0]
File "/usr/lib64/python3.9/asyncio/base_events.py", line 1041, in create_connection
sock = await self._connect_sock(
File "/usr/lib64/python3.9/asyncio/base_events.py", line 955, in _connect_sock
await self.sock_connect(sock, address)
File "/usr/lib64/python3.9/asyncio/selector_events.py", line 502, in sock_connect
return await fut
File "/usr/lib64/python3.9/asyncio/selector_events.py", line 537, in _sock_connect_cb
raise OSError(err, f'Connect call failed {address}')
ConnectionRefusedError: [Errno 111] Connect call failed ('127.0.0.1', 8500)
In the above instance, it is trying to talk to a local docker container but the same happens when trying to connect to external services:
Traceback (most recent call last):
File "/home/servicelayer/.local/lib/python3.7/site-packages/aiohttp/connector.py", line 969, in _wrap_create_connection
return await self._loop.create_connection(*args, **kwargs) # type: ignore # noqa
File "/usr/local/lib/python3.7/asyncio/base_events.py", line 954, in create_connection
raise exceptions[0]
File "/usr/local/lib/python3.7/asyncio/base_events.py", line 941, in create_connection
await self.sock_connect(sock, address)
File "/usr/local/lib/python3.7/asyncio/selector_events.py", line 464, in sock_connect
return await fut
File "/usr/local/lib/python3.7/asyncio/selector_events.py", line 494, in _sock_connect_cb
raise OSError(err, f'Connect call failed {address}')
ConnectionRefusedError: [Errno 111] Connect call failed ('172.17.11.3', 8500)
Some help will be highly appreciated...
Currently it says
:param loop: Optionally supply your own loop. If missing, the
default loop attached to the current thread context will
be used, i.e., whatever ``asyncio.get_event_loop()`` returns.
Indicating the use of asyncio.get_event_loop(), but
loop_was_supplied = bool(loop)
if not loop_was_supplied:
if use_uvloop:
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
else:
asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)Shows that we use asyncio.new_event_loop()
This will cause certain programs that use Tasks to stop working when plugging in aiorun until they provide the loop that is actually being used, and there is no easily found documentation to understand why a program has stopped working.
My opinion is to actually use asyncio.get_event_loop(), but I admit that I may not completely understand the benefits to using asyncio.new_event_loop() so a warning or indicator may be more appropriate.
An unhandled exception results in "Task exception was never retrieved".
Example:
import asyncio
from aiorun import run
async def main():
# Put your application code here
raise Exception('not consumed')
if __name__ == '__main__':
run(main())
Result:
TTask exception was never retrieved
future: <Task finished coro=<run.<locals>.new_coro() done, defined at C:\Users\Ian\.conda\envs\enerlytics\lib\site-packages\aiorun.py:165> exception=Exception('not consumed',)>
Traceback (most recent call last):
File "C:\Users\Ian\.conda\envs\enerlytics\lib\site-packages\aiorun.py", line 175, in new_coro
await coro
File "test.py", line 6, in main
raise Exception('not consumed')
Exception: not consumed
... and the event loop continues running.
If main() exits with an unhandled exception then aiorun should gracefully stop the event loop and reraise the exception.
The imports don't match the usages!
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
馃枛 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.
An Open Source Machine Learning Framework for Everyone
Django
The Web framework for perfectionists with deadlines.
Laravel
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 馃搳馃搱馃帀
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
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.