Merge pull request #3412 from pgjones/async

Add `async` support

CHANGES.rst

@@ -67,6 +67,8 @@ Unreleased
 -   Add route decorators for common HTTP methods. For example,
     ``@app.post("/login")`` is a shortcut for
     ``@app.route("/login", methods=["POST"])``. :pr:`3907`
+-   Support async views, error handlers, before and after request, and
+    teardown functions. :pr:`3412`
 
 
 Version 1.1.2

docs/async-await.rst

@@ -0,0 +1,81 @@
+.. _async_await:
+
+Using ``async`` and ``await``
+=============================
+
+.. versionadded:: 2.0
+
+Routes, error handlers, before request, after request, and teardown
+functions can all be coroutine functions if Flask is installed with the
+``async`` extra (``pip install flask[async]``). This allows views to be
+defined with ``async def`` and use ``await``.
+
+.. code-block:: python
+
+    @app.route("/get-data")
+    async def get_data():
+        data = await async_db_query(...)
+        return jsonify(data)
+
+
+Performance
+-----------
+
+Async functions require an event loop to run. Flask, as a WSGI
+application, uses one worker to handle one request/response cycle.
+When a request comes in to an async view, Flask will start an event loop
+in a thread, run the view function there, then return the result.
+
+Each request still ties up one worker, even for async views. The upside
+is that you can run async code within a view, for example to make
+multiple concurrent database queries, HTTP requests to an external API,
+etc. However, the number of requests your application can handle at one
+time will remain the same.
+
+**Async is not inherently faster than sync code.** Async is beneficial
+when performing concurrent IO-bound tasks, but will probably not improve
+CPU-bound tasks. Traditional Flask views will still be appropriate for
+most use cases, but Flask's async support enables writing and using
+code that wasn't possible natively before.
+
+
+When to use Quart instead
+-------------------------
+
+Flask's async support is less performant than async-first frameworks due
+to the way it is implemented. If you have a mainly async codebase it
+would make sense to consider `Quart`_. Quart is a reimplementation of
+Flask based on the `ASGI`_ standard instead of WSGI. This allows it to
+handle many concurrent requests, long running requests, and websockets
+without requiring individual worker processes or threads.
+
+It has also already been possible to run Flask with Gevent or Eventlet
+to get many of the benefits of async request handling. These libraries
+patch low-level Python functions to accomplish this, whereas ``async``/
+``await`` and ASGI use standard, modern Python capabilities. Deciding
+whether you should use Flask, Quart, or something else is ultimately up
+to understanding the specific needs of your project.
+
+.. _Quart: https://gitlab.com/pgjones/quart
+.. _ASGI: https://asgi.readthedocs.io/en/latest/
+
+
+Extensions
+----------
+
+Existing Flask extensions only expect views to be synchronous. If they
+provide decorators to add functionality to views, those will probably
+not work with async views because they will not await the function or be
+awaitable. Other functions they provide will not be awaitable either and
+will probably be blocking if called within an async view.
+
+Check the changelog of the extension you want to use to see if they've
+implemented async support, or make a feature request or PR to them.
+
+
+Other event loops
+-----------------
+
+At the moment Flask only supports :mod:`asyncio`. It's possible to
+override :meth:`flask.Flask.ensure_sync` to change how async functions
+are wrapped to use a different library.

docs/design.rst

@@ -171,6 +171,25 @@ Also see the :doc:`/becomingbig` section of the documentation for some
 inspiration for larger applications based on Flask.
 
 
+Async/await and ASGI support
+----------------------------
+
+Flask supports ``async`` coroutines for view functions by executing the
+coroutine on a separate thread instead of using an event loop on the
+main thread as an async-first (ASGI) framework would. This is necessary
+for Flask to remain backwards compatible with extensions and code built
+before ``async`` was introduced into Python. This compromise introduces
+a performance cost compared with the ASGI frameworks, due to the
+overhead of the threads.
+
+Due to how tied to WSGI Flask's code is, it's not clear if it's possible
+to make the ``Flask`` class support ASGI and WSGI at the same time. Work
+is currently being done in Werkzeug to work with ASGI, which may
+eventually enable support in Flask as well.
+
+See :doc:`/async-await` for more discussion.
+
+
 What Flask is, What Flask is Not
 --------------------------------
 

docs/index.rst

@@ -59,6 +59,7 @@ instructions for web development with Flask.
    patterns/index
    deploying/index
    becomingbig
+   async-await
 
 
 API Reference

requirements/tests.in

@@ -1,4 +1,5 @@
 pytest
+asgiref
 blinker
 greenlet
 python-dotenv

requirements/tests.txt

@@ -4,6 +4,8 @@
 #
 #    pip-compile requirements/tests.in
 #
+asgiref==3.2.10
+    # via -r requirements/tests.in
 attrs==20.3.0
     # via pytest
 blinker==1.4

setup.py

@@ -9,5 +9,8 @@ setup(
         "itsdangerous>=0.24",
         "click>=5.1",
     ],
-    extras_require={"dotenv": ["python-dotenv"]},
+    extras_require={
+        "async": ["asgiref>=3.2"],
+        "dotenv": ["python-dotenv"],
+    },
 )

src/flask/app.py

@@ -2,6 +2,7 @@ import os
 import sys
 import weakref
 from datetime import timedelta
+from inspect import iscoroutinefunction
 from itertools import chain
 from threading import Lock
 
@@ -34,6 +35,7 @@ from .helpers import get_env
 from .helpers import get_flashed_messages
 from .helpers import get_load_dotenv
 from .helpers import locked_cached_property
+from .helpers import run_async
 from .helpers import url_for
 from .json import jsonify
 from .logging import create_logger
@@ -1050,7 +1052,7 @@ class Flask(Scaffold):
                     "View function mapping is overwriting an existing"
                     f" endpoint function: {endpoint}"
                 )
-            self.view_functions[endpoint] = view_func
+            self.view_functions[endpoint] = self.ensure_sync(view_func)
 
     @setupmethod
     def template_filter(self, name=None):
@@ -1165,7 +1167,7 @@ class Flask(Scaffold):
 
         .. versionadded:: 0.8
         """
-        self.before_first_request_funcs.append(f)
+        self.before_first_request_funcs.append(self.ensure_sync(f))
         return f
 
     @setupmethod
@@ -1198,7 +1200,7 @@ class Flask(Scaffold):
 
         .. versionadded:: 0.9
         """
-        self.teardown_appcontext_funcs.append(f)
+        self.teardown_appcontext_funcs.append(self.ensure_sync(f))
         return f
 
     @setupmethod
@@ -1517,6 +1519,20 @@ class Flask(Scaffold):
         """
         return False
 
+    def ensure_sync(self, func):
+        """Ensure that the function is synchronous for WSGI workers.
+        Plain ``def`` functions are returned as-is. ``async def``
+        functions are wrapped to run and wait for the response.
+
+        Override this method to change how the app runs async views.
+
+        .. versionadded:: 2.0
+        """
+        if iscoroutinefunction(func):
+            return run_async(func)
+
+        return func
+
     def make_response(self, rv):
         """Convert the return value from a view function to an instance of
         :attr:`response_class`.

src/flask/blueprints.py

@@ -1,3 +1,4 @@
+from collections import defaultdict
 from functools import update_wrapper
 
 from .scaffold import _endpoint_from_view_func
@@ -235,24 +236,44 @@ class Blueprint(Scaffold):
         # Merge blueprint data into parent.
         if first_registration:
 
-            def extend(bp_dict, parent_dict):
+            def extend(bp_dict, parent_dict, ensure_sync=False):
                 for key, values in bp_dict.items():
                     key = self.name if key is None else f"{self.name}.{key}"
+
+                    if ensure_sync:
+                        values = [app.ensure_sync(func) for func in values]
+
                     parent_dict[key].extend(values)
 
-            def update(bp_dict, parent_dict):
+            for key, value in self.error_handler_spec.items():
-                for key, value in bp_dict.items():
+                key = self.name if key is None else f"{self.name}.{key}"
-                    key = self.name if key is None else f"{self.name}.{key}"
+                value = defaultdict(
-                    parent_dict[key] = value
+                    dict,
+                    {
+                        code: {
+                            exc_class: app.ensure_sync(func)
+                            for exc_class, func in code_values.items()
+                        }
+                        for code, code_values in value.items()
+                    },
+                )
+                app.error_handler_spec[key] = value
 
-            app.view_functions.update(self.view_functions)
+            for endpoint, func in self.view_functions.items():
-            extend(self.before_request_funcs, app.before_request_funcs)
+                app.view_functions[endpoint] = app.ensure_sync(func)
-            extend(self.after_request_funcs, app.after_request_funcs)
+
-            extend(self.teardown_request_funcs, app.teardown_request_funcs)
+            extend(
+                self.before_request_funcs, app.before_request_funcs, ensure_sync=True
+            )
+            extend(self.after_request_funcs, app.after_request_funcs, ensure_sync=True)
+            extend(
+                self.teardown_request_funcs,
+                app.teardown_request_funcs,
+                ensure_sync=True,
+            )
             extend(self.url_default_functions, app.url_default_functions)
             extend(self.url_value_preprocessors, app.url_value_preprocessors)
             extend(self.template_context_processors, app.template_context_processors)
-            update(self.error_handler_spec, app.error_handler_spec)
 
         for deferred in self.deferred_functions:
             deferred(state)
@@ -380,7 +401,9 @@ class Blueprint(Scaffold):
         before each request, even if outside of a blueprint.
         """
         self.record_once(
-            lambda s: s.app.before_request_funcs.setdefault(None, []).append(f)
+            lambda s: s.app.before_request_funcs.setdefault(None, []).append(
+                s.app.ensure_sync(f)
+            )
         )
         return f
 
@@ -388,7 +411,9 @@ class Blueprint(Scaffold):
         """Like :meth:`Flask.before_first_request`.  Such a function is
         executed before the first request to the application.
         """
-        self.record_once(lambda s: s.app.before_first_request_funcs.append(f))
+        self.record_once(
+            lambda s: s.app.before_first_request_funcs.append(s.app.ensure_sync(f))
+        )
         return f
 
     def after_app_request(self, f):
@@ -396,7 +421,9 @@ class Blueprint(Scaffold):
         is executed after each request, even if outside of the blueprint.
         """
         self.record_once(
-            lambda s: s.app.after_request_funcs.setdefault(None, []).append(f)
+            lambda s: s.app.after_request_funcs.setdefault(None, []).append(
+                s.app.ensure_sync(f)
+            )
         )
         return f
 
@@ -443,3 +470,14 @@ class Blueprint(Scaffold):
             lambda s: s.app.url_default_functions.setdefault(None, []).append(f)
         )
         return f
+
+    def ensure_sync(self, f):
+        """Ensure the function is synchronous.
+
+        Override if you would like custom async to sync behaviour in
+        this blueprint. Otherwise the app's
+        :meth:`~flask.Flask.ensure_sync` is used.
+
+        .. versionadded:: 2.0
+        """
+        return f

src/flask/helpers.py

@@ -2,10 +2,12 @@ import os
 import socket
 import warnings
 from functools import update_wrapper
+from functools import wraps
 from threading import RLock
 
 import werkzeug.utils
 from werkzeug.exceptions import NotFound
+from werkzeug.local import ContextVar
 from werkzeug.routing import BuildError
 from werkzeug.urls import url_quote
 
@@ -729,3 +731,50 @@ def is_ip(value):
             return True
 
     return False
+
+
+def run_async(func):
+    """Return a sync function that will run the coroutine function *func*."""
+    try:
+        from asgiref.sync import async_to_sync
+    except ImportError:
+        raise RuntimeError(
+            "Install Flask with the 'async' extra in order to use async views."
+        )
+
+    # Check that Werkzeug isn't using its fallback ContextVar class.
+    if ContextVar.__module__ == "werkzeug.local":
+        raise RuntimeError(
+            "Async cannot be used with this combination of Python & Greenlet versions."
+        )
+
+    @wraps(func)
+    def outer(*args, **kwargs):
+        """This function grabs the current context for the inner function.
+
+        This is similar to the copy_current_xxx_context functions in the
+        ctx module, except it has an async inner.
+        """
+        ctx = None
+
+        if _request_ctx_stack.top is not None:
+            ctx = _request_ctx_stack.top.copy()
+
+        @wraps(func)
+        async def inner(*a, **k):
+            """This restores the context before awaiting the func.
+
+            This is required as the function must be awaited within the
+            context. Only calling ``func`` (as per the
+            ``copy_current_xxx_context`` functions) doesn't work as the
+            with block will close before the coroutine is awaited.
+            """
+            if ctx is not None:
+                with ctx:
+                    return await func(*a, **k)
+            else:
+                return await func(*a, **k)
+
+        return async_to_sync(inner)(*args, **kwargs)
+
+    return outer

src/flask/scaffold.py

@@ -484,7 +484,7 @@ class Scaffold:
         """
 
         def decorator(f):
-            self.view_functions[endpoint] = f
+            self.view_functions[endpoint] = self.ensure_sync(f)
             return f
 
         return decorator
@@ -508,7 +508,7 @@ class Scaffold:
         return value from the view, and further request handling is
         stopped.
         """
-        self.before_request_funcs[None].append(f)
+        self.before_request_funcs.setdefault(None, []).append(self.ensure_sync(f))
         return f
 
     @setupmethod
@@ -524,7 +524,7 @@ class Scaffold:
         should not be used for actions that must execute, such as to
         close resources. Use :meth:`teardown_request` for that.
         """
-        self.after_request_funcs[None].append(f)
+        self.after_request_funcs.setdefault(None, []).append(self.ensure_sync(f))
         return f
 
     @setupmethod
@@ -563,7 +563,7 @@ class Scaffold:
            debugger can still access it.  This behavior can be controlled
            by the ``PRESERVE_CONTEXT_ON_EXCEPTION`` configuration variable.
         """
-        self.teardown_request_funcs[None].append(f)
+        self.teardown_request_funcs.setdefault(None, []).append(self.ensure_sync(f))
         return f
 
     @setupmethod
@@ -659,7 +659,7 @@ class Scaffold:
                 " instead."
             )
 
-        self.error_handler_spec[None][code][exc_class] = f
+        self.error_handler_spec[None][code][exc_class] = self.ensure_sync(f)
 
     @staticmethod
     def _get_exc_class_and_code(exc_class_or_code):
@@ -684,6 +684,9 @@ class Scaffold:
         else:
             return exc_class, None
 
+    def ensure_sync(self, func):
+        raise NotImplementedError()
+
 
 def _endpoint_from_view_func(view_func):
     """Internal helper that returns the default endpoint for a given

tests/test_async.py

@@ -0,0 +1,139 @@
+import asyncio
+import sys
+
+import pytest
+
+from flask import Blueprint
+from flask import Flask
+from flask import request
+from flask.helpers import run_async
+
+pytest.importorskip("asgiref")
+
+
+class AppError(Exception):
+    pass
+
+
+class BlueprintError(Exception):
+    pass
+
+
+@pytest.fixture(name="async_app")
+def _async_app():
+    app = Flask(__name__)
+
+    @app.route("/", methods=["GET", "POST"])
+    async def index():
+        await asyncio.sleep(0)
+        return request.method
+
+    @app.errorhandler(AppError)
+    async def handle(_):
+        return "", 412
+
+    @app.route("/error")
+    async def error():
+        raise AppError()
+
+    blueprint = Blueprint("bp", __name__)
+
+    @blueprint.route("/", methods=["GET", "POST"])
+    async def bp_index():
+        await asyncio.sleep(0)
+        return request.method
+
+    @blueprint.errorhandler(BlueprintError)
+    async def bp_handle(_):
+        return "", 412
+
+    @blueprint.route("/error")
+    async def bp_error():
+        raise BlueprintError()
+
+    app.register_blueprint(blueprint, url_prefix="/bp")
+
+    return app
+
+
+@pytest.mark.skipif(sys.version_info < (3, 7), reason="requires Python >= 3.7")
+@pytest.mark.parametrize("path", ["/", "/bp/"])
+def test_async_route(path, async_app):
+    test_client = async_app.test_client()
+    response = test_client.get(path)
+    assert b"GET" in response.get_data()
+    response = test_client.post(path)
+    assert b"POST" in response.get_data()
+
+
+@pytest.mark.skipif(sys.version_info < (3, 7), reason="requires Python >= 3.7")
+@pytest.mark.parametrize("path", ["/error", "/bp/error"])
+def test_async_error_handler(path, async_app):
+    test_client = async_app.test_client()
+    response = test_client.get(path)
+    assert response.status_code == 412
+
+
+@pytest.mark.skipif(sys.version_info < (3, 7), reason="requires Python >= 3.7")
+def test_async_before_after_request():
+    app_first_called = False
+    app_before_called = False
+    app_after_called = False
+    bp_before_called = False
+    bp_after_called = False
+
+    app = Flask(__name__)
+
+    @app.route("/")
+    def index():
+        return ""
+
+    @app.before_first_request
+    async def before_first():
+        nonlocal app_first_called
+        app_first_called = True
+
+    @app.before_request
+    async def before():
+        nonlocal app_before_called
+        app_before_called = True
+
+    @app.after_request
+    async def after(response):
+        nonlocal app_after_called
+        app_after_called = True
+        return response
+
+    blueprint = Blueprint("bp", __name__)
+
+    @blueprint.route("/")
+    def bp_index():
+        return ""
+
+    @blueprint.before_request
+    async def bp_before():
+        nonlocal bp_before_called
+        bp_before_called = True
+
+    @blueprint.after_request
+    async def bp_after(response):
+        nonlocal bp_after_called
+        bp_after_called = True
+        return response
+
+    app.register_blueprint(blueprint, url_prefix="/bp")
+
+    test_client = app.test_client()
+    test_client.get("/")
+    assert app_first_called
+    assert app_before_called
+    assert app_after_called
+    test_client.get("/bp/")
+    assert bp_before_called
+    assert bp_after_called
+
+
+@pytest.mark.skipif(sys.version_info >= (3, 7), reason="should only raise Python < 3.7")
+def test_async_runtime_error():
+    with pytest.raises(RuntimeError):
+        run_async(None)

tox.ini

@@ -25,5 +25,8 @@ skip_install = true
 commands = pre-commit run --all-files --show-diff-on-failure
 
 [testenv:docs]
-deps = -r requirements/docs.txt
+deps =
+    -r requirements/docs.txt
+
+    https://github.com/pallets/werkzeug/archive/master.tar.gz
 commands = sphinx-build -W -b html -d {envtmpdir}/doctrees docs {envtmpdir}/html