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`.

docs/index.md

@@ -82,7 +82,6 @@ For a more complete example, [see here](https://github.com/encode/starlette-exam
 Starlette does not have any hard dependencies, but 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`.
 * [`jinja2`][jinja2] - Required if you want to use `Jinja2Templates`.
 * [`python-multipart`][python-multipart] - Required if you want to support form parsing, with `request.form()`.
 * [`itsdangerous`][itsdangerous] - Required for `SessionMiddleware` support.
@@ -161,7 +160,6 @@ gunicorn -k uvicorn.workers.UvicornH11Worker ...
 <p align="center"><i>Starlette is <a href="https://github.com/encode/starlette/blob/master/LICENSE.md">BSD licensed</a> code. Designed & built in Brighton, England.</i></p>
 
 [requests]: http://docs.python-requests.org/en/master/
-[aiofiles]: https://github.com/Tinche/aiofiles
 [jinja2]: http://jinja.pocoo.org/
 [python-multipart]: https://andrew-d.github.io/python-multipart/
 [graphene]: https://graphene-python.org/

docs/testclient.md

@@ -31,6 +31,21 @@ application. Occasionally you might want to test the content of 500 error
 responses, rather than allowing client to raise the server exception. In this
 case you should use `client = TestClient(app, raise_server_exceptions=False)`.
 
+### Selecting the Async backend
+
+`TestClient.async_backend` is a dictionary which allows you to set the options
+for the backend used to run tests.  These options are passed to
+`anyio.start_blocking_portal()`.  By default, `asyncio` is used.
+
+To run `Trio`, set `async_backend["backend"] = "trio"`, for example:
+
+```python
+def test_app()
+    client = TestClient(app)
+    client.async_backend["backend"] = "trio"
+    ...
+```
+
 ### Testing WebSocket sessions
 
 You can also test websocket sessions with the test client.
@@ -72,6 +87,8 @@ always raised by the test client.
 
 May raise `starlette.websockets.WebSocketDisconnect` if the application does not accept the websocket connection.
 
+`websocket_connect()` must be used as a context manager (in a `with` block).
+
 #### Sending data
 
 * `.send_text(data)` - Send the given text to the application.

requirements.txt

@@ -16,7 +16,7 @@ isort==5.*
 mypy
 pytest
 pytest-cov
-pytest-asyncio
+trio
 
 # Documentation
 mkdocs

setup.py

@@ -48,9 +48,9 @@ def get_packages(package):
     packages=get_packages("starlette"),
     package_data={"starlette": ["py.typed"]},
     include_package_data=True,
+    install_requires=["anyio>=3.0.0rc3"],
     extras_require={
         "full": [
-            "aiofiles",
             "graphene",
             "itsdangerous",
             "jinja2",

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,42 @@
-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]
+    async with anyio.create_task_group() as task_group:
+
+        async def task(_handler: typing.Callable, _kwargs: dict) -> Any:
+            await _handler(**_kwargs)
+            task_group.cancel_scope.cancel()
+
+        for handler, kwargs in args:
+            task_group.spawn(task, handler, kwargs)
 
 
 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.to_thread.run_sync(func, *args)
 
 
 class _StopIteration(Exception):
@@ -57,6 +56,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.to_thread.run_sync(_next, iterator)
         except _StopIteration:
             break

starlette/graphql.py

@@ -31,29 +31,18 @@ class GraphQLApp:
     def __init__(
         self,
         schema: "graphene.Schema",
-        executor: typing.Any = None,
         executor_class: type = None,
         graphiql: bool = True,
     ) -> None:
         self.schema = schema
         self.graphiql = graphiql
-        if executor is None:
-            # New style in 0.10.0. Use 'executor_class'.
-            # See issue https://github.com/encode/starlette/issues/242
-            self.executor = executor
-            self.executor_class = executor_class
-            self.is_async = executor_class is not None and issubclass(
-                executor_class, AsyncioExecutor
-            )
-        else:
-            # Old style. Use 'executor'.
-            # We should remove this in the next median/major version bump.
-            self.executor = executor
-            self.executor_class = None
-            self.is_async = isinstance(executor, AsyncioExecutor)
+        self.executor_class = executor_class
+        self.is_async = executor_class is not None and issubclass(
+            executor_class, AsyncioExecutor
+        )
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
-        if self.executor is None and self.executor_class is not None:
+        if self.executor_class is not None:
             self.executor = self.executor_class()
 
         request = Request(scope, receive=receive)

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, request.receive, send_stream.send)
 
-        task = loop.create_task(coro())
-        message = await queue.get()
-        if message is None:
-            task.result()
+        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,11 @@
-import asyncio
 import io
+import math
 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 +70,9 @@ 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(
+            math.inf
+        )
         self.response_started = False
         self.exc_info = None  # type: typing.Any
 
@@ -83,31 +84,18 @@ 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:
+            task_group.spawn(self.sender, send)
+            async with self.stream_send:
+                await anyio.to_thread.run_sync(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])
 
     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 +112,22 @@ def start_response(
                 (name.strip().encode("ascii").lower(), value.strip().encode("ascii"))
                 for name, value in response_headers
             ]
-            self.send_queue.append(
+            anyio.from_thread.run(
+                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.from_thread.run(
+                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.from_thread.run(
+            self.stream_send.send, {"type": "http.response.body", "body": b""}
+        )