Leveraging Starlette in Django Applications.

Since the introduction of asyncio in Python 3.5+, a large number of async web frameworks in python has been on the rise, some of which include but not limited to

1. Sanic
2. aiohttp
3. Quart
4. Vibora

The main advantages of these async web frameworks are that they can handle a larger number of requests and are generally faster than their regular sync counterparts.

Something I struggled with, was how to integrate usage of any of these web frameworks into my current existing application written in Django so that I could benefit from these speed improvements.

At the same time, I wasn't interested in having to rewrite every aspect of my application from scratch and was still interested in using the Django ORM with whatever solution I came up with because it is the easiest ORM I know in python.
I have been actively following the development of Django Channels, an attempt that takes Django and extends its abilities beyond HTTP - to handle WebSockets, chat protocols, IoT protocols, and more. I was hopeful and optimistic about what possibilities such development would bring for Django developers. But I never sold a 100% into using it for my projects mostly because it required the usage of an external dependency on Redis and running into building twisted on Windows, during development.

I eventually found out about Starlette, a lightweight ASGI framework/toolkit, which is ideal for building high-performance Asyncio services. I came to know about it from following the development of Uvicorn, a lightning-fast ASGI server. After going through the documentation, I could immediately see how Starlette could be introduced into my current workflow without having to do so much work. What made Starlette stand out, in my opinion, was the fact that it was advertised as a toolkit. This meant you could use some of its functions in an existing project or as a standalone web framework.
Going through the documentation, the feature, which lit the bulb for me, was the support for Background Tasks. If you have ever done any form of background processing in Django before, you know you need a third party library like Celery or RQ along with either [RabbitMQ][RabbitMQ] or Redis as a message broker. Now I get to have this same functionality without the extra dependencies.

NB: Bear in mind that for long running background tasks and periodic actions, Celery might be a better fit. In this scenario, It was mostly overkill

The next step was to find a problem to force my usage of Starlette. I ran into a scenario where I was experiencing errors to Quickbooks API when trying to generate a sales receipt after payment had been made. Since this action wasn't exactly required before a response was sent back to the client, It was a good candidate as a background task.

The offending code is shown below

def verify_payment(request, order):
    amount = request.GET.get("amount")
    txrf = request.GET.get("trxref")
    paystack_instance = PaystackAPI()
    response = paystack_instance.verify_payment(txrf, amount=int(amount))
    if response[0]:
        p_signals.payment_verified.send(
            sender=PaystackAPI, ref=txrf, amount=int(amount), order=order
        )
        return JsonResponse({"success": True})
    return JsonResponse({"success": False}, status=400)

and the signal callback was this

@receiver(p_signals.payment_verified)
def on_payment_verified(sender, ref, amount, order, **kwargs):
    record = UserPayment.objects.filter(order=order).first()
    record.made_payment = True
    record.amount = Decimal(amount) / 100
    record.save()
    # process to quickbooks
    record.create_sales_receipt()
    record.add_to_mailing_list()

The way the view is currently structured, until the sales receipt is created. Ideally, once the payment has been verified, a the remaining action could be run in the background.

I created a sample Starlette view to replicate the functionality provided by Django

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.exceptions import ExceptionMiddleware
from starlette.background import BackgroundTask
from starlette.middleware.cors import CORSMiddleware

app = Starlette()
app.add_middleware(CORSMiddleware, allow_origins=['*'])
...

@app.route("/paystack/verify-payment/{order}/")
async def paystack_verify_payment(request, order):
    response = await services.verify_payment(request.query_params, order)
    if response[0]:
        task = BackgroundTask(process_payment, request.query_params, order)
        return JSONResponse({"success": True}, background=task)
    return JSONResponse({"success": False}, status_code=400)

My services.py consisted of the following.

from asgiref.sync import sync_to_async

import django

django.setup()
...
@sync_to_async
def verify_payment(request, order):
    amount = request.get("amount")
    txrf = request.get("trxref")
    paystack_instance = PaystackAPI()
    return paystack_instance.verify_payment(txrf, amount=int(amount))

like I mentioned earlier, I wasn't interested in having to change most of the application. Libraries like asgiref helped a lot in translating synchronous functions into a coroutine that can be awaited. Also since I was going to be using the Django ORM outside of Django i.e Starlette, I need to add django.setup() before making any Django model specific call.

The process_payment background task also needed to be a coroutine function

async def process_payment(request, order):
    services.process_paystack_payment(request, order)

The last piece of the puzzle was ensuring that other views were being served by Django while this specific view was been served by Starlette. Digging through the Starlette source code, I came across the following middleware starlette.middleware.wsgi.WSGIMiddleware. I was able to deduce from this that it had the capability of wrapping a WSGI application as an ASGI application. Since Starlette could easily compose different ASGI apps as dedicated routes, I ended up with the following helper function

from starlette.applications import Starlette
from starlette.middleware.wsgi import WSGIMiddleware
from starlette.routing import Router, Path, PathPrefix


def create_asgi_app():
    return Starlette()


def create_app(application, wsgi=False):
    if wsgi:
        return WSGIMiddleware(application)
    return application


def initialize_router(apps):
    return Router(
        [PathPrefix(x["path"], app=create_app(x["app"], x.get("wsgi"))) for x in apps]
    )

Finally, I created an entry point python file that would be referenced by Uvicorn, the ASGI web server

from payment_service.wsgi import application
from v2 import app as asgi_app
from cv_utils.starlette import initialize_router

app = initialize_router(
    [{"path": "/v2", "app": asgi_app}, {"path": "", "app": application, "wsgi": True}]
)

The wsgi instance for a Django application can be found in the wsgi.py file

The Starlette ASGI app would handle any request on the /v2 path, while every other routes would be dispatched to the Django app.

Since it was reported in the Starlette docs, that gunicorn, a battle tested application server in python could play well with Uvicorn, I ended up changing my dockerfile configuration to the following

CMD gunicorn --workers=4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:5000 run:app --access-logfile -

I also needed to upgrade my automated test for this particular view.

the old implementation looked like this

    ...
    @mock.patch("payment_service.models.m_send_mail")
    @mock.patch("requests.post")
    @mock.patch("payment_service.models.QuickbooksAPI")
    @mock.patch("payment_service.urls.PaystackAPI")
    def test_payment_made_with_pastack(
        self, mock_paystack, mock_q_books, mock_post, mock_mail
    ):
        mock_instance = self.get_mock(mock_paystack, (True, "verification successful"))
        mock_quickbooks = mock_q_books.return_value
        mock_quickbooks.create_customer.return_value = {
            "id": "23",
            "name": "Danny Novaka",
        }
        mock_quickbooks.create_sales_receipt.return_value = "2322"
        ...
        with self.env:
            response = self.client.get(
                "/paystack/verify-payment/ADESFG123453/",
                {"amount": 2000 * 100, "trxref": "freeze me"},
            )
            self.assertEqual(response.status_code, 200)
            self.assertEqual(response.json(), {"success": True})
            mock_instance.verify_payment.assert_called_once_with(
                "freeze me", amount=200000
            )
            record = models.UserPayment.objects.first()
            self.assertTrue(record.made_payment)
            extra_data = record.extra_data
            self.assertEqual(
                extra_data["quickbooks_customer_details"],
                {"id": "23", "name": "Danny Novaka"},
            )
            self.assertEqual(extra_data["quickbooks_receipt_id"], "2322")
            mock_quickbooks.create_customer.assert_called_once_with(
                **{
                    ...
            )
            mock_quickbooks.create_sales_receipt.assert_called_once_with(
                {"id": "23", "name": "Danny Novaka"},
                {
                    "currency": record.currency,
                    "description": record.description,
                    "price": record.price,
                    "amount": record.price,
                    "discount": 0,
                },
            )
            mock_post.assert_called_once_with(
                settings.AUTH_ENDPOINT + "/save-custom-data",
                json={
                    "user_id": record.user,
                    "data": {
                        "quickbooks_customer_details": {
                            "id": "23",
                            "name": "Danny Novaka",
                        }
                    },
                },
            )
            mock_mail.assert_called_once_with(
                "first_resume_download",
                {"first_name": "Danny"},
                [record.extra_data["email"]],
            )

Because of the level of mocking involved, I needed to install pytest,pytest-django and pytest-mocker to aid me in writing the equivalent test in Starlette

@pytest.mark.django_db
def test_post(mocker, monkeypatch):
    mock_paystack = mocker.patch("payment_service.services.PaystackAPI")
    mock_mail = mocker.patch("payment_service.models.m_send_mail")
    mock_post = mocker.patch("requests.post")
    mock_q_books = mocker.patch("payment_service.models.QuickbooksAPI")
    monkeypatch.setenv("PAYSTACK_SECRET_KEY", "MY-SECRET-KEY")
    mock_instance = get_mock(mock_paystack, (True, "verification successful"))
    mock_quickbooks = mock_q_books.return_value
    mock_quickbooks.create_customer.return_value = {"id": "23", "name": "Danny Novaka"}
    mock_quickbooks.create_sales_receipt.return_value = "2322"

    ...
    client = TestClient(app)
    response = client.get(
        "/v2/paystack/verify-payment/ADESFG123453/",
        params={"amount": 2000 * 100, "trxref": "freeze me"},
    )
    assert response.status_code == 200
    assert response.json() == {"success": True}
    mock_instance.verify_payment.assert_called_once_with("freeze me", amount=200000)
    record = models.UserPayment.objects.first()
    assert record.made_payment == True
    extra_data = record.extra_data
    assert extra_data["quickbooks_customer_details"] == {
        "id": "23",
        "name": "Danny Novaka",
    }

    assert extra_data["quickbooks_receipt_id"] == "2322"
    mock_quickbooks.create_customer.assert_called_once_with(
        **{
            "email": record.extra_data["email"],
            "full_name": f"{record.extra_data['first_name']} {record.extra_data['last_name']}",
            "phone_number": record.extra_data["phone_number"],
            "location": {
                "country": "NG",  # ensure country comes in full version e.g Nigeria
                "address": record.extra_data["contact_address"],
            },
        }
    )
    mock_quickbooks.create_sales_receipt.assert_called_once_with(
        {"id": "23", "name": "Danny Novaka"},
        {
            "currency": record.currency,
            "description": record.description,
            "price": record.price,
            "amount": record.price,
            "discount": 0,
        },
    )
    mock_post.assert_called_once_with(
        settings.AUTH_ENDPOINT + "/save-custom-data",
        json={
            "user_id": record.user,
            "data": {
                "quickbooks_customer_details": {"id": "23", "name": "Danny Novaka"}
            },
        },
    )
    mock_mail.assert_called_once_with(
        "first_resume_download", {"first_name": "Danny"}, [record.extra_data["email"]]
    )

I would say it was a pleasant experience using pytest and would use it more in other projects I work on.

The amount of change involved was minimal for the boost in performance to be gained.

Hopefully, this post highlights how you can gradually make your less performant views asynchronous with Starlette.