Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Use asyncio.run instead of get_event_loop() #155

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Use asyncio.run instead of get_event_loop() #155

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
18 changes: 15 additions & 3 deletions src/tests/test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
import unittest
import decimal
import inspect
from asyncio import get_event_loop
from collections import defaultdict, ChainMap, abc as c
from decorator import dispatch_on, contextmanager, decorator
try:
Expand All @@ -28,9 +27,22 @@ async def before_after(coro, *args, **kwargs):
return "<before>" + (await coro(*args, **kwargs)) + "<after>"


def asyncio_run(awaitable):
"""asyncio.run for Python 3.5"""
try:
from asyncio import run
aio_run = run
micheles marked this conversation as resolved.
Show resolved Hide resolved
except ImportError:
# Python 3.5
from asyncio import get_event_loop
return get_event_loop().run_until_complete(awaitable)
else:
return aio_run(awaitable)
Comment on lines +32 to +40
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess the aio_run = run was meant to define the runner on both try / except branches, right?
If you're not going to do it anymore, there's no need for the aio_run alias, let's drop it.



@decorator
def coro_to_func(coro, *args, **kw):
return get_event_loop().run_until_complete(coro(*args, **kw))
return asyncio_run(coro(*args, **kw))


class CoroutineTestCase(unittest.TestCase):
Expand All @@ -39,7 +51,7 @@ def test_before_after(self):
async def coro(x):
return x
self.assertTrue(inspect.iscoroutinefunction(coro))
out = get_event_loop().run_until_complete(coro('x'))
out = asyncio_run(coro('x'))
self.assertEqual(out, '<before>x<after>')

def test_coro_to_func(self):
Expand Down