Merge pull request piccolo-orm#18 from piccolo-orm/improved_run_sync

handling edge cases in run_sync

Author: dantownsend

piccolo/apps/user/tables.py

@@ -6,11 +6,10 @@
 import secrets
 import typing as t
 
-from asgiref.sync import async_to_sync
-
-from piccolo.table import Table
 from piccolo.columns import Varchar, Boolean, Secret
 from piccolo.columns.readable import Readable
+from piccolo.table import Table
+from piccolo.utils.sync import run_sync
 
 
 class BaseUser(Table, tablename="piccolo_user"):
@@ -50,7 +49,7 @@ def get_readable(cls) -> Readable:
 
     @classmethod
     def update_password_sync(cls, user: t.Union[str, int], password: str):
-        return async_to_sync(cls.update_password)(user, password)
+        return run_sync(cls.update_password(user, password))
 
     @classmethod
     async def update_password(cls, user: t.Union[str, int], password: str):
@@ -112,7 +111,7 @@ def login_sync(cls, username: str, password: str) -> t.Optional[int]:
         """
         Returns the user_id if a match is found.
         """
-        return async_to_sync(cls.login)(username, password)
+        return run_sync(cls.login(username, password))
 
     @classmethod
     async def login(cls, username: str, password: str) -> t.Optional[int]:

piccolo/utils/sync.py

@@ -1,5 +1,30 @@
 import asyncio
+from concurrent.futures import ThreadPoolExecutor
 
 
 def run_sync(coroutine):
-    return asyncio.run(coroutine)
+    """
+    Run the coroutine synchronously - trying to accomodate as many edge cases
+    as possible.
+
+     1. When called within a coroutine.
+     2. When called from `python -m asyncio`, or iPython with %autoawait
+        enabled, which means an event loop may already be running in the
+        current thread.
+
+    """
+    try:
+        loop = asyncio.get_event_loop()
+    except RuntimeError:
+        return asyncio.run(coroutine)
+    else:
+        if loop.is_running():
+            new_loop = asyncio.new_event_loop()
+
+            with ThreadPoolExecutor(max_workers=1) as executor:
+                future = executor.submit(
+                    new_loop.run_until_complete, coroutine
+                )
+                return future.result()
+        else:
+            return loop.run_until_complete(coroutine)

requirements.txt

@@ -1,5 +1,4 @@
 aiosqlite==0.16.0
-asgiref==3.3.0
 asyncpg==0.21.0
 black==19.10b0
 colorama==0.4.*

tests/utils/test_sync.py

@@ -0,0 +1,34 @@
+import asyncio
+from unittest import TestCase
+
+from piccolo.utils.sync import run_sync
+
+
+class TestSync(TestCase):
+    def test_sync_simple(self):
+        """
+        Test calling a simple coroutine.
+        """
+        run_sync(asyncio.sleep(0.1))
+
+    def test_sync_nested(self):
+        """
+        Test calling a coroutine, which contains a call to `run_sync`.
+        """
+
+        async def test():
+            run_sync(asyncio.sleep(0.1))
+
+        run_sync(test())
+
+    def test_sync_stopped_event_loop(self):
+        """
+        Test calling a coroutine, when the current thread has an event loop,
+        but it isn't running.
+        """
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        run_sync(asyncio.sleep(0.1))
+
+        asyncio.set_event_loop(None)