Use functools.partial; it’s a standard way to do such things, and it’s specifically recommended in the docs for loop.run_in_executor, as well as more generally in the Event Loop docs. Here’s how it might look for you: import functools # at the top with the other imports loop.run_in_executor(None, functools.partial(update_contacts, data={ ’email’: email, ‘access_token’: g.tokens[‘access_token’] })) You … Read more
Taking a guess at this, but I think you’re using the event loop wrong for asyncio. I don’t know enough about gevent to understand what type of object it’s creating to know if this will work. The documentation would indicate that the order in which stop() and run_forever() are called is relevant to the operational … Read more
May be you can find better way to do it, but, for example, you can just move your wrapping logic to some context manager to prevent code duplication: import asyncio import functools import time from contextlib import contextmanager def duration(func): @contextmanager def wrapping_logic(): start_ts = time.time() yield dur = time.time() – start_ts print(‘{} took {:.2} … Read more
I think you may need to make your add_task method aware of whether or not its being called from a thread other than the event loop’s. That way, if it’s being called from the same thread, you can just call asyncio.async directly, otherwise, it can do some extra work to pass the task from the … Read more
Syntax for python >=3.5 loop = asyncio.get_event_loop() for signame in (‘SIGINT’, ‘SIGTERM’): loop.add_signal_handler(getattr(signal, signame), lambda: asyncio.ensure_future(ask_exit(signame)))
Your code will run on Python3.10 but as of 3.11 it will be an error to call asyncio.get_event_loop when there is no running loop in the current thread. Since you need loop as an argument to amain, apparently, you must explicitly create and set it. It is better to launch your main task with asyncio.run … Read more
Just don’t use the result of session.get() as a context manager; use it as a coroutine directly instead. The request context manager that session.get() produces would normally release the request on exit, but so does using response.text(), so you could ignore that here: @asyncio.coroutine def fetch(session, url): with aiohttp.Timeout(10): response = yield from session.get(url) return … Read more
In this case, there is no way to cancel the Future once it has actually started running, because you’re relying on the behavior of concurrent.futures.Future, and its docs state the following: cancel() Attempt to cancel the call. If the call is currently being executed and cannot be cancelled then the method will return False, otherwise … Read more
There is workaround for Windows. Run another corouting which wake up loop every second and allow loop to react on keyboard interrupt Example with Echo server from asyncio doc async def wakeup(): while True: await asyncio.sleep(1) loop = asyncio.get_event_loop() coro = loop.create_server(EchoServerClientProtocol, ‘127.0.0.1’, 8888) server = loop.run_until_complete(coro) # add wakeup HACK loop.create_task(wakeup()) try: loop.run_forever() except … Read more
This works for me, please try: import asyncio import pytest pytest_plugins = (‘pytest_asyncio’,) @pytest.mark.asyncio async def test_simple(): await asyncio.sleep(0.5) Output of pytest -v confirms it passes: collected 1 item test_async.py::test_simple PASSED And I have installed: pytest 6.2.5 pytest-asyncio 0.16.0 # anyio not installed