Django middleware to cache request.user

middleware.py

from django.contrib.auth import SESSION_KEY
from django.core.cache import cache
from django.utils.functional import SimpleLazyObject
from django.contrib.auth.signals import user_logged_out
from django.contrib.auth import get_user_model
from django.db.models.signals import post_save, post_delete

from functools import partial


# Cache key for this middleware.
CACHE_KEY = "cached_user_middleware:%s"


class CachedUserMiddleware:
    """
    Middleware that caches the authenticated user. If the user is not in the
    cache, it will fall-back to the original next user (`request.user`, which
    is assumed to be a `SimpleLazyObject` as well).

    This middleware must come after the session middleware and any other
    middleware that (potentially) modifies `request.user`.

    Implementation based on
    https://github.com/ui/django-cached_authentication_middleware
    """

    def __init__(self, get_response=None):
        self.get_response = get_response

        self.request = None

        # Register signals.
        post_save.connect(self.on_user_save, sender=get_user_model())
        post_delete.connect(self.on_user_deleted, sender=get_user_model())
        user_logged_out.connect(self.on_user_logged_out)

    def __call__(self, request):
        """
        Invoke middleware.

        It will overwrite `request.user` with `SimpleLazyObject`, so that it
        will only hit the cache if the user object is actually accessed. If
        the user is not in the cache, the original `request.user` is
        considered. It is assumed that this is a `SimpleLazyObject` as well.
        """

        self.request = request

        request.user = SimpleLazyObject(
            partial(self.get_cached_user, request.user)
        )

        return self.get_response(request)

    def on_user_save(self, *args, **kwargs):
        """
        Invalidate the cached user (if set) when the user model is updated.
        """

        user = kwargs["instance"]

        if self.request is not None:
            if user.id == self.request.user.id:
                self.delete_cached_user()

    def on_user_deleted(self, *args, **kwargs):
        """
        Invalidate the cached user (if set) when the user model is deleted.
        """

        user = kwargs["instance"]

        if self.request is not None:
            if user.id == self.request.user.id:
                self.delete_cached_user()

    def on_user_logged_out(self, *args, **kwargs):
        """
        Invalidate the cached user (if set) when the user logout signal is
        received.
        """

        if self.request is not None:
            self.delete_cached_user()

    def delete_cached_user(self):
        """
        Delete the user from the cache.
        """

        key = CACHE_KEY % self.request.session[SESSION_KEY]
        cache.delete(key)

    def get_cached_user(self, next_user):
        """
        Try to get the user from the cache, or return the `next_user`.
        """

        if not hasattr(self.request, "_cached_user"):
            key = CACHE_KEY % self.request.session[SESSION_KEY]

            try:
                user = cache.get(key)
            except KeyError:
                user = None

            if user is None:
                user = next_user

                # Cache authenticated users only.
                if getattr(user, "is_authenticated", False):
                    cache.set(key, user)

            self.request._cached_user = user
        return self.request._cached_user