anyio integration

README.md

@@ -24,7 +24,7 @@
 # Starlette
 
 Starlette is a lightweight [ASGI](https://asgi.readthedocs.io/en/latest/) framework/toolkit,
-which is ideal for building high performance asyncio services.
+which is ideal for building high performance async services.
 
 It is production-ready, and gives you the following:
 
@@ -38,7 +38,8 @@ It is production-ready, and gives you the following:
 * Session and Cookie support.
 * 100% test coverage.
 * 100% type annotated codebase.
-* Zero hard dependencies.
+* Few hard dependencies.
+* Compatible with `asyncio` and `trio` backends.
 
 ## Requirements
 
@@ -86,7 +87,7 @@ For a more complete example, see [encode/starlette-example](https://github.com/e
 
 ## Dependencies
 
-Starlette does not have any hard dependencies, but the following are optional:
+Starlette only requires `anyio`, and the following are optional:
 
 * [`requests`][requests] - Required if you want to use the `TestClient`.
 * [`aiofiles`][aiofiles] - Required if you want to use `FileResponse` or `StaticFiles`.

setup.py

@@ -48,6 +48,7 @@ def get_packages(package):
     packages=get_packages("starlette"),
     package_data={"starlette": ["py.typed"]},
     include_package_data=True,
+    install_requires=["anyio<3,>=2"],
     extras_require={
         "full": [
             "aiofiles",

starlette/applications.py

@@ -1,5 +1,7 @@
 import typing
 
+import anyio
+
 from starlette.datastructures import State, URLPath
 from starlette.exceptions import ExceptionMiddleware
 from starlette.middleware import Middleware
@@ -109,7 +111,9 @@ def url_path_for(self, name: str, **path_params: str) -> URLPath:
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         scope["app"] = self
-        await self.middleware_stack(scope, receive, send)
+        task_group = scope["task_group"] = anyio.create_task_group()
+        async with task_group:
+            await self.middleware_stack(scope, receive, send)
 
     # The following usages are now discouraged in favour of configuration
     #  during Starlette.__init__(...)

starlette/concurrency.py

@@ -1,43 +1,46 @@
-import asyncio
 import functools
-import sys
 import typing
 from typing import Any, AsyncGenerator, Iterator
 
+import anyio
+
 try:
     import contextvars  # Python 3.7+ only or via contextvars backport.
 except ImportError:  # pragma: no cover
     contextvars = None  # type: ignore
 
-if sys.version_info >= (3, 7):  # pragma: no cover
-    from asyncio import create_task
-else:  # pragma: no cover
-    from asyncio import ensure_future as create_task
 
 T = typing.TypeVar("T")
 
 
 async def run_until_first_complete(*args: typing.Tuple[typing.Callable, dict]) -> None:
-    tasks = [create_task(handler(**kwargs)) for handler, kwargs in args]
-    (done, pending) = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
-    [task.cancel() for task in pending]
-    [task.result() for task in done]
+    result: Any = None
+    async with anyio.create_task_group() as task_group:
+
+        async def task(_handler: typing.Callable, _kwargs: dict) -> Any:
+            nonlocal result
+            result = await _handler(**_kwargs)
+            await task_group.cancel_scope.cancel()
+
+        for handler, kwargs in args:
+            await task_group.spawn(task, handler, kwargs)
+
+    return result
 
 
 async def run_in_threadpool(
     func: typing.Callable[..., T], *args: typing.Any, **kwargs: typing.Any
 ) -> T:
-    loop = asyncio.get_event_loop()
     if contextvars is not None:  # pragma: no cover
         # Ensure we run in the same context
         child = functools.partial(func, *args, **kwargs)
         context = contextvars.copy_context()
         func = context.run
         args = (child,)
     elif kwargs:  # pragma: no cover
-        # loop.run_in_executor doesn't accept 'kwargs', so bind them in here
+        # run_sync doesn't accept 'kwargs', so bind them in here
         func = functools.partial(func, **kwargs)
-    return await loop.run_in_executor(None, func, *args)
+    return await anyio.run_sync_in_worker_thread(func, *args)
 
 
 class _StopIteration(Exception):
@@ -57,6 +60,6 @@ def _next(iterator: Iterator) -> Any:
 async def iterate_in_threadpool(iterator: Iterator) -> AsyncGenerator:
     while True:
         try:
-            yield await run_in_threadpool(_next, iterator)
+            yield await anyio.run_sync_in_worker_thread(_next, iterator)
         except _StopIteration:
             break

starlette/middleware/base.py

@@ -1,9 +1,10 @@
-import asyncio
 import typing
 
+import anyio
+
 from starlette.requests import Request
 from starlette.responses import Response, StreamingResponse
-from starlette.types import ASGIApp, Message, Receive, Scope, Send
+from starlette.types import ASGIApp, Receive, Scope, Send
 
 RequestResponseEndpoint = typing.Callable[[Request], typing.Awaitable[Response]]
 DispatchFunction = typing.Callable[
@@ -26,34 +27,28 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         await response(scope, receive, send)
 
     async def call_next(self, request: Request) -> Response:
-        loop = asyncio.get_event_loop()
-        queue: "asyncio.Queue[typing.Optional[Message]]" = asyncio.Queue()
-
+        send_stream, recv_stream = anyio.create_memory_object_stream()
         scope = request.scope
-        receive = request.receive
-        send = queue.put
+        task_group = scope["task_group"]
 
         async def coro() -> None:
-            try:
-                await self.app(scope, receive, send)
-            finally:
-                await queue.put(None)
+            async with send_stream:
+                await self.app(scope, recv_stream.receive, send_stream.send)
 
-        task = loop.create_task(coro())
-        message = await queue.get()
-        if message is None:
-            task.result()
+        await task_group.spawn(coro)
+
+        try:
+            message = await recv_stream.receive()
+        except anyio.EndOfStream:
             raise RuntimeError("No response returned.")
+
         assert message["type"] == "http.response.start"
 
         async def body_stream() -> typing.AsyncGenerator[bytes, None]:
-            while True:
-                message = await queue.get()
-                if message is None:
-                    break
-                assert message["type"] == "http.response.body"
-                yield message.get("body", b"")
-            task.result()
+            async with recv_stream:
+                async for message in recv_stream:
+                    assert message["type"] == "http.response.body"
+                    yield message.get("body", b"")
 
         response = StreamingResponse(
             status_code=message["status"], content=body_stream()

starlette/middleware/wsgi.py

@@ -1,10 +1,10 @@
-import asyncio
 import io
 import sys
 import typing
 
-from starlette.concurrency import run_in_threadpool
-from starlette.types import Message, Receive, Scope, Send
+import anyio
+
+from starlette.types import Receive, Scope, Send
 
 
 def build_environ(scope: Scope, body: bytes) -> dict:
@@ -69,9 +69,7 @@ def __init__(self, app: typing.Callable, scope: Scope) -> None:
         self.scope = scope
         self.status = None
         self.response_headers = None
-        self.send_event = asyncio.Event()
-        self.send_queue = []  # type: typing.List[typing.Optional[Message]]
-        self.loop = asyncio.get_event_loop()
+        self.stream_send, self.stream_receive = anyio.create_memory_object_stream()
         self.response_started = False
         self.exc_info = None  # type: typing.Any
 
@@ -83,31 +81,25 @@ async def __call__(self, receive: Receive, send: Send) -> None:
             body += message.get("body", b"")
             more_body = message.get("more_body", False)
         environ = build_environ(self.scope, body)
-        sender = None
-        try:
-            sender = self.loop.create_task(self.sender(send))
-            await run_in_threadpool(self.wsgi, environ, self.start_response)
-            self.send_queue.append(None)
-            self.send_event.set()
-            await asyncio.wait_for(sender, None)
-            if self.exc_info is not None:
-                raise self.exc_info[0].with_traceback(
-                    self.exc_info[1], self.exc_info[2]
-                )
-        finally:
-            if sender and not sender.done():
-                sender.cancel()  # pragma: no cover
+
+        async with anyio.create_task_group() as task_group:
+            try:
+                await task_group.spawn(self.sender, send)
+                async with self.stream_send:
+                    await anyio.run_sync_in_worker_thread(
+                        self.wsgi, environ, self.start_response
+                    )
+                if self.exc_info is not None:
+                    raise self.exc_info[0].with_traceback(
+                        self.exc_info[1], self.exc_info[2]
+                    )
+            finally:
+                await task_group.cancel_scope.cancel()
 
     async def sender(self, send: Send) -> None:
-        while True:
-            if self.send_queue:
-                message = self.send_queue.pop(0)
-                if message is None:
-                    return
+        async with self.stream_receive:
+            async for message in self.stream_receive:
                 await send(message)
-            else:
-                await self.send_event.wait()
-                self.send_event.clear()
 
     def start_response(
         self,
@@ -124,21 +116,22 @@ def start_response(
                 (name.strip().encode("ascii").lower(), value.strip().encode("ascii"))
                 for name, value in response_headers
             ]
-            self.send_queue.append(
+            anyio.run_async_from_thread(
+                self.stream_send.send,
                 {
                     "type": "http.response.start",
                     "status": status_code,
                     "headers": headers,
-                }
+                },
             )
-            self.loop.call_soon_threadsafe(self.send_event.set)
 
     def wsgi(self, environ: dict, start_response: typing.Callable) -> None:
         for chunk in self.app(environ, start_response):
-            self.send_queue.append(
-                {"type": "http.response.body", "body": chunk, "more_body": True}
+            anyio.run_async_from_thread(
+                self.stream_send.send,
+                {"type": "http.response.body", "body": chunk, "more_body": True},
             )
-            self.loop.call_soon_threadsafe(self.send_event.set)
 
-        self.send_queue.append({"type": "http.response.body", "body": b""})
-        self.loop.call_soon_threadsafe(self.send_event.set)
+        anyio.run_async_from_thread(
+            self.stream_send.send, {"type": "http.response.body", "body": b""}
+        )

starlette/requests.py

@@ -1,9 +1,10 @@
-import asyncio
 import json
 import typing
 from collections.abc import Mapping
 from http import cookies as http_cookies
 
+import anyio
+
 from starlette.datastructures import URL, Address, FormData, Headers, QueryParams, State
 from starlette.formparsers import FormParser, MultiPartParser
 from starlette.types import Message, Receive, Scope, Send
@@ -251,10 +252,11 @@ async def close(self) -> None:
 
     async def is_disconnected(self) -> bool:
         if not self._is_disconnected:
-            try:
-                message = await asyncio.wait_for(self._receive(), timeout=0.0000001)
-            except asyncio.TimeoutError:
-                message = {}
+            message: Message = {}
+            async with anyio.move_on_after(
+                0.001
+            ):  # XXX: to small of a deadline and this blocks
+                message = await self._receive()
 
             if message.get("type") == "http.disconnect":
                 self._is_disconnected = True