Repository: django/asgiref

Issue:
Since 3.8.0 ApplicationCommunicator cannot be constructed in a synchronous context
#367 changed the initialisation of the task in `ApplicationCommunicator` from using `asyncio.ensure_future` to using `asyncio.create_task`

While is is generally a positive change and use of `asyncio.create_task` is generally preferred over low-level primitives like `asyncio.ensure_future` this has the side effect that, since `asyncio.create_task` uses `asyncio.get_running_loop` under the hood where `asyncio.ensure_future` used `asyncio.get_event_loop`, an `ApplicationCommunicator` can no longer be constructed outside of a coroutine or callback

In my test suite I'm using `unittest` with synchronous test functions and I have code to explicitly manage event loops and configure them for use using `asyncio.set_event_loop`

I've been able to workaround this using code like the following:
```
def _task():
    return ApplicationCommunicator(*args, **kwargs)
task = loop.create_task(_task())
while not task.done():
    pass
return task.result()
```

I wonder if this code could, instead, be calling `asyncio.get_event_loop` and calling `loop.create_task` instead of `asyncio.create_task` so that it works correctly in a synchronous context with an assigned event loop.

Discussion:
@graingert: 'asyncio.get_event_loop' is deprecated

perhaps we could make ApplicationCommunicator lazily bind the loop@riconnon: @graingert per the docs:
> Deprecated since version 3.12: Deprecation warning is emitted if there is no current event loop. In some future Python release this will become an error.

The function is not deprecated but the behaviour of calling it when there is no current event loop is.

However, your idea of lazy binding does seem like it could work well either way.@graingert: in 3.14 it's deprecated, along with the whole policy system and set_event_loop etc.


edit: it got undeprecated this morning@cjwatson: This change broke tests for https://salsa.debian.org/freexian-team/debusine for some reason after asgiref 3.9.0 was released today.  https://salsa.debian.org/cjwatson/debusine/-/jobs/7787027 shows the failure and I bisected it to this change.  It might be our problem - maybe there's an old exception that was silently being ignored and the new "Make sure there's not an exception to raise from the task" checks pick it up - but I'm still investigating.

By the way, this change apparently isn't mentioned in `CHANGELOG.txt`, which seems like an oversight if it potentially requires users to adapt to it in some way.@carltongibson: @cjwatson Can I ask you to open a fresh issue once you've finished your investigation? 

(Happy to add to the changelog, though I'd recalled the change as transparent when doing the notes for the release 🤔 cc @graingert )

Solution patch:
diff --git a/asgiref/server.py b/asgiref/server.py
index 43c28c6c..131ea430 100644
--- a/asgiref/server.py
+++ b/asgiref/server.py
@@ -57,12 +57,28 @@ def run(self):
         Runs the asyncio event loop with our handler loop.
         """
         event_loop = asyncio.get_event_loop()
-        asyncio.ensure_future(self.application_checker())
         try:
-            event_loop.run_until_complete(self.handle())
+            event_loop.run_until_complete(self.arun())
         except KeyboardInterrupt:
             logger.info("Exiting due to Ctrl-C/interrupt")
 
+    async def arun(self):
+        """
+        Runs the asyncio event loop with our handler loop.
+        """
+
+        class Done(Exception):
+            pass
+
+        async def handle():
+            await self.handle()
+            raise Done
+
+        try:
+            await asyncio.gather(self.application_checker(), handle())
+        except Done:
+            pass
+
     async def handle(self):
         raise NotImplementedError("You must implement handle()")
 
diff --git a/asgiref/sync.py b/asgiref/sync.py
index 377075d1..0c6ea98e 100644
--- a/asgiref/sync.py
+++ b/asgiref/sync.py
@@ -179,15 +179,14 @@ def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R:
 
         # You can't call AsyncToSync from a thread with a running event loop
         try:
-            event_loop = asyncio.get_running_loop()
+            asyncio.get_running_loop()
         except RuntimeError:
             pass
         else:
-            if event_loop.is_running():
-                raise RuntimeError(
-                    "You cannot use AsyncToSync in the same thread as an async event loop - "
-                    "just await the async function directly."
-                )
+            raise RuntimeError(
+                "You cannot use AsyncToSync in the same thread as an async event loop - "
+                "just await the async function directly."
+            )
 
         # Make a future for the return information
         call_result: "Future[_R]" = Future()
@@ -207,7 +206,6 @@ def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R:
         # an asyncio.CancelledError to.
         task_context = getattr(SyncToAsync.threadlocal, "task_context", None)
 
-        loop = None
         # Use call_soon_threadsafe to schedule a synchronous callback on the
         # main event loop's thread if it's there, otherwise make a new loop
         # in this thread.
@@ -225,29 +223,37 @@ def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R:
                 self.awaitable(*args, **kwargs),
             )
 
-            if not (self.main_event_loop and self.main_event_loop.is_running()):
-                # Make our own event loop - in a new thread - and run inside that.
-                loop = asyncio.new_event_loop()
+            async def new_loop_wrap() -> None:
+                loop = asyncio.get_running_loop()
                 self.loop_thread_executors[loop] = current_executor
+                try:
+                    await awaitable
+                finally:
+                    del self.loop_thread_executors[loop]
+
+            if self.main_event_loop is not None:
+                try:
+                    self.main_event_loop.call_soon_threadsafe(
+                        self.main_event_loop.create_task, awaitable
+                    )
+                except RuntimeError:
+                    running_in_main_event_loop = False
+                else:
+                    running_in_main_event_loop = True
+                    # Run the CurrentThreadExecutor until the future is done.
+                    current_executor.run_until_future(call_result)
+            else:
+                running_in_main_event_loop = False
+
+            if not running_in_main_event_loop:
+                # Make our own event loop - in a new thread - and run inside that.
                 loop_executor = ThreadPoolExecutor(max_workers=1)
-                loop_future = loop_executor.submit(
-                    self._run_event_loop, loop, awaitable
-                )
+                loop_future = loop_executor.submit(asyncio.run, new_loop_wrap())
                 # Run the CurrentThreadExecutor until the future is done.
                 current_executor.run_until_future(loop_future)
                 # Wait for future and/or allow for exception propagation
                 loop_future.result()
-            else:
-                # Call it inside the existing loop
-                self.main_event_loop.call_soon_threadsafe(
-                    self.main_event_loop.create_task, awaitable
-                )
-                # Run the CurrentThreadExecutor until the future is done.
-                current_executor.run_until_future(call_result)
         finally:
-            # Clean up any executor we were running
-            if loop is not None:
-                del self.loop_thread_executors[loop]
             _restore_context(context[0])
             # Restore old current thread executor state
             self.executors.current = old_executor
@@ -255,42 +261,6 @@ def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R:
         # Wait for results from the future.
         return call_result.result()
 
-    def _run_event_loop(self, loop, coro):
-        """
-        Runs the given event loop (designed to be called in a thread).
-        """
-        asyncio.set_event_loop(loop)
-        try:
-            loop.run_until_complete(coro)
-        finally:
-            try:
-                # mimic asyncio.run() behavior
-                # cancel unexhausted async generators
-                tasks = asyncio.all_tasks(loop)
-                for task in tasks:
-                    task.cancel()
-
-                async def gather():
-                    await asyncio.gather(*tasks, return_exceptions=True)
-
-                loop.run_until_complete(gather())
-                for task in tasks:
-                    if task.cancelled():
-                        continue
-                    if task.exception() is not None:
-                        loop.call_exception_handler(
-                            {
-                                "message": "unhandled exception during loop shutdown",
-                                "exception": task.exception(),
-                                "task": task,
-                            }
-                        )
-                if hasattr(loop, "shutdown_asyncgens"):
-                    loop.run_until_complete(loop.shutdown_asyncgens())
-            finally:
-                loop.close()
-                asyncio.set_event_loop(self.main_event_loop)
-
     def __get__(self, parent: Any, objtype: Any) -> Callable[_P, _R]:
         """
         Include self for methods
diff --git a/asgiref/testing.py b/asgiref/testing.py
index aa7cff1c..d6ccee58 100644
--- a/asgiref/testing.py
+++ b/asgiref/testing.py
@@ -13,18 +13,40 @@ class ApplicationCommunicator:
     """
 
     def __init__(self, application, scope):
+        self._future = None
         self.application = guarantee_single_callable(application)
         self.scope = scope
-        self.input_queue = asyncio.Queue()
-        self.output_queue = asyncio.Queue()
-        # Clear context - this ensures that context vars set in the testing scope
-        # are not "leaked" into the application which would normally begin with
-        # an empty context. In Python >= 3.11 this could also be written as:
-        # asyncio.create_task(..., context=contextvars.Context())
-        self.future = contextvars.Context().run(
-            asyncio.create_task,
-            self.application(scope, self.input_queue.get, self.output_queue.put),
-        )
+        self._input_queue = None
+        self._output_queue = None
+
+    # For Python 3.9 we need to lazily bind the queues, on 3.10+ they bind the
+    # event loop lazily.
+    @property
+    def input_queue(self):
+        if self._input_queue is None:
+            self._input_queue = asyncio.Queue()
+        return self._input_queue
+
+    @property
+    def output_queue(self):
+        if self._output_queue is None:
+            self._output_queue = asyncio.Queue()
+        return self._output_queue
+
+    @property
+    def future(self):
+        if self._future is None:
+            # Clear context - this ensures that context vars set in the testing scope
+            # are not "leaked" into the application which would normally begin with
+            # an empty context. In Python >= 3.11 this could also be written as:
+            # asyncio.create_task(..., context=contextvars.Context())
+            self._future = contextvars.Context().run(
+                asyncio.create_task,
+                self.application(
+                    self.scope, self.input_queue.get, self.output_queue.put
+                ),
+            )
+        return self._future
 
     async def wait(self, timeout=1):
         """
@@ -46,11 +68,15 @@ async def wait(self, timeout=1):
                     pass
 
     def stop(self, exceptions=True):
-        if not self.future.done():
-            self.future.cancel()
+        future = self._future
+        if future is None:
+            return
+
+        if not future.done():
+            future.cancel()
         elif exceptions:
             # Give a chance to raise any exceptions
-            self.future.result()
+            future.result()
 
     def __del__(self):
         # Clean up on deletion
@@ -64,6 +90,10 @@ async def send_input(self, message):
         """
         Sends a single message to the application
         """
+        # Make sure there's not an exception to raise from the task
+        if self.future.done():
+            self.future.result()
+
         # Give it the message
         await self.input_queue.put(message)
 
@@ -94,6 +124,10 @@ async def receive_nothing(self, timeout=0.1, interval=0.01):
         """
         Checks that there is no message to receive in the given time.
         """
+        # Make sure there's not an exception to raise from the task
+        if self.future.done():
+            self.future.result()
+
         # `interval` has precedence over `timeout`
         start = time.monotonic()
         while time.monotonic() - start < timeout:
diff --git a/docs/implementations.rst b/docs/implementations.rst
index 6a7729cd..fba4106f 100644
--- a/docs/implementations.rst
+++ b/docs/implementations.rst
@@ -118,6 +118,23 @@ standard Python type annotations and standards like OpenAPI, JSON Schema, and OA
 Supports HTTP and WebSockets.
 
 
+Flama
+-----
+
+*Stable* / https://github.com/vortico/flama
+
+Flama is a data-science oriented framework to rapidly build modern and robust machine 
+learning (ML) APIs. The main aim of the framework is to make ridiculously simple the 
+deployment of ML APIs. With Flama, data scientists can now quickly turn their ML models 
+into asynchronous, auto-documented APIs with just a single line of code. All in just few 
+seconds!
+
+Flama comes with an intuitive CLI, and provides an easy-to-learn philosophy to speed up 
+the building of highly performant GraphQL, REST, and ML APIs. Besides, it comprises an 
+ideal solution for the development of asynchronous and production-ready services, 
+offering automatic deployment for ML models. 
+
+
 Litestar
 --------
 
diff --git a/setup.cfg b/setup.cfg
index 1d029688..ef0a4314 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -44,6 +44,7 @@ tests =
 [tool:pytest]
 testpaths = tests
 asyncio_mode = strict
+asyncio_default_fixture_loop_scope=function
 
 [flake8]
 exclude = venv/*,tox/*,specs/*


Test patch:
diff --git a/tests/test_server.py b/tests/test_server.py
index 65f80043..f8aad448 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -1,8 +1,8 @@
 import asyncio
 import socket as sock
-from functools import partial
 
 import pytest
+import pytest_asyncio
 
 from asgiref.server import StatelessServer
 
@@ -74,8 +74,8 @@ def close(self):
         self._sock.close()
 
 
-@pytest.fixture(scope="function")
-def server():
+@pytest_asyncio.fixture(scope="function")
+async def server():
     async def app(scope, receive, send):
         while True:
             msg = await receive()
@@ -92,25 +92,12 @@ async def check_client_msg(client, expected_address, expected_msg):
     assert server_addr == expected_address
 
 
-async def server_auto_close(fut, timeout):
-    """Server run based on run_until_complete. It will block forever with handle
-    function because it is a while True loop without break.  Use this method to close
-    server automatically."""
-    loop = asyncio.get_running_loop()
-    task = asyncio.ensure_future(fut, loop=loop)
-    await asyncio.sleep(timeout)
-    task.cancel()
-
-
-def test_stateless_server(server):
+@pytest.mark.asyncio
+async def test_stateless_server(server):
     """StatelessServer can be instantiated with an ASGI 3 application."""
     """Create a UDP Server can register instance based on name from message of client.
     Clients can communicate to other client by name through server"""
 
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
-    server.handle = partial(server_auto_close, fut=server.handle(), timeout=1.0)
-
     client1 = Client(name="client1")
     client2 = Client(name="client2")
 
@@ -124,30 +111,35 @@ async def check_client2_behavior():
         await check_client_msg(client2, server.address, b"Welcome")
         await check_client_msg(client2, server.address, b"Hello")
 
-    task1 = loop.create_task(check_client1_behavior())
-    task2 = loop.create_task(check_client2_behavior())
+    class Done(Exception):
+        pass
 
-    server.run()
+    async def do_test():
+        await asyncio.gather(check_client1_behavior(), check_client2_behavior())
+        raise Done
 
-    assert task1.done()
-    assert task2.done()
+    try:
+        await asyncio.gather(server.arun(), do_test())
+    except Done:
+        pass
 
 
-def test_server_delete_instance(server):
+@pytest.mark.asyncio
+async def test_server_delete_instance(server):
     """The max_applications of Server is 10. After 20 times register, application number should be 10."""
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
-    server.handle = partial(server_auto_close, fut=server.handle(), timeout=1.0)
-
     client1 = Client(name="client1")
 
+    class Done(Exception):
+        pass
+
     async def client1_multiple_register():
         for i in range(20):
             await client1.register(server.address, name=f"client{i}")
             print(f"client{i}")
             await check_client_msg(client1, server.address, b"Welcome")
+        raise Done
 
-    task = loop.create_task(client1_multiple_register())
-    server.run()
-
-    assert task.done()
+    try:
+        await asyncio.gather(client1_multiple_register(), server.arun())
+    except Done:
+        pass
diff --git a/tests/test_testing.py b/tests/test_testing.py
index 62338961..aec5bf9c 100644
--- a/tests/test_testing.py
+++ b/tests/test_testing.py
@@ -1,3 +1,5 @@
+import asyncio
+
 import pytest
 
 from asgiref.testing import ApplicationCommunicator
@@ -43,3 +45,46 @@ def wsgi_application(environ, start_response):
     await instance.receive_output()
     # Response received completely
     assert await instance.receive_nothing(0.01) is True
+
+
+def test_receive_nothing_lazy_loop():
+    """
+    Tests ApplicationCommunicator.receive_nothing to return the correct value.
+    """
+    # Get an ApplicationCommunicator instance
+    def wsgi_application(environ, start_response):
+        start_response("200 OK", [])
+        yield b"content"
+
+    application = WsgiToAsgi(wsgi_application)
+    instance = ApplicationCommunicator(
+        application,
+        {
+            "type": "http",
+            "http_version": "1.0",
+            "method": "GET",
+            "path": "/foo/",
+            "query_string": b"bar=baz",
+            "headers": [],
+        },
+    )
+
+    async def test():
+        # No event
+        assert await instance.receive_nothing() is True
+
+        # Produce 3 events to receive
+        await instance.send_input({"type": "http.request"})
+        # Start event of the response
+        assert await instance.receive_nothing() is False
+        await instance.receive_output()
+        # First body event of the response announcing further body event
+        assert await instance.receive_nothing() is False
+        await instance.receive_output()
+        # Last body event of the response
+        assert await instance.receive_nothing() is False
+        await instance.receive_output()
+        # Response received completely
+        assert await instance.receive_nothing(0.01) is True
+
+    asyncio.run(test())