Async API Calls Unleashed: Exploring Django 4 and Django Rest Framework

Django recently introduced support for asynchronous operations as an experimental feature starting from Django 3. However, this async support is currently not available for all parts of Django, including the Object-Relational Mapping (ORM) and other components.

Asynchronous views work seamlessly within the Django Model-View-Controller (MVC) paradigm. However, difficulties arise when attempting to expose views as a REST layer using the Django Rest Framework (DRF) due to DRF's synchronous nature.

In this article, I will guide you through the steps of overriding DRF to enable support for asynchronous API calls. Please note that I won't delve deep into the setup of DRF. You can access the starter project we'll be using for this tutorial here.

Project Structure

We're going to modify the project structure of the provided link, transitioning it from using synchronous to asynchronous functions with the help of DRF. Below is the initial and modified project structure:

Initial structure:

drf_demo/
 |-- asyncdrf/
 |    |-- ...
 |-- clients/
 |    |-- ...
 |-- requirements.txt
 |-- manage.py

Modified structure

drf_demo/
 |-- asyncdrf/
 |    |-- ...
 |-- clients/
 |    |-- ...
 +-- drfutil/
 |    +-- ...
 |-- requirements.txt
 |-- manage.py

Async Django Rest Framework

To achieve asynchronous API calls with DRF, we'll create a new folder named drfutil in the root directory. Inside this folder, we'll implement utility classes to override DRF methods and make them asynchronous. The files to be added are as follows:

authentication_classes.py
requests.py
views.py

Each of these files will contain specific pieces of code that collectively enable asynchronous functionality in DRF.

Adding code to the files

Add the following code to authentication_classes.py

from rest_framework import HTTP_HEADER_ENCODING, exceptions
from asgiref.sync import sync_to_async
from rest_framework.authentication import BaseAuthentication
from rest_framework.permissions import BasePermission
from drfutil import AsyncRequest, AsyncAPIView
from rest_framework.throttling import BaseThrottle
import asyncio

@sync_to_async
def get_token(model, key):
    return model.objects.select_related('user').get(key=key)


def get_authorization_header(request):
    """
    Return request's 'Authorization:' header, as a byte string.
    Hide some test client ickiness where the header can be unicode.
    """
    auth = request.META.get('HTTP_AUTHORIZATION', b'')
    if isinstance(auth, str):
        # Work around Django test client oddness
        auth = auth.encode(HTTP_HEADER_ENCODING)
    return auth

class AsyncAuthentication(BaseAuthentication):    
    """
    Simple token based authentication.
    Clients should authenticate by passing the token key in the "Authorization"
    HTTP header, prepended with the string "Token ".  For example:
        Authorization: Token 401f7ac837da42b97f613d789819ff93537bee6a
    """

    keyword = 'Token'
    model = None

    def get_model(self):
        if self.model is not None:
            return self.model
        from rest_framework.authtoken.models import Token
        return Token


    """
    A custom token model may be used, but must have the following properties.
    * key -- The string identifying the token
    * user -- The user to which the token belongs
    """

    async def authenticate(self, request):
        auth = get_authorization_header(request).split()

        if not auth or auth[0].lower() != self.keyword.lower().encode():
            return None

        if len(auth) == 1:
            msg = _('Invalid token header. No credentials provided.')
            raise exceptions.AuthenticationFailed(msg)
        elif len(auth) > 2:
            msg = _('Invalid token header. Token string should not contain spaces.')
            raise exceptions.AuthenticationFailed(msg)

        try:
            token = auth[1].decode()
        except UnicodeError:
            msg = _('Invalid token header. Token string should not contain invalid characters.')
            raise exceptions.AuthenticationFailed(msg)

        auth_creds = await self.authenticate_credentials(token)
        return auth_creds

    async def authenticate_credentials(self, key):
        model = None
        if self.model is not None:
            model = self.model
        else:
            from rest_framework.authtoken.models import Token
            model = Token
        try:
            token = await get_token(model, key)

        except model.DoesNotExist:
            raise exceptions.AuthenticationFailed(_('Invalid token.'))

        if not token.user.is_active:
            raise exceptions.AuthenticationFailed(_('User inactive or deleted.'))

        return (token.user, token)

    async def authenticate_header(self, request):
        return self.keyword


class AsyncPermission(BasePermission):
    async def has_permission(self, request: AsyncRequest, view: AsyncAPIView) -> bool:
        await asyncio.sleep(0.01)
        return True


class AsyncThrottle(BaseThrottle):
    async def allow_request(self, request: AsyncRequest, view: AsyncAPIView) -> bool:
        await asyncio.sleep(0.01)
        return True

class AsyncIsAuthenticated(BasePermission):
    async def has_permission(self, request: AsyncRequest, view: AsyncAPIView):
        return bool(request.user and request.user.is_authenticated)

Add the following code to requests.py:

import asyncio

from rest_framework.request import Request
from rest_framework import exceptions
from asgiref.sync import sync_to_async, async_to_sync


class AsyncRequest(Request):
    async def authenticate(self):
        """
        Attempt to authenticate the request using each authentication instance
        in turn.
        """
        self._authenticator, self.user, self.auth = None, None, None

        for authenticator in self.authenticators:
            try:
                if asyncio.iscoroutinefunction(authenticator.authenticate):
                    user_auth_tuple = await authenticator.authenticate(self)

                else:
                    user_auth_tuple = authenticator.authenticate(self)
            except exceptions.APIException:
                self._not_authenticated()
                raise

            if user_auth_tuple is not None:
                self._authenticator = authenticator
                self.user, self.auth = user_auth_tuple
                return

        self._not_authenticated()

Add the following code to views.py

import asyncio

from django.http import HttpRequest, HttpResponse
from typing import Iterable, Generator, AsyncGenerator
from rest_framework.views import APIView
from rest_framework import exceptions
from rest_framework.permissions import BasePermission
from rest_framework.throttling import BaseThrottle
from http import HTTPStatus
from .requests import AsyncRequest


class AsyncAPIView(APIView):
    def initialize_request(self, request, *args, **kwargs) -> AsyncRequest:
        """
        Returns the initial request object.
        """
        parser_context = self.get_parser_context(request)

        return AsyncRequest(
            request,
            parsers=self.get_parsers(),
            authenticators=self.get_authenticators(),
            negotiator=self.get_content_negotiator(),
            parser_context=parser_context,
        )

    async def initial(self, request: AsyncRequest, *args, **kwargs) -> None:
        """
        `.dispatch()` is pretty much the same as Django's regular dispatch,
        but with extra hooks for startup, finalize, and exception handling.
        """
        self.format_kwarg = self.get_format_suffix(**kwargs)

        # Perform content negotiation and store the accepted info on the request
        neg = self.perform_content_negotiation(request)
        request.accepted_renderer, request.accepted_media_type = neg

        # Determine the API version, if versioning is in use.
        version, scheme = self.determine_version(request, *args, **kwargs)
        request.version, request.versioning_scheme = version, scheme

        # Ensure that the incoming request is permitted
        await request.authenticate()
        await self.check_permissions(request)
        await self.check_throttles(request)

    def _check_sync_permissions(self, request: AsyncRequest, permissions: Iterable[BasePermission]):
        for permission in permissions:
            if not permission.has_permission(request, self):
                self.permission_denied(
                    request, message=getattr(permission, "message", None), code=getattr(permission, "code", None)
                )

    async def _check_async_permissions(self, request: AsyncRequest, permissions: Iterable[BasePermission]):
        results = await asyncio.gather(
            *(permission.has_permission(request, self) for permission in permissions), return_exceptions=True
        )

        for idx in range(len(permissions)):
            if isinstance(results[idx], Exception):
                raise results[idx]
            elif not results[idx]:
                self.permission_denied(
                    request,
                    message=getattr(permissions[idx], "message", None),
                    code=getattr(permissions[idx], "code", None),
                )

    async def check_permissions(self, request: AsyncRequest) -> None:
        """
        Check if the request should be permitted.
        Raises an appropriate exception if the request is not permitted.
        """
        permissions = self.get_permissions()

        async_permissions, sync_permissions = [], []

        for permission in permissions:
            if asyncio.iscoroutinefunction(permission.has_permission):
                async_permissions.append(permission)
            else:
                sync_permissions.append(permission)

        self._check_sync_permissions(request, sync_permissions)
        await self._check_async_permissions(request, async_permissions)

    def _check_sync_throttles(
        self, request: AsyncRequest, throttles: Iterable[BaseThrottle]
    ) -> Generator[float, None, None]:
        for throttle in throttles:
            if not throttle.allow_request(request, self):
                yield throttle.wait()

    async def _check_async_throttles(
        self, request: AsyncRequest, throttles: Iterable[BaseThrottle]
    ) -> AsyncGenerator[float, None]:
        for throttle in throttles:
            if not await throttle.allow_request(request, self):
                yield throttle.wait()

    async def check_throttles(self, request: AsyncRequest) -> None:
        """
        Check if request should be throttled.
        Raises an appropriate exception if the request is throttled.
        """
        throttle_durations = []
        throttles = self.get_throttles()
        async_throttles = filter(lambda t: asyncio.iscoroutinefunction(t.allow_request), throttles)
        sync_throttles = filter(lambda t: not asyncio.iscoroutinefunction(t.allow_request), throttles)

        throttle_durations.extend(self._check_sync_throttles(request, sync_throttles))
        throttle_durations.extend(
            [duration async for duration in self._check_async_throttles(request, async_throttles)]
        )

        if throttle_durations:
            durations = [duration for duration in throttle_durations if duration is not None]
            duration = max(durations, default=None)
            self.throttled(request, duration)

    async def dispatch(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
        # Note: Views are made CSRF exempt from within `as_view` as to prevent
        # accidental removal of this exemption in cases where `dispatch` needs to
        # be overridden.
        self.args = args
        self.kwargs = kwargs
        request = self.initialize_request(request, *args, **kwargs)
        self.request = request
        self.headers = self.default_response_headers  # deprecate?

        try:
            await self.initial(request, *args, **kwargs)

            # Get the appropriate handler method
            if request.method.lower() in self.http_method_names:
                handler = getattr(self, request.method.lower(), None)
            else:
                handler = None

            if handler is None:
                raise exceptions.MethodNotAllowed(request.method)

            if asyncio.iscoroutinefunction(handler):
                response = await handler(request, *args, **kwargs)
            else:
                raise TypeError("Handler should be async function")

        except Exception as exc:
            response = await self.handle_exception(exc)

        self.response = self.finalize_response(request, response, *args, **kwargs)

        return self.response

    async def handle_exception(self, exc: Exception):
        """
        Handle any exception that occurs, by returning an appropriate response,
        or re-raising the error.
        """
        if isinstance(exc, (exceptions.NotAuthenticated, exceptions.AuthenticationFailed)):
            # WWW-Authenticate header for 401 responses, else coerce to 403
            auth_header = self.get_authenticate_header(self.request)

            if auth_header:
                exc.auth_header = auth_header
            else:
                exc.status_code = HTTPStatus.FORBIDDEN

        exception_handler = self.get_exception_handler()

        context = self.get_exception_handler_context()

        response = exception_handler(exc, context)

        if response is None:
            self.raise_uncaught_exception(exc)

        response.exception = True
        return response

Now that we have created the utility classes to make async API calls. Lets modify the clients/views.py.

First, we will modify ClientsView to inherit from AsyncAPIView class that we defined in drfutil/views.py instead of DRF default APIView class.

We will change the authentication_classes and permission_classes to use the classes that we defined in drfutl/views.py as follows:

authentication_classes = [AsyncAuthentication, ]
permission_classes = [AsyncIsAuthenticated, ]

Now we can change the post and get methods in ClientsView to be async by just prefixing them with the key word async.

Every method that might block - such as get() or delete() - has an asynchronous variant (aget() or adelete()), and when you iterate over results, you can use asynchronous iteration (async for) instead. We are going to change a few places in the code to make async ORM calls using this pattern. Your clients.views.py should now look like this:

from rest_framework import status
from rest_framework.response import Response
from clients.models import Client
from drfutil.authentication_class import AsyncAuthentication, AsyncIsAuthenticated
from drfutil.views import AsyncAPIView

class ClientsView(AsyncAPIView):
    authentication_classes = [AsyncAuthentication, ]
    permission_classes = [AsyncIsAuthenticated, ]

    async def get(self, request):
        data = Client.objects.filter().values()
        return Response(data, status=status.HTTP_200_OK)

    async def post(self, request):
        if not request.data:
            return Response({"error": "Invalid request data"}, status=status.HTTP_400_BAD_REQUEST)
        try:
            first_name = request.data["first_name"]
            last_name = request.data["last_name"]
            email = request.data["email"]

            await Client.objects.acreate(
                first_name = first_name,
                last_name = last_name,
                email = email
            )
            return Response({"message": "Client successfully created!"}, status=status.HTTP_201_CREATED)

        except Exception as e:
            print("error: ", e)
            return Response({"error": "Could not create client."}, status=status.HTTP_400_BAD_REQUEST)

Code description

Let's dive a bit deeper into the last part where we modify the views to make them asynchronous and utilise the AsyncAPIView class along with asynchronous ORM calls.

In the provided code, you'll notice that we've created a new class named ClientsView within the clients/views.py file. This class is responsible for handling the API endpoints related to clients. We're going to modify the methods within this class to make them asynchronous and utilise the utility classes from drfutil.

Async Authentication and Permissions:

In the ClientsView class, you'll see that we're using the AsyncAuthentication class and AsyncIsAuthenticated class for authentication and permissions. These classes are defined in drfutil/authentication_classes.py. By using these async authentication and permission classes, we enable the API views to work asynchronously.

Async Methods:

The methods get and post in the ClientsView class are the key focus. We prefix them with the async keyword to make them asynchronous. Here's what we do with each method:

async def get(self, request):
In the get method, we make a call to the ORM using Client.objects.filter().values(). There isn't a separate asynchronous version of the filter() method because it is non-blocking.

In the get method, we utilise the ORM through the Client.objects.filter().values() query. It's important to note that the filter() method inherently operates in a non-blocking manner, meaning it doesn't require a separate asynchronous version. This design ensures that other concurrent operations can seamlessly proceed without hinderance.
async def post(self, request):

For the post method, we first validate the request data and then create a new client asynchronously in the database using await Client.objects.acreate(...). This ensures that the client creation process doesn't block the event loop.

AsyncAPIView:

The ClientsView class inherits from the AsyncAPIView class defined in drfutil/views.py. This custom class is responsible for initialising the request object in an asynchronous manner and handling various aspects like authentication, permissions, and throttling asynchronously.

Conclusion

In conclusion, Django's recent introduction of asynchronous support has opened up exciting possibilities for building more responsive and scalable web applications. While the core features of Django are still primarily synchronous, we've explored how to extend this support to asynchronous API views using the Django Rest Framework (DRF).

By leveraging utility classes provided in our drfutil package, we've demonstrated how to seamlessly integrate asynchronous functionality into your DRF-powered APIs. This transition allows your application to handle multiple requests simultaneously, enhancing user experiences and maximising server resources.

Let's consider a practical example to illustrate the power of asynchronicity. Imagine you need to send confirmation emails to users after they complete a certain action on your website. Traditionally, sending emails is a blocking operation that could slow down the user experience. However, with asynchronous programming, you can use a function like sync_to_async to convert a blocking operation, such as sending an email, into an asynchronous one.

Here's a glimpse of how you could use sync_to_async:

from asgiref.sync import sync_to_async
import asyncio

def sendEmail(subject, body, email):
    # Your email-sending code here

# Convert the blocking sendEmail function to asynchronous
asend_mail = sync_to_async(sendEmail, thread_sensitive=False)

async def post(request):
    # Your code before sending the email...

    # Send the email asynchronously using asend_mail
    asyncio.create_task(asend_mail(subject, body, email))

    # Your code after sending the email...

By applying sync_to_async to the sendEmail function, you've transformed a potentially blocking operation into an asynchronous one. This enhances the efficiency of your application by ensuring that the email-sending process doesn't hinder other concurrent tasks and operations.

This approach is particularly valuable when dealing with time-consuming tasks like sending emails, as it enables your application to maintain responsiveness and handle multiple operations efficiently.

In this article, we've explored how to extend Django's synchronous framework into the realm of asynchronous programming, focusing on implementing asynchronous API calls using the Django Rest Framework. We've showcased how you can harness the power of sync_to_async to manage potentially blocking tasks like sending emails in an asynchronous context.

To see the complete project and delve deeper into the code, you can access the full example on my GitHub repository here.

I encourage you to share your thoughts and experiences in the comments section below. Have you encountered challenges or successes while transitioning to asynchronous programming with Django? What other aspects of Django's ecosystem would you like to see explored in future articles? Your insights are valuable and can contribute to a vibrant community discussion. Happy coding!

Blog

Async API Calls Unleashed: Exploring Django 4 and Django Rest Framework

Dominic Chingoma

Project Structure

Async Django Rest Framework

Adding code to the files

Code description

Conclusion

Join Our Newsletter. No Spam, Only the good stuff.

Related