Fixed #30451 -- Added ASGI handler and coroutine-safety.

This adds an ASGI handler, asgi.py file for the default project layout,
a few async utilities and adds async-safety to many parts of Django.

19 files changed, 477 insertions, 27 deletions

diff --git a/django/conf/project_template/project_name/asgi.py-tpl b/django/conf/project_template/project_name/asgi.py-tpl
new file mode 100644
index 0000000000..a827238196
--- /dev/null
+++ b/django/conf/project_template/project_name/asgi.py-tpl
@@ -0,0 +1,16 @@
+"""
+ASGI config for {{ project_name }} project.
+
+It exposes the ASGI callable as a module-level variable named ``application``.
+
+For more information on this file, see
+https://docs.djangoproject.com/en/{{ docs_version }}/howto/deployment/asgi/
+"""
+
+import os
+
+from django.core.asgi import get_asgi_application
+
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', '{{ project_name }}.settings')
+
+application = get_asgi_application()
diff --git a/django/contrib/staticfiles/handlers.py b/django/contrib/staticfiles/handlers.py
index b0aa66036d..711d8864ad 100644
--- a/django/contrib/staticfiles/handlers.py
+++ b/django/contrib/staticfiles/handlers.py
@@ -4,25 +4,20 @@ from urllib.request import url2pathname
 from django.conf import settings
 from django.contrib.staticfiles import utils
 from django.contrib.staticfiles.views import serve
+from django.core.handlers.asgi import ASGIHandler
 from django.core.handlers.exception import response_for_exception
 from django.core.handlers.wsgi import WSGIHandler, get_path_info
 from django.http import Http404
 
 
-class StaticFilesHandler(WSGIHandler):
+class StaticFilesHandlerMixin:
     """
-    WSGI middleware that intercepts calls to the static files directory, as
-    defined by the STATIC_URL setting, and serves those files.
+    Common methods used by WSGI and ASGI handlers.
     """
     # May be used to differentiate between handler types (e.g. in a
     # request_finished signal)
     handles_files = True
 
-    def __init__(self, application):
-        self.application = application
-        self.base_url = urlparse(self.get_base_url())
-        super().__init__()
-
     def load_middleware(self):
         # Middleware are already loaded for self.application; no need to reload
         # them for self.
@@ -57,7 +52,37 @@ class StaticFilesHandler(WSGIHandler):
         except Http404 as e:
             return response_for_exception(request, e)
 
+
+class StaticFilesHandler(StaticFilesHandlerMixin, WSGIHandler):
+    """
+    WSGI middleware that intercepts calls to the static files directory, as
+    defined by the STATIC_URL setting, and serves those files.
+    """
+    def __init__(self, application):
+        self.application = application
+        self.base_url = urlparse(self.get_base_url())
+        super().__init__()
+
     def __call__(self, environ, start_response):
         if not self._should_handle(get_path_info(environ)):
             return self.application(environ, start_response)
         return super().__call__(environ, start_response)
+
+
+class ASGIStaticFilesHandler(StaticFilesHandlerMixin, ASGIHandler):
+    """
+    ASGI application which wraps another and intercepts requests for static
+    files, passing them off to Django's static file serving.
+    """
+    def __init__(self, application):
+        self.application = application
+        self.base_url = urlparse(self.get_base_url())
+
+    async def __call__(self, scope, receive, send):
+        # Only even look at HTTP requests
+        if scope['type'] == 'http' and self._should_handle(scope['path']):
+            # Serve static content
+            # (the one thing super() doesn't do is __call__, apparently)
+            return await super().__call__(scope, receive, send)
+        # Hand off to the main app
+        return await self.application(scope, receive, send)
diff --git a/django/core/asgi.py b/django/core/asgi.py
new file mode 100644
index 0000000000..0d846ccd16
--- /dev/null
+++ b/django/core/asgi.py
@@ -0,0 +1,13 @@
+import django
+from django.core.handlers.asgi import ASGIHandler
+
+
+def get_asgi_application():
+    """
+    The public interface to Django's ASGI support. Return an ASGI 3 callable.
+
+    Avoids making django.core.handlers.ASGIHandler a public API, in case the
+    internal implementation changes or moves in the future.
+    """
+    django.setup(set_prefix=False)
+    return ASGIHandler()
diff --git a/django/core/exceptions.py b/django/core/exceptions.py
index 0e85397b9c..dc084b8692 100644
--- a/django/core/exceptions.py
+++ b/django/core/exceptions.py
@@ -63,6 +63,11 @@ class RequestDataTooBig(SuspiciousOperation):
     pass
 
 
+class RequestAborted(Exception):
+    """The request was closed before it was completed, or timed out."""
+    pass
+
+
 class PermissionDenied(Exception):
     """The user did not have permission to do that"""
     pass
@@ -181,3 +186,8 @@ class ValidationError(Exception):
 class EmptyResultSet(Exception):
     """A database query predicate is impossible."""
     pass
+
+
+class SynchronousOnlyOperation(Exception):
+    """The user tried to call a sync-only function from an async context."""
+    pass
diff --git a/django/core/handlers/asgi.py b/django/core/handlers/asgi.py
new file mode 100644
index 0000000000..021298e55d
--- /dev/null
+++ b/django/core/handlers/asgi.py
@@ -0,0 +1,297 @@
+import asyncio
+import logging
+import sys
+import tempfile
+import traceback
+from io import BytesIO
+
+from asgiref.sync import sync_to_async
+
+from django.conf import settings
+from django.core import signals
+from django.core.exceptions import RequestAborted, RequestDataTooBig
+from django.core.handlers import base
+from django.http import (
+    FileResponse, HttpRequest, HttpResponse, HttpResponseBadRequest,
+    HttpResponseServerError, QueryDict, parse_cookie,
+)
+from django.urls import set_script_prefix
+from django.utils.functional import cached_property
+
+logger = logging.getLogger('django.request')
+
+
+class ASGIRequest(HttpRequest):
+    """
+    Custom request subclass that decodes from an ASGI-standard request dict
+    and wraps request body handling.
+    """
+    # Number of seconds until a Request gives up on trying to read a request
+    # body and aborts.
+    body_receive_timeout = 60
+
+    def __init__(self, scope, body_file):
+        self.scope = scope
+        self._post_parse_error = False
+        self._read_started = False
+        self.resolver_match = None
+        self.script_name = self.scope.get('root_path', '')
+        if self.script_name and scope['path'].startswith(self.script_name):
+            # TODO: Better is-prefix checking, slash handling?
+            self.path_info = scope['path'][len(self.script_name):]
+        else:
+            self.path_info = scope['path']
+        # The Django path is different from ASGI scope path args, it should
+        # combine with script name.
+        if self.script_name:
+            self.path = '%s/%s' % (
+                self.script_name.rstrip('/'),
+                self.path_info.replace('/', '', 1),
+            )
+        else:
+            self.path = scope['path']
+        # HTTP basics.
+        self.method = self.scope['method'].upper()
+        # Ensure query string is encoded correctly.
+        query_string = self.scope.get('query_string', '')
+        if isinstance(query_string, bytes):
+            query_string = query_string.decode()
+        self.META = {
+            'REQUEST_METHOD': self.method,
+            'QUERY_STRING': query_string,
+            'SCRIPT_NAME': self.script_name,
+            'PATH_INFO': self.path_info,
+            # WSGI-expecting code will need these for a while
+            'wsgi.multithread': True,
+            'wsgi.multiprocess': True,
+        }
+        if self.scope.get('client'):
+            self.META['REMOTE_ADDR'] = self.scope['client'][0]
+            self.META['REMOTE_HOST'] = self.META['REMOTE_ADDR']
+            self.META['REMOTE_PORT'] = self.scope['client'][1]
+        if self.scope.get('server'):
+            self.META['SERVER_NAME'] = self.scope['server'][0]
+            self.META['SERVER_PORT'] = str(self.scope['server'][1])
+        else:
+            self.META['SERVER_NAME'] = 'unknown'
+            self.META['SERVER_PORT'] = '0'
+        # Headers go into META.
+        for name, value in self.scope.get('headers', []):
+            name = name.decode('latin1')
+            if name == 'content-length':
+                corrected_name = 'CONTENT_LENGTH'
+            elif name == 'content-type':
+                corrected_name = 'CONTENT_TYPE'
+            else:
+                corrected_name = 'HTTP_%s' % name.upper().replace('-', '_')
+            # HTTP/2 say only ASCII chars are allowed in headers, but decode
+            # latin1 just in case.
+            value = value.decode('latin1')
+            if corrected_name in self.META:
+                value = self.META[corrected_name] + ',' + value
+            self.META[corrected_name] = value
+        # Pull out request encoding, if provided.
+        self._set_content_type_params(self.META)
+        # Directly assign the body file to be our stream.
+        self._stream = body_file
+        # Other bits.
+        self.resolver_match = None
+
+    @cached_property
+    def GET(self):
+        return QueryDict(self.META['QUERY_STRING'])
+
+    def _get_scheme(self):
+        return self.scope.get('scheme') or super()._get_scheme()
+
+    def _get_post(self):
+        if not hasattr(self, '_post'):
+            self._load_post_and_files()
+        return self._post
+
+    def _set_post(self, post):
+        self._post = post
+
+    def _get_files(self):
+        if not hasattr(self, '_files'):
+            self._load_post_and_files()
+        return self._files
+
+    POST = property(_get_post, _set_post)
+    FILES = property(_get_files)
+
+    @cached_property
+    def COOKIES(self):
+        return parse_cookie(self.META.get('HTTP_COOKIE', ''))
+
+
+class ASGIHandler(base.BaseHandler):
+    """Handler for ASGI requests."""
+    request_class = ASGIRequest
+    # Size to chunk response bodies into for multiple response messages.
+    chunk_size = 2 ** 16
+
+    def __init__(self):
+        super(ASGIHandler, self).__init__()
+        self.load_middleware()
+
+    async def __call__(self, scope, receive, send):
+        """
+        Async entrypoint - parses the request and hands off to get_response.
+        """
+        # Serve only HTTP connections.
+        # FIXME: Allow to override this.
+        if scope['type'] != 'http':
+            raise ValueError(
+                'Django can only handle ASGI/HTTP connections, not %s'
+                % scope['type']
+            )
+        # Receive the HTTP request body as a stream object.
+        try:
+            body_file = await self.read_body(receive)
+        except RequestAborted:
+            return
+        # Request is complete and can be served.
+        set_script_prefix(self.get_script_prefix(scope))
+        await sync_to_async(signals.request_started.send)(sender=self.__class__, scope=scope)
+        # Get the request and check for basic issues.
+        request, error_response = self.create_request(scope, body_file)
+        if request is None:
+            await self.send_response(error_response, send)
+            return
+        # Get the response, using a threadpool via sync_to_async, if needed.
+        if asyncio.iscoroutinefunction(self.get_response):
+            response = await self.get_response(request)
+        else:
+            # If get_response is synchronous, run it non-blocking.
+            response = await sync_to_async(self.get_response)(request)
+        response._handler_class = self.__class__
+        # Increase chunk size on file responses (ASGI servers handles low-level
+        # chunking).
+        if isinstance(response, FileResponse):
+            response.block_size = self.chunk_size
+        # Send the response.
+        await self.send_response(response, send)
+
+    async def read_body(self, receive):
+        """Reads a HTTP body from an ASGI connection."""
+        # Use the tempfile that auto rolls-over to a disk file as it fills up,
+        # if a maximum in-memory size is set. Otherwise use a BytesIO object.
+        if settings.FILE_UPLOAD_MAX_MEMORY_SIZE is None:
+            body_file = BytesIO()
+        else:
+            body_file = tempfile.SpooledTemporaryFile(max_size=settings.FILE_UPLOAD_MAX_MEMORY_SIZE, mode='w+b')
+        while True:
+            message = await receive()
+            if message['type'] == 'http.disconnect':
+                # Early client disconnect.
+                raise RequestAborted()
+            # Add a body chunk from the message, if provided.
+            if 'body' in message:
+                body_file.write(message['body'])
+            # Quit out if that's the end.
+            if not message.get('more_body', False):
+                break
+        body_file.seek(0)
+        return body_file
+
+    def create_request(self, scope, body_file):
+        """
+        Create the Request object and returns either (request, None) or
+        (None, response) if there is an error response.
+        """
+        try:
+            return self.request_class(scope, body_file), None
+        except UnicodeDecodeError:
+            logger.warning(
+                'Bad Request (UnicodeDecodeError)',
+                exc_info=sys.exc_info(),
+                extra={'status_code': 400},
+            )
+            return None, HttpResponseBadRequest()
+        except RequestDataTooBig:
+            return None, HttpResponse('413 Payload too large', status=413)
+
+    def handle_uncaught_exception(self, request, resolver, exc_info):
+        """Last-chance handler for exceptions."""
+        # There's no WSGI server to catch the exception further up
+        # if this fails, so translate it into a plain text response.
+        try:
+            return super().handle_uncaught_exception(request, resolver, exc_info)
+        except Exception:
+            return HttpResponseServerError(
+                traceback.format_exc() if settings.DEBUG else 'Internal Server Error',
+                content_type='text/plain',
+            )
+
+    async def send_response(self, response, send):
+        """Encode and send a response out over ASGI."""
+        # Collect cookies into headers. Have to preserve header case as there
+        # are some non-RFC compliant clients that require e.g. Content-Type.
+        response_headers = []
+        for header, value in response.items():
+            if isinstance(header, str):
+                header = header.encode('ascii')
+            if isinstance(value, str):
+                value = value.encode('latin1')
+            response_headers.append((bytes(header), bytes(value)))
+        for c in response.cookies.values():
+            response_headers.append(
+                (b'Set-Cookie', c.output(header='').encode('ascii').strip())
+            )
+        # Initial response message.
+        await send({
+            'type': 'http.response.start',
+            'status': response.status_code,
+            'headers': response_headers,
+        })
+        # Streaming responses need to be pinned to their iterator.
+        if response.streaming:
+            # Access `__iter__` and not `streaming_content` directly in case
+            # it has been overridden in a subclass.
+            for part in response:
+                for chunk, _ in self.chunk_bytes(part):
+                    await send({
+                        'type': 'http.response.body',
+                        'body': chunk,
+                        # Ignore "more" as there may be more parts; instead,
+                        # use an empty final closing message with False.
+                        'more_body': True,
+                    })
+            # Final closing message.
+            await send({'type': 'http.response.body'})
+        # Other responses just need chunking.
+        else:
+            # Yield chunks of response.
+            for chunk, last in self.chunk_bytes(response.content):
+                await send({
+                    'type': 'http.response.body',
+                    'body': chunk,
+                    'more_body': not last,
+                })
+        response.close()
+
+    @classmethod
+    def chunk_bytes(cls, data):
+        """
+        Chunks some data up so it can be sent in reasonable size messages.
+        Yields (chunk, last_chunk) tuples.
+        """
+        position = 0
+        if not data:
+            yield data, True
+            return
+        while position < len(data):
+            yield (
+                data[position:position + cls.chunk_size],
+                (position + cls.chunk_size) >= len(data),
+            )
+            position += cls.chunk_size
+
+    def get_script_prefix(self, scope):
+        """
+        Return the script prefix to use from either the scope or a setting.
+        """
+        if settings.FORCE_SCRIPT_NAME:
+            return settings.FORCE_SCRIPT_NAME
+        return scope.get('root_path', '') or ''
diff --git a/django/core/signals.py b/django/core/signals.py
index 5d9618dd0c..c4288edeb5 100644
--- a/django/core/signals.py
+++ b/django/core/signals.py
@@ -1,6 +1,6 @@
 from django.dispatch import Signal
 
-request_started = Signal(providing_args=["environ"])
+request_started = Signal(providing_args=["environ", "scope"])
 request_finished = Signal()
 got_request_exception = Signal(providing_args=["request"])
 setting_changed = Signal(providing_args=["setting", "value", "enter"])
diff --git a/django/db/backends/base/base.py b/django/db/backends/base/base.py
index 057fe8ac43..6435f478dd 100644
--- a/django/db/backends/base/base.py
+++ b/django/db/backends/base/base.py
@@ -17,6 +17,7 @@ from django.db.backends.signals import connection_created
 from django.db.transaction import TransactionManagementError
 from django.db.utils import DatabaseError, DatabaseErrorWrapper
 from django.utils import timezone
+from django.utils.asyncio import async_unsafe
 from django.utils.functional import cached_property
 
 NO_DB_ALIAS = '__no_db__'
@@ -177,6 +178,7 @@ class BaseDatabaseWrapper:
 
     # ##### Backend-specific methods for creating connections #####
 
+    @async_unsafe
     def connect(self):
         """Connect to the database. Assume that the connection is closed."""
         # Check for invalid configurations.
@@ -210,6 +212,7 @@ class BaseDatabaseWrapper:
                     "Connection '%s' cannot set TIME_ZONE because its engine "
                     "handles time zones conversions natively." % self.alias)
 
+    @async_unsafe
     def ensure_connection(self):
         """Guarantee that a connection to the database is established."""
         if self.connection is None:
@@ -251,10 +254,12 @@ class BaseDatabaseWrapper:
 
     # ##### Generic wrappers for PEP-249 connection methods #####
 
+    @async_unsafe
     def cursor(self):
         """Create a cursor, opening a connection if necessary."""
         return self._cursor()
 
+    @async_unsafe
     def commit(self):
         """Commit a transaction and reset the dirty flag."""
         self.validate_thread_sharing()
@@ -264,6 +269,7 @@ class BaseDatabaseWrapper:
         self.errors_occurred = False
         self.run_commit_hooks_on_set_autocommit_on = True
 
+    @async_unsafe
     def rollback(self):
         """Roll back a transaction and reset the dirty flag."""
         self.validate_thread_sharing()
@@ -274,6 +280,7 @@ class BaseDatabaseWrapper:
         self.needs_rollback = False
         self.run_on_commit = []
 
+    @async_unsafe
     def close(self):
         """Close the connection to the database."""
         self.validate_thread_sharing()
@@ -313,6 +320,7 @@ class BaseDatabaseWrapper:
 
     # ##### Generic savepoint management methods #####
 
+    @async_unsafe
     def savepoint(self):
         """
         Create a savepoint inside the current transaction. Return an
@@ -333,6 +341,7 @@ class BaseDatabaseWrapper:
 
         return sid
 
+    @async_unsafe
     def savepoint_rollback(self, sid):
         """
         Roll back to a savepoint. Do nothing if savepoints are not supported.
@@ -348,6 +357,7 @@ class BaseDatabaseWrapper:
             (sids, func) for (sids, func) in self.run_on_commit if sid not in sids
         ]
 
+    @async_unsafe
     def savepoint_commit(self, sid):
         """
         Release a savepoint. Do nothing if savepoints are not supported.
@@ -358,6 +368,7 @@ class BaseDatabaseWrapper:
         self.validate_thread_sharing()
         self._savepoint_commit(sid)
 
+    @async_unsafe
     def clean_savepoints(self):
         """
         Reset the counter used to generate unique savepoint ids in this thread.
diff --git a/django/db/backends/mysql/base.py b/django/db/backends/mysql/base.py
index 6613a85b1b..9b88c5ac25 100644
--- a/django/db/backends/mysql/base.py
+++ b/django/db/backends/mysql/base.py
@@ -9,6 +9,7 @@ from django.core.exceptions import ImproperlyConfigured
 from django.db import utils
 from django.db.backends import utils as backend_utils
 from django.db.backends.base.base import BaseDatabaseWrapper
+from django.utils.asyncio import async_unsafe
 from django.utils.functional import cached_property
 
 try:
@@ -223,6 +224,7 @@ class DatabaseWrapper(BaseDatabaseWrapper):
         kwargs.update(options)
         return kwargs
 
+    @async_unsafe
     def get_new_connection(self, conn_params):
         return Database.connect(**conn_params)
 
@@ -242,6 +244,7 @@ class DatabaseWrapper(BaseDatabaseWrapper):
             with self.cursor() as cursor:
                 cursor.execute('; '.join(assignments))
 
+    @async_unsafe
     def create_cursor(self, name=None):
         cursor = self.connection.cursor()
         return CursorWrapper(cursor)
diff --git a/django/db/backends/oracle/base.py b/django/db/backends/oracle/base.py
index b19361b157..0fbe96ab9b 100644
--- a/django/db/backends/oracle/base.py
+++ b/django/db/backends/oracle/base.py
@@ -13,6 +13,7 @@ from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
 from django.db import utils
 from django.db.backends.base.base import BaseDatabaseWrapper
+from django.utils.asyncio import async_unsafe
 from django.utils.encoding import force_bytes, force_str
 from django.utils.functional import cached_property
 
@@ -221,6 +222,7 @@ class DatabaseWrapper(BaseDatabaseWrapper):
             del conn_params['use_returning_into']
         return conn_params
 
+    @async_unsafe
     def get_new_connection(self, conn_params):
         return Database.connect(
             user=self.settings_dict['USER'],
@@ -269,6 +271,7 @@ class DatabaseWrapper(BaseDatabaseWrapper):
         if not self.get_autocommit():
             self.commit()
 
+    @async_unsafe
     def create_cursor(self, name=None):
         return FormatStylePlaceholderCursor(self.connection)
 
diff --git a/django/db/backends/postgresql/base.py b/django/db/backends/postgresql/base.py
index 6f8e06fe23..7e34a3a177 100644
--- a/django/db/backends/postgresql/base.py
+++ b/django/db/backends/postgresql/base.py
@@ -4,6 +4,7 @@ PostgreSQL database backend for Django.
 Requires psycopg 2: http://initd.org/projects/psycopg2
 """
 
+import asyncio
 import threading
 import warnings
 
@@ -15,6 +16,7 @@ from django.db.backends.utils import (
     CursorDebugWrapper as BaseCursorDebugWrapper,
 )
 from django.db.utils import DatabaseError as WrappedDatabaseError
+from django.utils.asyncio import async_unsafe
 from django.utils.functional import cached_property
 from django.utils.safestring import SafeString
 from django.utils.version import get_version_tuple
@@ -177,6 +179,7 @@ class DatabaseWrapper(BaseDatabaseWrapper):
             conn_params['port'] = settings_dict['PORT']
         return conn_params
 
+    @async_unsafe
     def get_new_connection(self, conn_params):
         connection = Database.connect(**conn_params)
 
@@ -217,6 +220,7 @@ class DatabaseWrapper(BaseDatabaseWrapper):
             if not self.get_autocommit():
                 self.connection.commit()
 
+    @async_unsafe
     def create_cursor(self, name=None):
         if name:
             # In autocommit mode, the cursor will be used outside of a
@@ -227,12 +231,34 @@ class DatabaseWrapper(BaseDatabaseWrapper):
         cursor.tzinfo_factory = utc_tzinfo_factory if settings.USE_TZ else None
         return cursor
 
+    @async_unsafe
     def chunked_cursor(self):
         self._named_cursor_idx += 1
+        # Get the current async task
+        # Note that right now this is behind @async_unsafe, so this is
+        # unreachable, but in future we'll start loosening this restriction.
+        # For now, it's here so that every use of "threading" is
+        # also async-compatible.
+        try:
+            if hasattr(asyncio, 'current_task'):
+                # Python 3.7 and up
+                current_task = asyncio.current_task()
+            else:
+                # Python 3.6
+                current_task = asyncio.Task.current_task()
+        except RuntimeError:
+            current_task = None
+        # Current task can be none even if the current_task call didn't error
+        if current_task:
+            task_ident = str(id(current_task))
+        else:
+            task_ident = 'sync'
+        # Use that and the thread ident to get a unique name
         return self._cursor(
-            name='_django_curs_%d_%d' % (
-                # Avoid reusing name in other threads
+            name='_django_curs_%d_%s_%d' % (
+                # Avoid reusing name in other threads / tasks
                 threading.current_thread().ident,
+                task_ident,
                 self._named_cursor_idx,
             )
         )
diff --git a/django/db/backends/sqlite3/base.py b/django/db/backends/sqlite3/base.py
index f4184fce05..fff65197f9 100644
--- a/django/db/backends/sqlite3/base.py
+++ b/django/db/backends/sqlite3/base.py
@@ -20,6 +20,7 @@ from django.db import utils
 from django.db.backends import utils as backend_utils
 from django.db.backends.base.base import BaseDatabaseWrapper
 from django.utils import timezone
+from django.utils.asyncio import async_unsafe
 from django.utils.dateparse import parse_datetime, parse_time
 from django.utils.duration import duration_microseconds
 
@@ -191,6 +192,7 @@ class DatabaseWrapper(BaseDatabaseWrapper):
         kwargs.update({'check_same_thread': False, 'uri': True})
         return kwargs
 
+    @async_unsafe
     def get_new_connection(self, conn_params):
         conn = Database.connect(**conn_params)
         conn.create_function("django_date_extract", 2, _sqlite_datetime_extract)
@@ -248,6 +250,7 @@ class DatabaseWrapper(BaseDatabaseWrapper):
     def create_cursor(self, name=None):
         return self.connection.cursor(factory=SQLiteCursorWrapper)
 
+    @async_unsafe
     def close(self):
         self.validate_thread_sharing()
         # If database is in memory, closing the connection destroys the
diff --git a/django/db/utils.py b/django/db/utils.py
index cb7f3d0f0b..4bd119227f 100644
--- a/django/db/utils.py
+++ b/django/db/utils.py
@@ -1,7 +1,8 @@
 import pkgutil
 from importlib import import_module
 from pathlib import Path
-from threading import local
+
+from asgiref.local import Local
 
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
@@ -139,7 +140,12 @@ class ConnectionHandler:
         like settings.DATABASES).
         """
         self._databases = databases
-        self._connections = local()
+        # Connections needs to still be an actual thread local, as it's truly
+        # thread-critical. Database backends should use @async_unsafe to protect
+        # their code from async contexts, but this will give those contexts
+        # separate connections in case it's needed as well. There's no cleanup
+        # after async contexts, though, so we don't allow that if we can help it.
+        self._connections = Local(thread_critical=True)
 
     @cached_property
     def databases(self):
diff --git a/django/test/signals.py b/django/test/signals.py
index a623e756ce..31a5017602 100644
--- a/django/test/signals.py
+++ b/django/test/signals.py
@@ -1,8 +1,9 @@
 import os
-import threading
 import time
 import warnings
 
+from asgiref.local import Local
+
 from django.apps import apps
 from django.core.exceptions import ImproperlyConfigured
 from django.core.signals import setting_changed
@@ -26,7 +27,7 @@ COMPLEX_OVERRIDE_SETTINGS = {'DATABASES'}
 def clear_cache_handlers(**kwargs):
     if kwargs['setting'] == 'CACHES':
         from django.core.cache import caches
-        caches._caches = threading.local()
+        caches._caches = Local()
 
 
 @receiver(setting_changed)
@@ -113,7 +114,7 @@ def language_changed(**kwargs):
     if kwargs['setting'] in {'LANGUAGES', 'LANGUAGE_CODE', 'LOCALE_PATHS'}:
         from django.utils.translation import trans_real
         trans_real._default = None
-        trans_real._active = threading.local()
+        trans_real._active = Local()
     if kwargs['setting'] in {'LANGUAGES', 'LOCALE_PATHS'}:
         from django.utils.translation import trans_real
         trans_real._translations = {}
diff --git a/django/urls/base.py b/django/urls/base.py
index 1200d9a25b..0e1c3d909c 100644
--- a/django/urls/base.py
+++ b/django/urls/base.py
@@ -1,6 +1,7 @@
-from threading import local
 from urllib.parse import urlsplit, urlunsplit
 
+from asgiref.local import Local
+
 from django.utils.encoding import iri_to_uri
 from django.utils.functional import lazy
 from django.utils.translation import override
@@ -12,10 +13,10 @@ from .utils import get_callable
 # SCRIPT_NAME prefixes for each thread are stored here. If there's no entry for
 # the current thread (which is the only one we ever access), it is assumed to
 # be empty.
-_prefixes = local()
+_prefixes = Local()
 
 # Overridden URLconfs for each thread are stored here.
-_urlconfs = local()
+_urlconfs = Local()
 
 
 def resolve(path, urlconf=None):
diff --git a/django/urls/resolvers.py b/django/urls/resolvers.py
index 9d3379a821..af0508f94e 100644
--- a/django/urls/resolvers.py
+++ b/django/urls/resolvers.py
@@ -8,10 +8,11 @@ attributes of the resolved URL match.
 import functools
 import inspect
 import re
-import threading
 from importlib import import_module
 from urllib.parse import quote
 
+from asgiref.local import Local
+
 from django.conf import settings
 from django.core.checks import Error, Warning
 from django.core.checks.urls import check_resolver
@@ -380,7 +381,7 @@ class URLResolver:
         # urlpatterns
         self._callback_strs = set()
         self._populated = False
-        self._local = threading.local()
+        self._local = Local()
 
     def __repr__(self):
         if isinstance(self.urlconf_name, list) and self.urlconf_name:
diff --git a/django/utils/asyncio.py b/django/utils/asyncio.py
new file mode 100644
index 0000000000..c4de04ba12
--- /dev/null
+++ b/django/utils/asyncio.py
@@ -0,0 +1,32 @@
+import asyncio
+import functools
+
+from django.core.exceptions import SynchronousOnlyOperation
+
+
+def async_unsafe(message):
+    """
+    Decorator to mark functions as async-unsafe. Someone trying to access
+    the function while in an async context will get an error message.
+    """
+    def decorator(func):
+        @functools.wraps(func)
+        def inner(*args, **kwargs):
+            # Detect a running event loop in this thread.
+            try:
+                event_loop = asyncio.get_event_loop()
+            except RuntimeError:
+                pass
+            else:
+                if event_loop.is_running():
+                    raise SynchronousOnlyOperation(message)
+            # Pass onwards.
+            return func(*args, **kwargs)
+        return inner
+    # If the message is actually a function, then be a no-arguments decorator.
+    if callable(message):
+        func = message
+        message = 'You cannot call this from an async context - use a thread or sync_to_async.'
+        return decorator(func)
+    else:
+        return decorator
diff --git a/django/utils/timezone.py b/django/utils/timezone.py
index 58e92c1fa8..4c43377447 100644
--- a/django/utils/timezone.py
+++ b/django/utils/timezone.py
@@ -6,9 +6,9 @@ import functools
 import warnings
 from contextlib import ContextDecorator
 from datetime import datetime, timedelta, timezone, tzinfo
-from threading import local
 
 import pytz
+from asgiref.local import Local
 
 from django.conf import settings
 from django.utils.deprecation import RemovedInDjango31Warning
@@ -89,7 +89,7 @@ def get_default_timezone_name():
     return _get_timezone_name(get_default_timezone())
 
 
-_active = local()
+_active = Local()
 
 
 def get_current_timezone():
diff --git a/django/utils/translation/reloader.py b/django/utils/translation/reloader.py
index 8e2d320208..2d69ad44e0 100644
--- a/django/utils/translation/reloader.py
+++ b/django/utils/translation/reloader.py
@@ -1,6 +1,7 @@
-import threading
 from pathlib import Path
 
+from asgiref.local import Local
+
 from django.apps import apps
 
 
@@ -25,5 +26,5 @@ def translation_file_changed(sender, file_path, **kwargs):
         gettext._translations = {}
         trans_real._translations = {}
         trans_real._default = None
-        trans_real._active = threading.local()
+        trans_real._active = Local()
         return True
diff --git a/django/utils/translation/trans_real.py b/django/utils/translation/trans_real.py
index f4985fb3c1..e089597ccb 100644
--- a/django/utils/translation/trans_real.py
+++ b/django/utils/translation/trans_real.py
@@ -5,7 +5,8 @@ import os
 import re
 import sys
 import warnings
-from threading import local
+
+from asgiref.local import Local
 
 from django.apps import apps
 from django.conf import settings
@@ -20,7 +21,7 @@ from . import to_language, to_locale
 # Translations are cached in a dictionary for every language.
 # The active translations are stored by threadid to make them thread local.
 _translations = {}
-_active = local()
+_active = Local()
 
 # The default translation is based on the settings file.
 _default = None