diff --git a/django/contrib/auth/__init__.py b/django/contrib/auth/__init__.py
index 91257dd3da56..3db1445d9ea9 100644
--- a/django/contrib/auth/__init__.py
+++ b/django/contrib/auth/__init__.py
@@ -269,4 +269,6 @@ def update_session_auth_hash(request, user):
 
 async def aupdate_session_auth_hash(request, user):
     """See update_session_auth_hash()."""
-    return await sync_to_async(update_session_auth_hash)(request, user)
+    await request.session.acycle_key()
+    if hasattr(user, "get_session_auth_hash") and request.user == user:
+        await request.session.aset(HASH_SESSION_KEY, user.get_session_auth_hash())
diff --git a/django/contrib/sessions/backends/base.py b/django/contrib/sessions/backends/base.py
index 019ac0f4c67d..69f756a22876 100644
--- a/django/contrib/sessions/backends/base.py
+++ b/django/contrib/sessions/backends/base.py
@@ -2,6 +2,8 @@
 import string
 from datetime import datetime, timedelta
 
+from asgiref.sync import sync_to_async
+
 from django.conf import settings
 from django.core import signing
 from django.utils import timezone
@@ -56,6 +58,10 @@ def __setitem__(self, key, value):
         self._session[key] = value
         self.modified = True
 
+    async def aset(self, key, value):
+        (await self._aget_session())[key] = value
+        self.modified = True
+
     def __delitem__(self, key):
         del self._session[key]
         self.modified = True
@@ -67,11 +73,19 @@ def key_salt(self):
     def get(self, key, default=None):
         return self._session.get(key, default)
 
+    async def aget(self, key, default=None):
+        return (await self._aget_session()).get(key, default)
+
     def pop(self, key, default=__not_given):
         self.modified = self.modified or key in self._session
         args = () if default is self.__not_given else (default,)
         return self._session.pop(key, *args)
 
+    async def apop(self, key, default=__not_given):
+        self.modified = self.modified or key in (await self._aget_session())
+        args = () if default is self.__not_given else (default,)
+        return (await self._aget_session()).pop(key, *args)
+
     def setdefault(self, key, value):
         if key in self._session:
             return self._session[key]
@@ -79,15 +93,32 @@ def setdefault(self, key, value):
             self[key] = value
             return value
 
+    async def asetdefault(self, key, value):
+        session = await self._aget_session()
+        if key in session:
+            return session[key]
+        else:
+            await self.aset(key, value)
+            return value
+
     def set_test_cookie(self):
         self[self.TEST_COOKIE_NAME] = self.TEST_COOKIE_VALUE
 
+    async def aset_test_cookie(self):
+        await self.aset(self.TEST_COOKIE_NAME, self.TEST_COOKIE_VALUE)
+
     def test_cookie_worked(self):
         return self.get(self.TEST_COOKIE_NAME) == self.TEST_COOKIE_VALUE
 
+    async def atest_cookie_worked(self):
+        return (await self.aget(self.TEST_COOKIE_NAME)) == self.TEST_COOKIE_VALUE
+
     def delete_test_cookie(self):
         del self[self.TEST_COOKIE_NAME]
 
+    async def adelete_test_cookie(self):
+        del (await self._aget_session())[self.TEST_COOKIE_NAME]
+
     def encode(self, session_dict):
         "Return the given session dictionary serialized and encoded as a string."
         return signing.dumps(
@@ -115,18 +146,34 @@ def update(self, dict_):
         self._session.update(dict_)
         self.modified = True
 
+    async def aupdate(self, dict_):
+        (await self._aget_session()).update(dict_)
+        self.modified = True
+
     def has_key(self, key):
         return key in self._session
 
+    async def ahas_key(self, key):
+        return key in (await self._aget_session())
+
     def keys(self):
         return self._session.keys()
 
+    async def akeys(self):
+        return (await self._aget_session()).keys()
+
     def values(self):
         return self._session.values()
 
+    async def avalues(self):
+        return (await self._aget_session()).values()
+
     def items(self):
         return self._session.items()
 
+    async def aitems(self):
+        return (await self._aget_session()).items()
+
     def clear(self):
         # To avoid unnecessary persistent storage accesses, we set up the
         # internals directly (loading data wastes time, since we are going to
@@ -149,11 +196,22 @@ def _get_new_session_key(self):
             if not self.exists(session_key):
                 return session_key
 
+    async def _aget_new_session_key(self):
+        while True:
+            session_key = get_random_string(32, VALID_KEY_CHARS)
+            if not await self.aexists(session_key):
+                return session_key
+
     def _get_or_create_session_key(self):
         if self._session_key is None:
             self._session_key = self._get_new_session_key()
         return self._session_key
 
+    async def _aget_or_create_session_key(self):
+        if self._session_key is None:
+            self._session_key = await self._aget_new_session_key()
+        return self._session_key
+
     def _validate_session_key(self, key):
         """
         Key must be truthy and at least 8 characters long. 8 characters is an
@@ -191,6 +249,17 @@ def _get_session(self, no_load=False):
                 self._session_cache = self.load()
         return self._session_cache
 
+    async def _aget_session(self, no_load=False):
+        self.accessed = True
+        try:
+            return self._session_cache
+        except AttributeError:
+            if self.session_key is None or no_load:
+                self._session_cache = {}
+            else:
+                self._session_cache = await self.aload()
+        return self._session_cache
+
     _session = property(_get_session)
 
     def get_session_cookie_age(self):
@@ -223,6 +292,25 @@ def get_expiry_age(self, **kwargs):
         delta = expiry - modification
         return delta.days * 86400 + delta.seconds
 
+    async def aget_expiry_age(self, **kwargs):
+        try:
+            modification = kwargs["modification"]
+        except KeyError:
+            modification = timezone.now()
+        try:
+            expiry = kwargs["expiry"]
+        except KeyError:
+            expiry = await self.aget("_session_expiry")
+
+        if not expiry:  # Checks both None and 0 cases
+            return self.get_session_cookie_age()
+        if not isinstance(expiry, (datetime, str)):
+            return expiry
+        if isinstance(expiry, str):
+            expiry = datetime.fromisoformat(expiry)
+        delta = expiry - modification
+        return delta.days * 86400 + delta.seconds
+
     def get_expiry_date(self, **kwargs):
         """Get session the expiry date (as a datetime object).
 
@@ -246,6 +334,23 @@ def get_expiry_date(self, **kwargs):
         expiry = expiry or self.get_session_cookie_age()
         return modification + timedelta(seconds=expiry)
 
+    async def aget_expiry_date(self, **kwargs):
+        try:
+            modification = kwargs["modification"]
+        except KeyError:
+            modification = timezone.now()
+        try:
+            expiry = kwargs["expiry"]
+        except KeyError:
+            expiry = await self.aget("_session_expiry")
+
+        if isinstance(expiry, datetime):
+            return expiry
+        elif isinstance(expiry, str):
+            return datetime.fromisoformat(expiry)
+        expiry = expiry or self.get_session_cookie_age()
+        return modification + timedelta(seconds=expiry)
+
     def set_expiry(self, value):
         """
         Set a custom expiration for the session. ``value`` can be an integer,
@@ -274,6 +379,20 @@ def set_expiry(self, value):
             value = value.isoformat()
         self["_session_expiry"] = value
 
+    async def aset_expiry(self, value):
+        if value is None:
+            # Remove any custom expiration for this session.
+            try:
+                await self.apop("_session_expiry")
+            except KeyError:
+                pass
+            return
+        if isinstance(value, timedelta):
+            value = timezone.now() + value
+        if isinstance(value, datetime):
+            value = value.isoformat()
+        await self.aset("_session_expiry", value)
+
     def get_expire_at_browser_close(self):
         """
         Return ``True`` if the session is set to expire when the browser
@@ -285,6 +404,11 @@ def get_expire_at_browser_close(self):
             return settings.SESSION_EXPIRE_AT_BROWSER_CLOSE
         return expiry == 0
 
+    async def aget_expire_at_browser_close(self):
+        if (expiry := await self.aget("_session_expiry")) is None:
+            return settings.SESSION_EXPIRE_AT_BROWSER_CLOSE
+        return expiry == 0
+
     def flush(self):
         """
         Remove the current session data from the database and regenerate the
@@ -294,6 +418,11 @@ def flush(self):
         self.delete()
         self._session_key = None
 
+    async def aflush(self):
+        self.clear()
+        await self.adelete()
+        self._session_key = None
+
     def cycle_key(self):
         """
         Create a new session key, while retaining the current session data.
@@ -305,6 +434,17 @@ def cycle_key(self):
         if key:
             self.delete(key)
 
+    async def acycle_key(self):
+        """
+        Create a new session key, while retaining the current session data.
+        """
+        data = await self._aget_session()
+        key = self.session_key
+        await self.acreate()
+        self._session_cache = data
+        if key:
+            await self.adelete(key)
+
     # Methods that child classes must implement.
 
     def exists(self, session_key):
@@ -315,6 +455,9 @@ def exists(self, session_key):
             "subclasses of SessionBase must provide an exists() method"
         )
 
+    async def aexists(self, session_key):
+        return await sync_to_async(self.exists)(session_key)
+
     def create(self):
         """
         Create a new session instance. Guaranteed to create a new object with
@@ -325,6 +468,9 @@ def create(self):
             "subclasses of SessionBase must provide a create() method"
         )
 
+    async def acreate(self):
+        return await sync_to_async(self.create)()
+
     def save(self, must_create=False):
         """
         Save the session data. If 'must_create' is True, create a new session
@@ -335,6 +481,9 @@ def save(self, must_create=False):
             "subclasses of SessionBase must provide a save() method"
         )
 
+    async def asave(self, must_create=False):
+        return await sync_to_async(self.save)(must_create)
+
     def delete(self, session_key=None):
         """
         Delete the session data under this key. If the key is None, use the
@@ -344,6 +493,9 @@ def delete(self, session_key=None):
             "subclasses of SessionBase must provide a delete() method"
         )
 
+    async def adelete(self, session_key=None):
+        return await sync_to_async(self.delete)(session_key)
+
     def load(self):
         """
         Load the session data and return a dictionary.
@@ -352,6 +504,9 @@ def load(self):
             "subclasses of SessionBase must provide a load() method"
         )
 
+    async def aload(self):
+        return await sync_to_async(self.load)()
+
     @classmethod
     def clear_expired(cls):
         """
@@ -362,3 +517,7 @@ def clear_expired(cls):
         a built-in expiration mechanism, it should be a no-op.
         """
         raise NotImplementedError("This backend does not support clear_expired().")
+
+    @classmethod
+    async def aclear_expired(cls):
+        return await sync_to_async(cls.clear_expired)()
diff --git a/django/contrib/sessions/backends/cache.py b/django/contrib/sessions/backends/cache.py
index 0c9d244f5618..b87afd6f666b 100644
--- a/django/contrib/sessions/backends/cache.py
+++ b/django/contrib/sessions/backends/cache.py
@@ -20,6 +20,9 @@ def __init__(self, session_key=None):
     def cache_key(self):
         return self.cache_key_prefix + self._get_or_create_session_key()
 
+    async def acache_key(self):
+        return self.cache_key_prefix + await self._aget_or_create_session_key()
+
     def load(self):
         try:
             session_data = self._cache.get(self.cache_key)
@@ -32,6 +35,16 @@ def load(self):
         self._session_key = None
         return {}
 
+    async def aload(self):
+        try:
+            session_data = await self._cache.aget(await self.acache_key())
+        except Exception:
+            session_data = None
+        if session_data is not None:
+            return session_data
+        self._session_key = None
+        return {}
+
     def create(self):
         # Because a cache can fail silently (e.g. memcache), we don't know if
         # we are failing to create a new session because of a key collision or
@@ -51,6 +64,20 @@ def create(self):
             "It is likely that the cache is unavailable."
         )
 
+    async def acreate(self):
+        for i in range(10000):
+            self._session_key = await self._aget_new_session_key()
+            try:
+                await self.asave(must_create=True)
+            except CreateError:
+                continue
+            self.modified = True
+            return
+        raise RuntimeError(
+            "Unable to create a new session key. "
+            "It is likely that the cache is unavailable."
+        )
+
     def save(self, must_create=False):
         if self.session_key is None:
             return self.create()
@@ -68,11 +95,33 @@ def save(self, must_create=False):
         if must_create and not result:
             raise CreateError
 
+    async def asave(self, must_create=False):
+        if self.session_key is None:
+            return await self.acreate()
+        if must_create:
+            func = self._cache.aadd
+        elif await self._cache.aget(await self.acache_key()) is not None:
+            func = self._cache.aset
+        else:
+            raise UpdateError
+        result = await func(
+            await self.acache_key(),
+            await self._aget_session(no_load=must_create),
+            await self.aget_expiry_age(),
+        )
+        if must_create and not result:
+            raise CreateError
+
     def exists(self, session_key):
         return (
             bool(session_key) and (self.cache_key_prefix + session_key) in self._cache
         )
 
+    async def aexists(self, session_key):
+        return bool(session_key) and await self._cache.ahas_key(
+            self.cache_key_prefix + session_key
+        )
+
     def delete(self, session_key=None):
         if session_key is None:
             if self.session_key is None:
@@ -80,6 +129,17 @@ def delete(self, session_key=None):
             session_key = self.session_key
         self._cache.delete(self.cache_key_prefix + session_key)
 
+    async def adelete(self, session_key=None):
+        if session_key is None:
+            if self.session_key is None:
+                return
+            session_key = self.session_key
+        await self._cache.adelete(self.cache_key_prefix + session_key)
+
     @classmethod
     def clear_expired(cls):
         pass
+
+    @classmethod
+    async def aclear_expired(cls):
+        pass
diff --git a/django/contrib/sessions/backends/cached_db.py b/django/contrib/sessions/backends/cached_db.py
index a2a8cf47afd4..2195f57acccd 100644
--- a/django/contrib/sessions/backends/cached_db.py
+++ b/django/contrib/sessions/backends/cached_db.py
@@ -28,6 +28,9 @@ def __init__(self, session_key=None):
     def cache_key(self):
         return self.cache_key_prefix + self._get_or_create_session_key()
 
+    async def acache_key(self):
+        return self.cache_key_prefix + await self._aget_or_create_session_key()
+
     def load(self):
         try:
             data = self._cache.get(self.cache_key)
@@ -47,6 +50,27 @@ def load(self):
                 data = {}
         return data
 
+    async def aload(self):
+        try:
+            data = await self._cache.aget(await self.acache_key())
+        except Exception:
+            # Some backends (e.g. memcache) raise an exception on invalid
+            # cache keys. If this happens, reset the session. See #17810.
+            data = None
+
+        if data is None:
+            s = await self._aget_session_from_db()
+            if s:
+                data = self.decode(s.session_data)
+                await self._cache.aset(
+                    await self.acache_key(),
+                    data,
+                    await self.aget_expiry_age(expiry=s.expire_date),
+                )
+            else:
+                data = {}
+        return data
+
     def exists(self, session_key):
         return (
             session_key
@@ -54,6 +78,13 @@ def exists(self, session_key):
             or super().exists(session_key)
         )
 
+    async def aexists(self, session_key):
+        return (
+            session_key
+            and (self.cache_key_prefix + session_key) in self._cache
+            or await super().aexists(session_key)
+        )
+
     def save(self, must_create=False):
         super().save(must_create)
         try:
@@ -61,6 +92,17 @@ def save(self, must_create=False):
         except Exception:
             logger.exception("Error saving to cache (%s)", self._cache)
 
+    async def asave(self, must_create=False):
+        await super().asave(must_create)
+        try:
+            await self._cache.aset(
+                await self.acache_key(),
+                self._session,
+                await self.aget_expiry_age(),
+            )
+        except Exception:
+            logger.exception("Error saving to cache (%s)", self._cache)
+
     def delete(self, session_key=None):
         super().delete(session_key)
         if session_key is None:
@@ -69,6 +111,14 @@ def delete(self, session_key=None):
             session_key = self.session_key
         self._cache.delete(self.cache_key_prefix + session_key)
 
+    async def adelete(self, session_key=None):
+        await super().adelete(session_key)
+        if session_key is None:
+            if self.session_key is None:
+                return
+            session_key = self.session_key
+        await self._cache.adelete(self.cache_key_prefix + session_key)
+
     def flush(self):
         """
         Remove the current session data from the database and regenerate the
@@ -77,3 +127,9 @@ def flush(self):
         self.clear()
         self.delete(self.session_key)
         self._session_key = None
+
+    async def aflush(self):
+        """See flush()."""
+        self.clear()
+        await self.adelete(self.session_key)
+        self._session_key = None
diff --git a/django/contrib/sessions/backends/db.py b/django/contrib/sessions/backends/db.py
index e1f6b69c5582..6d6247d6c94b 100644
--- a/django/contrib/sessions/backends/db.py
+++ b/django/contrib/sessions/backends/db.py
@@ -1,5 +1,7 @@
 import logging
 
+from asgiref.sync import sync_to_async
+
 from django.contrib.sessions.backends.base import CreateError, SessionBase, UpdateError
 from django.core.exceptions import SuspiciousOperation
 from django.db import DatabaseError, IntegrityError, router, transaction
@@ -38,13 +40,31 @@ def _get_session_from_db(self):
                 logger.warning(str(e))
             self._session_key = None
 
+    async def _aget_session_from_db(self):
+        try:
+            return await self.model.objects.aget(
+                session_key=self.session_key, expire_date__gt=timezone.now()
+            )
+        except (self.model.DoesNotExist, SuspiciousOperation) as e:
+            if isinstance(e, SuspiciousOperation):
+                logger = logging.getLogger("django.security.%s" % e.__class__.__name__)
+                logger.warning(str(e))
+            self._session_key = None
+
     def load(self):
         s = self._get_session_from_db()
         return self.decode(s.session_data) if s else {}
 
+    async def aload(self):
+        s = await self._aget_session_from_db()
+        return self.decode(s.session_data) if s else {}
+
     def exists(self, session_key):
         return self.model.objects.filter(session_key=session_key).exists()
 
+    async def aexists(self, session_key):
+        return await self.model.objects.filter(session_key=session_key).aexists()
+
     def create(self):
         while True:
             self._session_key = self._get_new_session_key()
@@ -58,6 +78,19 @@ def create(self):
             self.modified = True
             return
 
+    async def acreate(self):
+        while True:
+            self._session_key = await self._aget_new_session_key()
+            try:
+                # Save immediately to ensure we have a unique entry in the
+                # database.
+                await self.asave(must_create=True)
+            except CreateError:
+                # Key wasn't unique. Try again.
+                continue
+            self.modified = True
+            return
+
     def create_model_instance(self, data):
         """
         Return a new instance of the session model object, which represents the
@@ -70,6 +103,14 @@ def create_model_instance(self, data):
             expire_date=self.get_expiry_date(),
         )
 
+    async def acreate_model_instance(self, data):
+        """See create_model_instance()."""
+        return self.model(
+            session_key=await self._aget_or_create_session_key(),
+            session_data=self.encode(data),
+            expire_date=await self.aget_expiry_date(),
+        )
+
     def save(self, must_create=False):
         """
         Save the current session data to the database. If 'must_create' is
@@ -95,6 +136,36 @@ def save(self, must_create=False):
                 raise UpdateError
             raise
 
+    async def asave(self, must_create=False):
+        """See save()."""
+        if self.session_key is None:
+            return await self.acreate()
+        data = await self._aget_session(no_load=must_create)
+        obj = await self.acreate_model_instance(data)
+        using = router.db_for_write(self.model, instance=obj)
+        try:
+            # This code MOST run in a transaction, so it requires
+            # @sync_to_async wrapping until transaction.atomic() supports
+            # async.
+            @sync_to_async
+            def sync_transaction():
+                with transaction.atomic(using=using):
+                    obj.save(
+                        force_insert=must_create,
+                        force_update=not must_create,
+                        using=using,
+                    )
+
+            await sync_transaction()
+        except IntegrityError:
+            if must_create:
+                raise CreateError
+            raise
+        except DatabaseError:
+            if not must_create:
+                raise UpdateError
+            raise
+
     def delete(self, session_key=None):
         if session_key is None:
             if self.session_key is None:
@@ -105,6 +176,23 @@ def delete(self, session_key=None):
         except self.model.DoesNotExist:
             pass
 
+    async def adelete(self, session_key=None):
+        if session_key is None:
+            if self.session_key is None:
+                return
+            session_key = self.session_key
+        try:
+            obj = await self.model.objects.aget(session_key=session_key)
+            await obj.adelete()
+        except self.model.DoesNotExist:
+            pass
+
     @classmethod
     def clear_expired(cls):
         cls.get_model_class().objects.filter(expire_date__lt=timezone.now()).delete()
+
+    @classmethod
+    async def aclear_expired(cls):
+        await cls.get_model_class().objects.filter(
+            expire_date__lt=timezone.now()
+        ).adelete()
diff --git a/django/contrib/sessions/backends/file.py b/django/contrib/sessions/backends/file.py
index 6f54690a9c57..e15b3f0141ec 100644
--- a/django/contrib/sessions/backends/file.py
+++ b/django/contrib/sessions/backends/file.py
@@ -104,6 +104,9 @@ def load(self):
             self._session_key = None
         return session_data
 
+    async def aload(self):
+        return self.load()
+
     def create(self):
         while True:
             self._session_key = self._get_new_session_key()
@@ -114,6 +117,9 @@ def create(self):
             self.modified = True
             return
 
+    async def acreate(self):
+        return self.create()
+
     def save(self, must_create=False):
         if self.session_key is None:
             return self.create()
@@ -177,9 +183,15 @@ def save(self, must_create=False):
         except (EOFError, OSError):
             pass
 
+    async def asave(self, must_create=False):
+        return self.save(must_create=must_create)
+
     def exists(self, session_key):
         return os.path.exists(self._key_to_file(session_key))
 
+    async def aexists(self, session_key):
+        return self.exists(session_key)
+
     def delete(self, session_key=None):
         if session_key is None:
             if self.session_key is None:
@@ -190,6 +202,9 @@ def delete(self, session_key=None):
         except OSError:
             pass
 
+    async def adelete(self, session_key=None):
+        return self.delete(session_key=session_key)
+
     @classmethod
     def clear_expired(cls):
         storage_path = cls._get_storage_path()
@@ -205,3 +220,7 @@ def clear_expired(cls):
             # the create() method.
             session.create = lambda: None
             session.load()
+
+    @classmethod
+    async def aclear_expired(cls):
+        cls.clear_expired()
diff --git a/django/contrib/sessions/backends/signed_cookies.py b/django/contrib/sessions/backends/signed_cookies.py
index dc41c6f12b0d..604cb99808f4 100644
--- a/django/contrib/sessions/backends/signed_cookies.py
+++ b/django/contrib/sessions/backends/signed_cookies.py
@@ -23,6 +23,9 @@ def load(self):
             self.create()
         return {}
 
+    async def aload(self):
+        return self.load()
+
     def create(self):
         """
         To create a new key, set the modified flag so that the cookie is set
@@ -30,6 +33,9 @@ def create(self):
         """
         self.modified = True
 
+    async def acreate(self):
+        return self.create()
+
     def save(self, must_create=False):
         """
         To save, get the session key as a securely signed string and then set
@@ -39,6 +45,9 @@ def save(self, must_create=False):
         self._session_key = self._get_session_key()
         self.modified = True
 
+    async def asave(self, must_create=False):
+        return self.save(must_create=must_create)
+
     def exists(self, session_key=None):
         """
         This method makes sense when you're talking to a shared resource, but
@@ -47,6 +56,9 @@ def exists(self, session_key=None):
         """
         return False
 
+    async def aexists(self, session_key=None):
+        return self.exists(session_key=session_key)
+
     def delete(self, session_key=None):
         """
         To delete, clear the session key and the underlying data structure
@@ -57,6 +69,9 @@ def delete(self, session_key=None):
         self._session_cache = {}
         self.modified = True
 
+    async def adelete(self, session_key=None):
+        return self.delete(session_key=session_key)
+
     def cycle_key(self):
         """
         Keep the same data but with a new key. Call save() and it will
@@ -64,6 +79,9 @@ def cycle_key(self):
         """
         self.save()
 
+    async def acycle_key(self):
+        return self.cycle_key()
+
     def _get_session_key(self):
         """
         Instead of generating a random string, generate a secure url-safe
@@ -79,3 +97,7 @@ def _get_session_key(self):
     @classmethod
     def clear_expired(cls):
         pass
+
+    @classmethod
+    async def aclear_expired(cls):
+        pass
diff --git a/django/test/client.py b/django/test/client.py
index d1fd428ea8cc..0964f87866be 100644
--- a/django/test/client.py
+++ b/django/test/client.py
@@ -817,7 +817,14 @@ def session(self):
         return session
 
     async def asession(self):
-        return await sync_to_async(lambda: self.session)()
+        engine = import_module(settings.SESSION_ENGINE)
+        cookie = self.cookies.get(settings.SESSION_COOKIE_NAME)
+        if cookie:
+            return engine.SessionStore(cookie.value)
+        session = engine.SessionStore()
+        await session.asave()
+        self.cookies[settings.SESSION_COOKIE_NAME] = session.session_key
+        return session
 
     def login(self, **credentials):
         """
@@ -893,7 +900,7 @@ async def _alogin(self, user, backend=None):
 
         await alogin(request, user, backend)
         # Save the session values.
-        await sync_to_async(request.session.save)()
+        await request.session.asave()
         self._set_login_cookies(request)
 
     def _set_login_cookies(self, request):
diff --git a/docs/releases/5.1.txt b/docs/releases/5.1.txt
index e7a99e8d7b80..5fe2be96c893 100644
--- a/docs/releases/5.1.txt
+++ b/docs/releases/5.1.txt
@@ -125,6 +125,10 @@ Minor features
   error messages with their traceback via the newly added
   :ref:`sessions logger <django-contrib-sessions-logger>`.
 
+* :class:`django.contrib.sessions.backends.base.SessionBase` and all built-in
+  session engines now provide async API. The new asynchronous methods all have
+  ``a`` prefixed names, e.g. ``aget()``, ``akeys()``, or ``acycle_key()``.
+
 :mod:`django.contrib.sitemaps`
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
diff --git a/docs/topics/http/sessions.txt b/docs/topics/http/sessions.txt
index d799c245de93..e670292ca86c 100644
--- a/docs/topics/http/sessions.txt
+++ b/docs/topics/http/sessions.txt
@@ -196,54 +196,156 @@ You can edit it multiple times.
       Example: ``'fav_color' in request.session``
 
     .. method:: get(key, default=None)
+    .. method:: aget(key, default=None)
+
+      *Asynchronous version*: ``aget()``
 
       Example: ``fav_color = request.session.get('fav_color', 'red')``
 
+      .. versionchanged:: 5.1
+
+        ``aget()`` function was added.
+
+    .. method:: aset(key, value)
+
+      .. versionadded:: 5.1
+
+      Example: ``await request.session.aset('fav_color', 'red')``
+
+    .. method:: update(dict)
+    .. method:: aupdate(dict)
+
+      *Asynchronous version*: ``aupdate()``
+
+      Example: ``request.session.update({'fav_color': 'red'})``
+
+      .. versionchanged:: 5.1
+
+        ``aupdate()`` function was added.
+
     .. method:: pop(key, default=__not_given)
+    .. method:: apop(key, default=__not_given)
+
+      *Asynchronous version*: ``apop()``
 
       Example: ``fav_color = request.session.pop('fav_color', 'blue')``
 
+      .. versionchanged:: 5.1
+
+        ``apop()`` function was added.
+
     .. method:: keys()
+    .. method:: akeys()
+
+      *Asynchronous version*: ``akeys()``
+
+      .. versionchanged:: 5.1
+
+        ``akeys()`` function was added.
+
+    .. method:: values()
+    .. method:: avalues()
+
+      *Asynchronous version*: ``avalues()``
+
+      .. versionchanged:: 5.1
+
+        ``avalues()`` function was added.
+
+    .. method:: has_key(key)
+    .. method:: ahas_key(key)
+
+      *Asynchronous version*: ``ahas_key()``
+
+      .. versionchanged:: 5.1
+
+        ``ahas_key()`` function was added.
 
     .. method:: items()
+    .. method:: aitems()
+
+      *Asynchronous version*: ``aitems()``
+
+      .. versionchanged:: 5.1
+
+        ``aitems()`` function was added.
 
     .. method:: setdefault()
+    .. method:: asetdefault()
+
+      *Asynchronous version*: ``asetdefault()``
+
+      .. versionchanged:: 5.1
+
+        ``asetdefault()`` function was added.
 
     .. method:: clear()
 
     It also has these methods:
 
     .. method:: flush()
+    .. method:: aflush()
+
+      *Asynchronous version*: ``aflush()``
 
       Deletes the current session data from the session and deletes the session
       cookie. This is used if you want to ensure that the previous session data
       can't be accessed again from the user's browser (for example, the
       :func:`django.contrib.auth.logout()` function calls it).
 
+      .. versionchanged:: 5.1
+
+        ``aflush()`` function was added.
+
     .. method:: set_test_cookie()
+    .. method:: aset_test_cookie()
+
+      *Asynchronous version*: ``aset_test_cookie()``
 
       Sets a test cookie to determine whether the user's browser supports
       cookies. Due to the way cookies work, you won't be able to test this
       until the user's next page request. See `Setting test cookies`_ below for
       more information.
 
+      .. versionchanged:: 5.1
+
+        ``aset_test_cookie()`` function was added.
+
     .. method:: test_cookie_worked()
+    .. method:: atest_cookie_worked()
+
+      *Asynchronous version*: ``atest_cookie_worked()``
 
       Returns either ``True`` or ``False``, depending on whether the user's
       browser accepted the test cookie. Due to the way cookies work, you'll
-      have to call ``set_test_cookie()`` on a previous, separate page request.
+      have to call ``set_test_cookie()`` or ``aset_test_cookie()`` on a
+      previous, separate page request.
       See `Setting test cookies`_ below for more information.
 
+      .. versionchanged:: 5.1
+
+        ``atest_cookie_worked()`` function was added.
+
     .. method:: delete_test_cookie()
+    .. method:: adelete_test_cookie()
+
+      *Asynchronous version*: ``adelete_test_cookie()``
 
       Deletes the test cookie. Use this to clean up after yourself.
 
+      .. versionchanged:: 5.1
+
+        ``adelete_test_cookie()`` function was added.
+
     .. method:: get_session_cookie_age()
 
       Returns the value of the setting :setting:`SESSION_COOKIE_AGE`. This can
       be overridden in a custom session backend.
 
     .. method:: set_expiry(value)
+    .. method:: aset_expiry(value)
+
+      *Asynchronous version*: ``aset_expiry()``
 
       Sets the expiration time for the session. You can pass a number of
       different values:
@@ -266,7 +368,14 @@ You can edit it multiple times.
       purposes. Session expiration is computed from the last time the
       session was *modified*.
 
+      .. versionchanged:: 5.1
+
+        ``aset_expiry()`` function was added.
+
     .. method:: get_expiry_age()
+    .. method:: aget_expiry_age()
+
+      *Asynchronous version*: ``aget_expiry_age()``
 
       Returns the number of seconds until this session expires. For sessions
       with no custom expiration (or those set to expire at browser close), this
@@ -279,7 +388,7 @@ You can edit it multiple times.
       - ``expiry``: expiry information for the session, as a
         :class:`~datetime.datetime` object, an :class:`int` (in seconds), or
         ``None``. Defaults to the value stored in the session by
-        :meth:`set_expiry`, if there is one, or ``None``.
+        :meth:`set_expiry`/:meth:`aset_expiry`, if there is one, or ``None``.
 
       .. note::
 
@@ -295,7 +404,14 @@ You can edit it multiple times.
 
             expires_at = modification + timedelta(seconds=settings.SESSION_COOKIE_AGE)
 
+      .. versionchanged:: 5.1
+
+        ``aget_expiry_age()`` function was added.
+
     .. method:: get_expiry_date()
+    .. method:: aget_expiry_date()
+
+      *Asynchronous version*: ``aget_expiry_date()``
 
       Returns the date this session will expire. For sessions with no custom
       expiration (or those set to expire at browser close), this will equal the
@@ -304,22 +420,47 @@ You can edit it multiple times.
       This function accepts the same keyword arguments as
       :meth:`get_expiry_age`, and similar notes on usage apply.
 
+      .. versionchanged:: 5.1
+
+        ``aget_expiry_date()`` function was added.
+
     .. method:: get_expire_at_browser_close()
+    .. method:: aget_expire_at_browser_close()
+
+      *Asynchronous version*: ``aget_expire_at_browser_close()``
 
       Returns either ``True`` or ``False``, depending on whether the user's
       session cookie will expire when the user's web browser is closed.
 
+      .. versionchanged:: 5.1
+
+        ``aget_expire_at_browser_close()`` function was added.
+
     .. method:: clear_expired()
+    .. method:: aclear_expired()
+
+      *Asynchronous version*: ``aclear_expired()``
 
       Removes expired sessions from the session store. This class method is
       called by :djadmin:`clearsessions`.
 
+      .. versionchanged:: 5.1
+
+        ``aclear_expired()`` function was added.
+
     .. method:: cycle_key()
+    .. method:: acycle_key()
+
+      *Asynchronous version*: ``acycle_key()``
 
       Creates a new session key while retaining the current session data.
       :func:`django.contrib.auth.login()` calls this method to mitigate against
       session fixation.
 
+      .. versionchanged:: 5.1
+
+        ``acycle_key()`` function was added.
+
 .. _session_serialization:
 
 Session serialization
@@ -475,6 +616,10 @@ Here's a typical usage example::
         request.session.set_test_cookie()
         return render(request, "foo/login_form.html")
 
+.. versionchanged:: 5.1
+
+    Support for setting test cookies in asynchronous view functions was added.
+
 Using sessions out of views
 ===========================
 
@@ -694,16 +839,26 @@ the corresponding session engine. By convention, the session store object class
 is named ``SessionStore`` and is located in the module designated by
 :setting:`SESSION_ENGINE`.
 
-All ``SessionStore`` classes available in Django inherit from
-:class:`~backends.base.SessionBase` and implement data manipulation methods,
-namely:
+All ``SessionStore`` subclasses available in Django implement the following
+data manipulation methods:
 
 * ``exists()``
 * ``create()``
 * ``save()``
 * ``delete()``
 * ``load()``
-* :meth:`~backends.base.SessionBase.clear_expired`
+* :meth:`~.SessionBase.clear_expired`
+
+An asynchronous interface for these methods is provided by wrapping them with
+``sync_to_async()``. They can be implemented directly if an async-native
+implementation is available:
+
+* ``aexists()``
+* ``acreate()``
+* ``asave()``
+* ``adelete()``
+* ``aload()``
+* :meth:`~.SessionBase.aclear_expired`
 
 In order to build a custom session engine or to customize an existing one, you
 may create a new class inheriting from :class:`~backends.base.SessionBase` or
@@ -713,6 +868,11 @@ You can extend the session engines, but doing so with database-backed session
 engines generally requires some extra effort (see the next section for
 details).
 
+.. versionchanged:: 5.1
+
+    ``aexists()``, ``acreate()``, ``asave()``, ``adelete()``, ``aload()``, and
+    ``aclear_expired()`` methods were added.
+
 .. _extending-database-backed-session-engines:
 
 Extending database-backed session engines
diff --git a/tests/cache/failing_cache.py b/tests/cache/failing_cache.py
index e2f0043bb7b9..1c9b5996d61f 100644
--- a/tests/cache/failing_cache.py
+++ b/tests/cache/failing_cache.py
@@ -5,3 +5,6 @@ class CacheClass(LocMemCache):
 
     def set(self, *args, **kwargs):
         raise Exception("Faked exception saving to cache")
+
+    async def aset(self, *args, **kwargs):
+        raise Exception("Faked exception saving to cache")
diff --git a/tests/sessions_tests/tests.py b/tests/sessions_tests/tests.py
index c8c556b4dccb..9eabb933a8ab 100644
--- a/tests/sessions_tests/tests.py
+++ b/tests/sessions_tests/tests.py
@@ -61,11 +61,19 @@ def test_new_session(self):
     def test_get_empty(self):
         self.assertIsNone(self.session.get("cat"))
 
+    async def test_get_empty_async(self):
+        self.assertIsNone(await self.session.aget("cat"))
+
     def test_store(self):
         self.session["cat"] = "dog"
         self.assertIs(self.session.modified, True)
         self.assertEqual(self.session.pop("cat"), "dog")
 
+    async def test_store_async(self):
+        await self.session.aset("cat", "dog")
+        self.assertIs(self.session.modified, True)
+        self.assertEqual(await self.session.apop("cat"), "dog")
+
     def test_pop(self):
         self.session["some key"] = "exists"
         # Need to reset these to pretend we haven't accessed it:
@@ -77,6 +85,17 @@ def test_pop(self):
         self.assertIs(self.session.modified, True)
         self.assertIsNone(self.session.get("some key"))
 
+    async def test_pop_async(self):
+        await self.session.aset("some key", "exists")
+        # Need to reset these to pretend we haven't accessed it:
+        self.accessed = False
+        self.modified = False
+
+        self.assertEqual(await self.session.apop("some key"), "exists")
+        self.assertIs(self.session.accessed, True)
+        self.assertIs(self.session.modified, True)
+        self.assertIsNone(await self.session.aget("some key"))
+
     def test_pop_default(self):
         self.assertEqual(
             self.session.pop("some key", "does not exist"), "does not exist"
@@ -84,6 +103,13 @@ def test_pop_default(self):
         self.assertIs(self.session.accessed, True)
         self.assertIs(self.session.modified, False)
 
+    async def test_pop_default_async(self):
+        self.assertEqual(
+            await self.session.apop("some key", "does not exist"), "does not exist"
+        )
+        self.assertIs(self.session.accessed, True)
+        self.assertIs(self.session.modified, False)
+
     def test_pop_default_named_argument(self):
         self.assertEqual(
             self.session.pop("some key", default="does not exist"), "does not exist"
@@ -91,22 +117,46 @@ def test_pop_default_named_argument(self):
         self.assertIs(self.session.accessed, True)
         self.assertIs(self.session.modified, False)
 
+    async def test_pop_default_named_argument_async(self):
+        self.assertEqual(
+            await self.session.apop("some key", default="does not exist"),
+            "does not exist",
+        )
+        self.assertIs(self.session.accessed, True)
+        self.assertIs(self.session.modified, False)
+
     def test_pop_no_default_keyerror_raised(self):
         with self.assertRaises(KeyError):
             self.session.pop("some key")
 
+    async def test_pop_no_default_keyerror_raised_async(self):
+        with self.assertRaises(KeyError):
+            await self.session.apop("some key")
+
     def test_setdefault(self):
         self.assertEqual(self.session.setdefault("foo", "bar"), "bar")
         self.assertEqual(self.session.setdefault("foo", "baz"), "bar")
         self.assertIs(self.session.accessed, True)
         self.assertIs(self.session.modified, True)
 
+    async def test_setdefault_async(self):
+        self.assertEqual(await self.session.asetdefault("foo", "bar"), "bar")
+        self.assertEqual(await self.session.asetdefault("foo", "baz"), "bar")
+        self.assertIs(self.session.accessed, True)
+        self.assertIs(self.session.modified, True)
+
     def test_update(self):
         self.session.update({"update key": 1})
         self.assertIs(self.session.accessed, True)
         self.assertIs(self.session.modified, True)
         self.assertEqual(self.session.get("update key", None), 1)
 
+    async def test_update_async(self):
+        await self.session.aupdate({"update key": 1})
+        self.assertIs(self.session.accessed, True)
+        self.assertIs(self.session.modified, True)
+        self.assertEqual(await self.session.aget("update key", None), 1)
+
     def test_has_key(self):
         self.session["some key"] = 1
         self.session.modified = False
@@ -115,6 +165,14 @@ def test_has_key(self):
         self.assertIs(self.session.accessed, True)
         self.assertIs(self.session.modified, False)
 
+    async def test_has_key_async(self):
+        await self.session.aset("some key", 1)
+        self.session.modified = False
+        self.session.accessed = False
+        self.assertIs(await self.session.ahas_key("some key"), True)
+        self.assertIs(self.session.accessed, True)
+        self.assertIs(self.session.modified, False)
+
     def test_values(self):
         self.assertEqual(list(self.session.values()), [])
         self.assertIs(self.session.accessed, True)
@@ -125,6 +183,16 @@ def test_values(self):
         self.assertIs(self.session.accessed, True)
         self.assertIs(self.session.modified, False)
 
+    async def test_values_async(self):
+        self.assertEqual(list(await self.session.avalues()), [])
+        self.assertIs(self.session.accessed, True)
+        await self.session.aset("some key", 1)
+        self.session.modified = False
+        self.session.accessed = False
+        self.assertEqual(list(await self.session.avalues()), [1])
+        self.assertIs(self.session.accessed, True)
+        self.assertIs(self.session.modified, False)
+
     def test_keys(self):
         self.session["x"] = 1
         self.session.modified = False
@@ -133,6 +201,14 @@ def test_keys(self):
         self.assertIs(self.session.accessed, True)
         self.assertIs(self.session.modified, False)
 
+    async def test_keys_async(self):
+        await self.session.aset("x", 1)
+        self.session.modified = False
+        self.session.accessed = False
+        self.assertEqual(list(await self.session.akeys()), ["x"])
+        self.assertIs(self.session.accessed, True)
+        self.assertIs(self.session.modified, False)
+
     def test_items(self):
         self.session["x"] = 1
         self.session.modified = False
@@ -141,6 +217,14 @@ def test_items(self):
         self.assertIs(self.session.accessed, True)
         self.assertIs(self.session.modified, False)
 
+    async def test_items_async(self):
+        await self.session.aset("x", 1)
+        self.session.modified = False
+        self.session.accessed = False
+        self.assertEqual(list(await self.session.aitems()), [("x", 1)])
+        self.assertIs(self.session.accessed, True)
+        self.assertIs(self.session.modified, False)
+
     def test_clear(self):
         self.session["x"] = 1
         self.session.modified = False
@@ -155,11 +239,20 @@ def test_save(self):
         self.session.save()
         self.assertIs(self.session.exists(self.session.session_key), True)
 
+    async def test_save_async(self):
+        await self.session.asave()
+        self.assertIs(await self.session.aexists(self.session.session_key), True)
+
     def test_delete(self):
         self.session.save()
         self.session.delete(self.session.session_key)
         self.assertIs(self.session.exists(self.session.session_key), False)
 
+    async def test_delete_async(self):
+        await self.session.asave()
+        await self.session.adelete(self.session.session_key)
+        self.assertIs(await self.session.aexists(self.session.session_key), False)
+
     def test_flush(self):
         self.session["foo"] = "bar"
         self.session.save()
@@ -171,6 +264,17 @@ def test_flush(self):
         self.assertIs(self.session.modified, True)
         self.assertIs(self.session.accessed, True)
 
+    async def test_flush_async(self):
+        await self.session.aset("foo", "bar")
+        await self.session.asave()
+        prev_key = self.session.session_key
+        await self.session.aflush()
+        self.assertIs(await self.session.aexists(prev_key), False)
+        self.assertNotEqual(self.session.session_key, prev_key)
+        self.assertIsNone(self.session.session_key)
+        self.assertIs(self.session.modified, True)
+        self.assertIs(self.session.accessed, True)
+
     def test_cycle(self):
         self.session["a"], self.session["b"] = "c", "d"
         self.session.save()
@@ -181,6 +285,17 @@ def test_cycle(self):
         self.assertNotEqual(self.session.session_key, prev_key)
         self.assertEqual(list(self.session.items()), prev_data)
 
+    async def test_cycle_async(self):
+        await self.session.aset("a", "c")
+        await self.session.aset("b", "d")
+        await self.session.asave()
+        prev_key = self.session.session_key
+        prev_data = list(await self.session.aitems())
+        await self.session.acycle_key()
+        self.assertIs(await self.session.aexists(prev_key), False)
+        self.assertNotEqual(self.session.session_key, prev_key)
+        self.assertEqual(list(await self.session.aitems()), prev_data)
+
     def test_cycle_with_no_session_cache(self):
         self.session["a"], self.session["b"] = "c", "d"
         self.session.save()
@@ -190,11 +305,26 @@ def test_cycle_with_no_session_cache(self):
         self.session.cycle_key()
         self.assertCountEqual(self.session.items(), prev_data)
 
+    async def test_cycle_with_no_session_cache_async(self):
+        await self.session.aset("a", "c")
+        await self.session.aset("b", "d")
+        await self.session.asave()
+        prev_data = await self.session.aitems()
+        self.session = self.backend(self.session.session_key)
+        self.assertIs(hasattr(self.session, "_session_cache"), False)
+        await self.session.acycle_key()
+        self.assertCountEqual(await self.session.aitems(), prev_data)
+
     def test_save_doesnt_clear_data(self):
         self.session["a"] = "b"
         self.session.save()
         self.assertEqual(self.session["a"], "b")
 
+    async def test_save_doesnt_clear_data_async(self):
+        await self.session.aset("a", "b")
+        await self.session.asave()
+        self.assertEqual(await self.session.aget("a"), "b")
+
     def test_invalid_key(self):
         # Submitting an invalid session key (either by guessing, or if the db has
         # removed the key) results in a new key being generated.
@@ -209,6 +339,20 @@ def test_invalid_key(self):
             # session key; make sure that entry is manually deleted
             session.delete("1")
 
+    async def test_invalid_key_async(self):
+        # Submitting an invalid session key (either by guessing, or if the db has
+        # removed the key) results in a new key being generated.
+        try:
+            session = self.backend("1")
+            await session.asave()
+            self.assertNotEqual(session.session_key, "1")
+            self.assertIsNone(await session.aget("cat"))
+            await session.adelete()
+        finally:
+            # Some backends leave a stale cache entry for the invalid
+            # session key; make sure that entry is manually deleted
+            await session.adelete("1")
+
     def test_session_key_empty_string_invalid(self):
         """Falsey values (Such as an empty string) are rejected."""
         self.session._session_key = ""
@@ -241,6 +385,18 @@ def test_default_expiry(self):
         self.session.set_expiry(0)
         self.assertEqual(self.session.get_expiry_age(), settings.SESSION_COOKIE_AGE)
 
+    async def test_default_expiry_async(self):
+        # A normal session has a max age equal to settings.
+        self.assertEqual(
+            await self.session.aget_expiry_age(), settings.SESSION_COOKIE_AGE
+        )
+        # So does a custom session with an idle expiration time of 0 (but it'll
+        # expire at browser close).
+        await self.session.aset_expiry(0)
+        self.assertEqual(
+            await self.session.aget_expiry_age(), settings.SESSION_COOKIE_AGE
+        )
+
     def test_custom_expiry_seconds(self):
         modification = timezone.now()
 
@@ -252,6 +408,17 @@ def test_custom_expiry_seconds(self):
         age = self.session.get_expiry_age(modification=modification)
         self.assertEqual(age, 10)
 
+    async def test_custom_expiry_seconds_async(self):
+        modification = timezone.now()
+
+        await self.session.aset_expiry(10)
+
+        date = await self.session.aget_expiry_date(modification=modification)
+        self.assertEqual(date, modification + timedelta(seconds=10))
+
+        age = await self.session.aget_expiry_age(modification=modification)
+        self.assertEqual(age, 10)
+
     def test_custom_expiry_timedelta(self):
         modification = timezone.now()
 
@@ -269,6 +436,23 @@ def test_custom_expiry_timedelta(self):
         age = self.session.get_expiry_age(modification=modification)
         self.assertEqual(age, 10)
 
+    async def test_custom_expiry_timedelta_async(self):
+        modification = timezone.now()
+
+        # Mock timezone.now, because set_expiry calls it on this code path.
+        original_now = timezone.now
+        try:
+            timezone.now = lambda: modification
+            await self.session.aset_expiry(timedelta(seconds=10))
+        finally:
+            timezone.now = original_now
+
+        date = await self.session.aget_expiry_date(modification=modification)
+        self.assertEqual(date, modification + timedelta(seconds=10))
+
+        age = await self.session.aget_expiry_age(modification=modification)
+        self.assertEqual(age, 10)
+
     def test_custom_expiry_datetime(self):
         modification = timezone.now()
 
@@ -280,12 +464,31 @@ def test_custom_expiry_datetime(self):
         age = self.session.get_expiry_age(modification=modification)
         self.assertEqual(age, 10)
 
+    async def test_custom_expiry_datetime_async(self):
+        modification = timezone.now()
+
+        await self.session.aset_expiry(modification + timedelta(seconds=10))
+
+        date = await self.session.aget_expiry_date(modification=modification)
+        self.assertEqual(date, modification + timedelta(seconds=10))
+
+        age = await self.session.aget_expiry_age(modification=modification)
+        self.assertEqual(age, 10)
+
     def test_custom_expiry_reset(self):
         self.session.set_expiry(None)
         self.session.set_expiry(10)
         self.session.set_expiry(None)
         self.assertEqual(self.session.get_expiry_age(), settings.SESSION_COOKIE_AGE)
 
+    async def test_custom_expiry_reset_async(self):
+        await self.session.aset_expiry(None)
+        await self.session.aset_expiry(10)
+        await self.session.aset_expiry(None)
+        self.assertEqual(
+            await self.session.aget_expiry_age(), settings.SESSION_COOKIE_AGE
+        )
+
     def test_get_expire_at_browser_close(self):
         # Tests get_expire_at_browser_close with different settings and different
         # set_expiry calls
@@ -309,6 +512,29 @@ def test_get_expire_at_browser_close(self):
             self.session.set_expiry(None)
             self.assertIs(self.session.get_expire_at_browser_close(), True)
 
+    async def test_get_expire_at_browser_close_async(self):
+        # Tests get_expire_at_browser_close with different settings and different
+        # set_expiry calls
+        with override_settings(SESSION_EXPIRE_AT_BROWSER_CLOSE=False):
+            await self.session.aset_expiry(10)
+            self.assertIs(await self.session.aget_expire_at_browser_close(), False)
+
+            await self.session.aset_expiry(0)
+            self.assertIs(await self.session.aget_expire_at_browser_close(), True)
+
+            await self.session.aset_expiry(None)
+            self.assertIs(await self.session.aget_expire_at_browser_close(), False)
+
+        with override_settings(SESSION_EXPIRE_AT_BROWSER_CLOSE=True):
+            await self.session.aset_expiry(10)
+            self.assertIs(await self.session.aget_expire_at_browser_close(), False)
+
+            await self.session.aset_expiry(0)
+            self.assertIs(await self.session.aget_expire_at_browser_close(), True)
+
+            await self.session.aset_expiry(None)
+            self.assertIs(await self.session.aget_expire_at_browser_close(), True)
+
     def test_decode(self):
         # Ensure we can decode what we encode
         data = {"a test key": "a test value"}
@@ -350,6 +576,22 @@ def test_actual_expiry(self):
             self.session.delete(old_session_key)
             self.session.delete(new_session_key)
 
+    async def test_actual_expiry_async(self):
+        old_session_key = None
+        new_session_key = None
+        try:
+            await self.session.aset("foo", "bar")
+            await self.session.aset_expiry(-timedelta(seconds=10))
+            await self.session.asave()
+            old_session_key = self.session.session_key
+            # With an expiry date in the past, the session expires instantly.
+            new_session = self.backend(self.session.session_key)
+            new_session_key = new_session.session_key
+            self.assertIs(await new_session.ahas_key("foo"), False)
+        finally:
+            await self.session.adelete(old_session_key)
+            await self.session.adelete(new_session_key)
+
     def test_session_load_does_not_create_record(self):
         """
         Loading an unknown session key does not create a session record.
@@ -364,6 +606,15 @@ def test_session_load_does_not_create_record(self):
         # provided unknown key was cycled, not reused
         self.assertNotEqual(session.session_key, "someunknownkey")
 
+    async def test_session_load_does_not_create_record_async(self):
+        session = self.backend("someunknownkey")
+        await session.aload()
+
+        self.assertIsNone(session.session_key)
+        self.assertIs(await session.aexists(session.session_key), False)
+        # Provided unknown key was cycled, not reused.
+        self.assertNotEqual(session.session_key, "someunknownkey")
+
     def test_session_save_does_not_resurrect_session_logged_out_in_other_context(self):
         """
         Sessions shouldn't be resurrected by a concurrent request.
@@ -386,6 +637,28 @@ def test_session_save_does_not_resurrect_session_logged_out_in_other_context(sel
 
         self.assertEqual(s1.load(), {})
 
+    async def test_session_asave_does_not_resurrect_session_logged_out_in_other_context(
+        self,
+    ):
+        """Sessions shouldn't be resurrected by a concurrent request."""
+        # Create new session.
+        s1 = self.backend()
+        await s1.aset("test_data", "value1")
+        await s1.asave(must_create=True)
+
+        # Logout in another context.
+        s2 = self.backend(s1.session_key)
+        await s2.adelete()
+
+        # Modify session in first context.
+        await s1.aset("test_data", "value2")
+        with self.assertRaises(UpdateError):
+            # This should throw an exception as the session is deleted, not
+            # resurrect the session.
+            await s1.asave()
+
+        self.assertEqual(await s1.aload(), {})
+
 
 class DatabaseSessionTests(SessionTestsMixin, TestCase):
     backend = DatabaseSession
@@ -456,6 +729,25 @@ def test_clearsessions_command(self):
         # ... and one is deleted.
         self.assertEqual(1, self.model.objects.count())
 
+    async def test_aclear_expired(self):
+        self.assertEqual(await self.model.objects.acount(), 0)
+
+        # Object in the future.
+        await self.session.aset("key", "value")
+        await self.session.aset_expiry(3600)
+        await self.session.asave()
+        # Object in the past.
+        other_session = self.backend()
+        await other_session.aset("key", "value")
+        await other_session.aset_expiry(-3600)
+        await other_session.asave()
+
+        # Two sessions are in the database before clearing expired.
+        self.assertEqual(await self.model.objects.acount(), 2)
+        await self.session.aclear_expired()
+        await other_session.aclear_expired()
+        self.assertEqual(await self.model.objects.acount(), 1)
+
 
 @override_settings(USE_TZ=True)
 class DatabaseSessionWithTimeZoneTests(DatabaseSessionTests):
@@ -491,11 +783,28 @@ def test_custom_expiry_reset(self):
         self.session.set_expiry(None)
         self.assertEqual(self.session.get_expiry_age(), self.custom_session_cookie_age)
 
+    async def test_custom_expiry_reset_async(self):
+        await self.session.aset_expiry(None)
+        await self.session.aset_expiry(10)
+        await self.session.aset_expiry(None)
+        self.assertEqual(
+            await self.session.aget_expiry_age(), self.custom_session_cookie_age
+        )
+
     def test_default_expiry(self):
         self.assertEqual(self.session.get_expiry_age(), self.custom_session_cookie_age)
         self.session.set_expiry(0)
         self.assertEqual(self.session.get_expiry_age(), self.custom_session_cookie_age)
 
+    async def test_default_expiry_async(self):
+        self.assertEqual(
+            await self.session.aget_expiry_age(), self.custom_session_cookie_age
+        )
+        await self.session.aset_expiry(0)
+        self.assertEqual(
+            await self.session.aget_expiry_age(), self.custom_session_cookie_age
+        )
+
 
 class CacheDBSessionTests(SessionTestsMixin, TestCase):
     backend = CacheDBSession
@@ -533,6 +842,22 @@ def test_cache_set_failure_non_fatal(self):
         self.assertEqual(log.message, f"Error saving to cache ({session._cache})")
         self.assertEqual(str(log.exc_info[1]), "Faked exception saving to cache")
 
+    @override_settings(
+        CACHES={"default": {"BACKEND": "cache.failing_cache.CacheClass"}}
+    )
+    async def test_cache_async_set_failure_non_fatal(self):
+        """Failing to write to the cache does not raise errors."""
+        session = self.backend()
+        await session.aset("key", "val")
+
+        with self.assertLogs("django.contrib.sessions", "ERROR") as cm:
+            await session.asave()
+
+        # A proper ERROR log message was recorded.
+        log = cm.records[-1]
+        self.assertEqual(log.message, f"Error saving to cache ({session._cache})")
+        self.assertEqual(str(log.exc_info[1]), "Faked exception saving to cache")
+
 
 @override_settings(USE_TZ=True)
 class CacheDBSessionWithTimeZoneTests(CacheDBSessionTests):
@@ -673,6 +998,12 @@ def test_create_and_save(self):
         self.session.save()
         self.assertIsNotNone(caches["default"].get(self.session.cache_key))
 
+    async def test_create_and_save_async(self):
+        self.session = self.backend()
+        await self.session.acreate()
+        await self.session.asave()
+        self.assertIsNotNone(caches["default"].get(await self.session.acache_key()))
+
 
 class SessionMiddlewareTests(TestCase):
     request_factory = RequestFactory()
@@ -899,6 +1230,9 @@ def test_save(self):
         """
         pass
 
+    async def test_save_async(self):
+        pass
+
     def test_cycle(self):
         """
         This test tested cycle_key() which would create a new session
@@ -908,11 +1242,17 @@ def test_cycle(self):
         """
         pass
 
+    async def test_cycle_async(self):
+        pass
+
     @unittest.expectedFailure
     def test_actual_expiry(self):
         # The cookie backend doesn't handle non-default expiry dates, see #19201
         super().test_actual_expiry()
 
+    async def test_actual_expiry_async(self):
+        pass
+
     def test_unpickling_exception(self):
         # signed_cookies backend should handle unpickle exceptions gracefully
         # by creating a new session
@@ -927,12 +1267,26 @@ def test_unpickling_exception(self):
     def test_session_load_does_not_create_record(self):
         pass
 
+    @unittest.skip(
+        "Cookie backend doesn't have an external store to create records in."
+    )
+    async def test_session_load_does_not_create_record_async(self):
+        pass
+
     @unittest.skip(
         "CookieSession is stored in the client and there is no way to query it."
     )
     def test_session_save_does_not_resurrect_session_logged_out_in_other_context(self):
         pass
 
+    @unittest.skip(
+        "CookieSession is stored in the client and there is no way to query it."
+    )
+    async def test_session_asave_does_not_resurrect_session_logged_out_in_other_context(
+        self,
+    ):
+        pass
+
 
 class ClearSessionsCommandTests(SimpleTestCase):
     def test_clearsessions_unsupported(self):
@@ -956,26 +1310,51 @@ def test_create(self):
         with self.assertRaisesMessage(NotImplementedError, msg):
             self.session.create()
 
+    async def test_acreate(self):
+        msg = self.not_implemented_msg % "a create"
+        with self.assertRaisesMessage(NotImplementedError, msg):
+            await self.session.acreate()
+
     def test_delete(self):
         msg = self.not_implemented_msg % "a delete"
         with self.assertRaisesMessage(NotImplementedError, msg):
             self.session.delete()
 
+    async def test_adelete(self):
+        msg = self.not_implemented_msg % "a delete"
+        with self.assertRaisesMessage(NotImplementedError, msg):
+            await self.session.adelete()
+
     def test_exists(self):
         msg = self.not_implemented_msg % "an exists"
         with self.assertRaisesMessage(NotImplementedError, msg):
             self.session.exists(None)
 
+    async def test_aexists(self):
+        msg = self.not_implemented_msg % "an exists"
+        with self.assertRaisesMessage(NotImplementedError, msg):
+            await self.session.aexists(None)
+
     def test_load(self):
         msg = self.not_implemented_msg % "a load"
         with self.assertRaisesMessage(NotImplementedError, msg):
             self.session.load()
 
+    async def test_aload(self):
+        msg = self.not_implemented_msg % "a load"
+        with self.assertRaisesMessage(NotImplementedError, msg):
+            await self.session.aload()
+
     def test_save(self):
         msg = self.not_implemented_msg % "a save"
         with self.assertRaisesMessage(NotImplementedError, msg):
             self.session.save()
 
+    async def test_asave(self):
+        msg = self.not_implemented_msg % "a save"
+        with self.assertRaisesMessage(NotImplementedError, msg):
+            await self.session.asave()
+
     def test_test_cookie(self):
         self.assertIs(self.session.has_key(self.session.TEST_COOKIE_NAME), False)
         self.session.set_test_cookie()
@@ -983,5 +1362,12 @@ def test_test_cookie(self):
         self.session.delete_test_cookie()
         self.assertIs(self.session.has_key(self.session.TEST_COOKIE_NAME), False)
 
+    async def test_atest_cookie(self):
+        self.assertIs(await self.session.ahas_key(self.session.TEST_COOKIE_NAME), False)
+        await self.session.aset_test_cookie()
+        self.assertIs(await self.session.atest_cookie_worked(), True)
+        await self.session.adelete_test_cookie()
+        self.assertIs(await self.session.ahas_key(self.session.TEST_COOKIE_NAME), False)
+
     def test_is_empty(self):
         self.assertIs(self.session.is_empty(), True)