DEV Community: J.V. Zammit

A minimal Django testing style guide

J.V. Zammit — Fri, 01 Oct 2021 15:15:42 +0000

You started on a MVP. Usage is increasing. The occasional bug.

More users. Suddenly this MVP is not an MVP anymore. A core business process depends on this system your team is maintaining.

You ship a user-reported bug fix. Users report another two. Sounds familiar?

Since it was an MVP, you paid little attention to writing automated tests.

What is the current codebase's test coverage? Wait. Do you even measure it?

This post is not about setting the gold standard on how to write tests for your Django application.

It's about setting a simple direction as possible. And getting started. Wherever you are.

In this I also an attempt to document some of my practices around writing tests for Django projects.

Terminology

As in this Python testing style guide I followed in the past:

I do not make a distinction between unit tests and integration tests. I generally refer to these types of tests as unit tests. I do not go out of my way to isolate a single unit of code and test it without consideration of the rest of the project. I do try to isolate a single unit of behaviour.

Where do I place tests?

The greatest advantage of having tests is that they are living docs about your code. The inputs/outputs tested either work as expected or they don't. They cannot go outdated. Unlike written docs.

So it's important to have an obvious way to tell where the test for a piece of code resides in your codebase.

pytest's conventions for Python test discovery are simple to agree on and follow.

Let's say you have an Article model in app articles at articles/models.py:

class Article(models.Model):
    ...

Tests for this model class would go in articles/tests/test_models.py:

class ArticleTest(TestCase):
    ...

In a similar fashion, if we have a function in articles/utils.py:

def do_something():
    ...

its tests would go in articles/tests/test_utils.py:

class DoSomethingTest(TestCase):
    def test_do_something(self):
        ...

Why create a DoSomethingTest extending a TestCase? Rather than a standalone test function?

The default way to write tests in Django is using Python's built-in unittest module. This defines tests using a class-based approach.

django.test.TestCase is in fact a subclass of unittest.TestCase that runs each test inside a transaction to provide database-oriented goodies. Such as transaction isolation. Django's writing and running tests docs provide more details on this.

Naming test classes

As in the case of "placing" a test in your codebase, a TestCase's name should be indicative of the class (or standalone function) being tested. So for classes:

class Article(model.model)
    ...

class ArticleTest(TestCase)
    ...

and for functions:

def do_something():

...

class DoSomethingTest(TestCase):
    def test_do_something(self):
        ...

Naming test functions

As mentioned above, your individual tests should test for a single unit of behaviour. In the below case, behaviour depends on the data (state) the logic depends on:

class Article(models.Model):
    title = models.CharField(max_length=32)
    slug = models.CharField(max_length=32, unique=True)
    content = models.TextField()

    def __str__(self):
        return self.title

    def get_short_title(self):
        if len(self.title) > 15:
            return self.title[: (15 - 3)] + "..."
        return self.title

So we have to test get_short_title based on the title's length. The more "explicit" the name the better. Have test function names that reflect the state relevant for the expected result:

class ArticleTest(TestCase):
    def test_get_short_title_title_is_longer_than_15_characters_yes(self):
        ...

    def test_get_short_title_title_is_longer_than_15_characters_no(self):
        ...

Notice how the whole class name to function name structure tells us:

name of class Article
name of function get_short_title
state (input) determining behaviour (output) per test

This also makes obvious what should be changed in case you move/rename classes/functions. Drawn out it looks like this:

url="" %}

Below is a screenshot of my setup. With code on the left, and tests code on the right. No fancy screen required. I took this on a laptop screen:

I find this a game changer when reading code. Seeing a function's inputs and expected outputs next to the function's code speeds things up. A lot.

Creating test data

Use factory libraries to standardise test data creation.

The most popular two in Django are factory-boy and model bakery. The latter is simpler, so we use it in this article's examples. But the former is more feature-rich.

As a guiding principle keep your factory code light. I.e. specify only the fields that need to be set for the state required by the test.

For example, when creating the article to test __str__:

article = baker.make("articles.Article", title="Test Title")

we are not passing slug or description. These would be noise for our test's purpose.

The way we create data via factories should be about the data state expected by the test. No more, no less.

Why not just use fixtures? You may already have fixtures, with a lot of data as expected in production, lying around.

Fixtures are an overkill for tests because:

Maintaining them is hell. For example, what if you rename a model field?
Maintaining fixtures is usually append-only. So the fixture's data increases over time, loading a lot of data the test doesn't need, every time the test is run.
Fixtures make it hard to tell what data is being loaded. It's even harder to tell why a fixture is loading the data it is loading in the context of tests!

I find eliminating fixtures in tests is one of the most effective steps to speed up a test suite's runtime.

Not convinced? Section 11.1 in Speed up your Django Tests delves into more detail why fixture files should just be avoided.

Testing views

The above basic examples were all about testing models. Models are the usual starting point in writing a Django app.

But a lot of logic resides in views. And the forms and/or serializers that go with them.

Unit vs integration, again

Django and DRF provide the test client and APIClient to make HTTP calls against your application programmatically.

The problem with making full HTTP calls against your application is that not only your view code wiill run. HTTP requests pass through the whole Django middleware machinery, including, for example, authentication.

So when using the test client, you would be doing more of an integration test, rather than a unit test.

The alternative would be instantiating views. And passing instances of RequestFactory, or APIRequestFactory in DRF, directly to the view functions. Same approach applies for serializer (or form) functions.

Which way do you go? Ask yourself a couple of questions:

Is your current test suite running slow?
Do you have a lot of added middleware a request needs to go through?

If both answers are "no" (especially if you have no tests!) then the test client is likely enough to start with.

The upside of making HTTP requests is taht you are testing the whole URL/view binding. Rather than a standalone class-based view function like get_queryset.

Let's say a FE engineer on your team asks you a question about how a particular URL is supposed to work. Having tests with the test client allows you to be 100% sure of what the view at that URL is supposed to do. This is an example of what I meant above by living docs.

How to structure your view tests

When using the test client I find it helpful to think of my application as a "black box":

Assuming there is this data set up (inputs)
What should happen if I submit this request to this URL? (action)
Assert response returned (output)

The pattern is the one when writing tests for functions. But in this case rather than calling the specific function under test, the test makes an HTTP request.

Let's say we want CRUD views for our Article model. This is what we'd find in articles/serializers.py:

class ArticleSerializer(serializers.ModelSerializer):
    class Meta:
        model = Article
        fields = "__all__"

And in articles/views.py

class ArticleListCreateAPIView(generics.ListCreateAPIView):
    queryset = Article.objects.all()
    serializer_class = ArticleSerializer


class ArticleDetailAPIView(generics.RetrieveUpdateDestroyAPIView):
    queryset = Article.objects.all()
    serializer_class = ArticleSerializer

These are the tests we'd need to write to assert the API behaves as expected:

We do not need to write tests for the serializer, as it is used by the views.

Testing configuration

In many cases, as shown above, and especially when using class-based views, functionality is achieved through configuration of classes rather than coding. ¹

How to test in this case? The quickest way, i.e. the minimal in terms of time, is to test using the client.

Let's extend our view to be able to order articles by title. The easiest option would be to extend ArticleListCreateAPIViewTest.test_list to:

create test data with different titles to be able to assert results order
call the URL with ordering parameter
test outcome

View code:

class ArticleListCreateAPIView(generics.ListCreateAPIView):
    queryset = Article.objects.all()
    serializer_class = ArticleSerializer
    filter_backends = [filters.OrderingFilter]
    ordering_fields = ["title"]

Test code:

class ArticleListCreateAPIViewTest(APITestCase):
    def test_list(self):
        article1 = baker.make("articles.Article", title="Title A")
        article2 = baker.make("articles.Article", title="Title C")
        article3 = baker.make("articles.Article", title="Title B")

        # default ordering by id:
        response = self.client.get("/api/articles/")
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(
            [article["id"] for article in response.data],
            [article1.id, article2.id, article3.id]
        )

        # ordering by title:
        response = self.client.get("/api/articles/?ordering=title")
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(
            [article["id"] for article in response.data],
            [article1.id, article3.id, article2.id]
        )

While testing "code owned by others" these tests ensure that the API provided works as expected.

Diving deeper

This guide is intended to be minimal. As such it misses going into some important aspects of writing tests.

Test runners other than `TestCase`

Django's TestCase extends unittest's TestCase. And DRF's APITestCase extends Django's TestCase. This guide does not go into other test runners like pytest. Which take a more functional approach.

Mocking

What if your code needs to read data from an external API? Or upload a file to S3?

If the test needs to interact with external systems, it would become a "system" test. A test that depends on components "outside" the local environment where it runs.

Your automated tests should always mock such interactions.

Getting into mocking is out of this style guide's scope. The tool Python provides is unittest.mock.

Coverage

On its own, coverage is a "dumb" measure of how many lines in your codebase are "covered by" tests. Why dumb? You can write a bunch HTTP GET requests in your tests, write zero assertions, and increase coverage a lot. That, of course, doesn't add any value.

So do not write tests for coverage.

Rather use coverage.py to identify which lines, i.e. behaviour, your tests do not cover.

Call me naive, but I still strive for 100% coverage.

But how to get started?

In any project/team I work with, I fight for a non-negative diff coverage on every PR on CI. Diff coverage definition:

Diff coverage is the percentage of new or modified lines that are covered by tests. This provides a clear and achievable standard for code review: If you touch a line of code, that line should be covered. Code coverage is every developer’s responsibility!

Over time you'll get to 100% coverage 🚀

Speeding tests up

In this guide we go over creating test data using factories as opposed to fixtures. But that's just it.

In the "views" section I mentioned avoiding rendering responses to only test the behaviour "under test".

This is scratching the surface. Speed Up Your Django Tests by Adam Johnson is the best resource on this topic.

And not only. It is in fact a great resource when starting to write tests from scratch. The below chapters, in particular, are an excellent guide:

Chapter 10, Test Structure, on how to organise your tests' code.
Chatper 11, Test Data, on best practices in managing data for your tests, in the fastest way possible.
Chapter 12, Targetted Mocking, on how to mock behaviour the right way.

References / Further Reading

unittest module - Python's standard library.
Testing in Django - Django docs.
My Python testing style guide - Thea "Stargirl" Flowers. Main inspiration for this article.
Python Style Guide - Octopus Energy. This style guide is a gold mine for any Python project. As in my article's case, no need to treat it as dogma. But it provides insight into how a large organisation, with presumably a large codebase, handles its "style guide". Kudos for sharing!
Why and how I get 100% test coverage for my Django projects, and you should too. Sasha Romjin.
Speed Up Your Django Tests - Adam Johnson. Can't recommend this book enough!

Aka convention over configuration. ↩

Backend Engineer Interview Script

J.V. Zammit — Tue, 08 Jun 2021 09:24:18 +0000

Last month I've written a post with Django-specific interview questions. That post is about verifying a candidate’s depth of knowledge with Django.

This post is about less specific questions I ask a candidate. Before proceeding to the Django specific ones. If I do proceed after all. In case the candidate does not have any Django-specific experience, what's the point?

These questions do not look for specific answers. Rather the aim is to open up a discussion. So the questions are more in the form of how or why rather than what.

And no. How do you describe yourself? is not one of these questions. If you ever get that question, the only correct answer is that by Dwight Schrute as shown above.

The questions in this article revolve around:

improving web application response time
designing authentication
handling asynchronous work / task queues
HTTP REST API design

1. Server response time

Q: In the context a dynamic "data driven" web application, how would you decrease a page's response time?

Asking this should allow the candidate to come up with multiple answers. Among the common answers you should note the below for further discussion:

caching
database query optimisation

Caching

Caching provides a form of "low hanging fruit" to speed up serving a response in web applications.

Q: Where can you cache results in a Web application?

Allow the candidate to go into the multiple levels of caching a Web application offers.

Brownie points for mentioning:

caching HTTP response via HTTP headers, i.e. cache-control
caching response via web server
server-side in memory caching

The cache-control HTTP header allows caching resources across various points on the network Reference docs on HTTP caching.

Web server caching caches resources on the server end of the network. The Nginx cache is a fine example.

Server-side caching using an in-memory cache as memcached or redis is another option.

Caching can also be enabled at "database layer". Through "views" or outright database denormalisation. Ideally we do not go there during this interview.

Caching though comes with its own challenges.

Q: What are the tradeoffs of caching?

Discuss expiration or "freshness" methods. For any of the caching methods the candidates brought up for the previous question.

Q: When would serving stale data be incorrect?

Some scenarios are better suited to caching. Think Wikipedia. Others are not. Think live football scores. How to optimise the latter?

Segway the discussion into query optimisation.

Database Query Optimisation

Q: What makes a query "slow"?

Any query whose result can be retrieved faster.

Brownie points the candidate mentions redundant outer joins. A Visual Explanation of SQL Joins by Jeff Atwood provides a great revision on this topic.

Once the candidate mentions indexes the question is:

Q: How to decide which columns to index?

Brownie points for mentioning:

Column selectivity. For example a gender column with a few values is not a good candidate for an index. Whereas a unique id column is.
The field type. For instance, a numeric integer type is index-friendly. A "text" field type is not. Because it will result in a large index which takes computing resources to scan.

2. Authentication

Q: We have a monolith central web app where users data is stored and managed. And we have several microservices that need to authenticate those users. How would you go about it?

At the time of writing JWT is the way to go.

Still, any answer involving "tokens" and token-based authentication should be good. Such an approach allows not only authenticating requests for the microservices above. It is also a great way to authenticate frontend requests.

3. Task queues and async work

Q: Our users need to generate reports on the fly. But an individual report takes several minutes to generate. How would you structure this in a web application?

Allow the candidate to distinguish between the synchronous part of this feature. And async parts.

One such way would be:

Web request queues task to generate report (syncronhous).
Report generation task writes file to storage, e.g. AWS S3 (async).
Once the task writes the file strage, task sends out email with link to file (async).

This should give a nice opener to discuss task queues with the candidate.

A lot has been said and written about task queues. I have written an article myself: Do you need a task queue in your web app?

Allow the candidate explain how the task queue works in this context:

Q: What are the basic components for async task processing in a web context?

Answer should cover these basic building blocks:

the web server process "producing" tasks
the task queue component managing the tasks' lifecycle
workers "consuming" or "executing" those tasks

Brownie points for getting the basics right.

Do you think the candidate has several years' experience dealing with queues? You can dig deeper. How?

David Yanacek at AWS has written on strategies to deal with queue backlog scenarios. This provides some excellent talking points:

In a queue-based system, when processing stops but messages keep arriving, the message debt can accumulate into a large backlog, driving up processing time. Work can be completed too late for the results to be useful, essentially causing the availability hit that queueing was meant to guard against.

Putting it another way, a queue-based system has two modes of operation, or bimodal behavior. When there is no backlog in the queue, the system’s latency is low, and the system is in fast mode. But if a failure or unexpected load pattern causes the arrival rate to exceed the processing rate, it quickly flips into a more sinister operating mode. In this mode, the end-to-end latency grows higher and higher, and it can take a great deal of time to work through the backlog to return to the fast mode.

The above should motivate the next question:

Q: How do you make sure that tasks are being executed quicker than they are being queued?

There is no one correct answer. The point of this is to allow the candidate to explore alternatives with you. Which is what you'd be doing together in case you go on with the candidate and they join the team.

Brownie points if the candidate mentions making tasks idempotent. I.e. have the task achieve the same result, whether it runs 1 time or x times. This removes a lot of headaches in managing a task queue.

Explore with the candidate how they would go about achieving this.

4. REST API design

Most, if not all, web applications, provide some form of REST API backend component. A "rich" frontend consumes this REST API. Such rich frontend is usually implemented by the "hot" JS framework at the time. At the time of writing, among the hottest we arguably find React and VueJS.

Candidates submit some code as part of the recruitment process. This coding assignment has the candidates add functionality to an existing REST API. I.e. in the form of new URIs.

Based on submitted assignment discuss their choices made in designing URLs:

Endpoint names. Hint: use nouns rather than verbs. And name collections with plural nouns.
Endpoint ability to allow filtering, sorting, and pagination.
Request and response encoding. Hint: use JSON.
Proper usage of HTTP methods or "verbs".

Couple of excellent resources on REST API design best practices:

Best practices for REST API design as written up by StackOverflow engineers.
Best Practices in API Design as documented in Swagger docs.

To Conclude

The "brownie points" above pinpoint some "good answers". If the candidate does not mention those it's not a problem.

We are humans not search engines. We forget stuff. Especially during interviews.

Keep in mind the candidate can answer questions well even when the answer is not the one you expect.

You can use the comments box below to let me know:

about anything I'm missing
how I can improve this interview script

May the interviews you're involved in lead to a better life!

Django Engineer Interview Script

J.V. Zammit — Sat, 22 May 2021 13:29:58 +0000

Unlike Michael Scott above I'm not a manager.

Yet I have worked as a Python/Django backend engineer for more than a decade. This puts me in a "senior" role in any team I usually join.

As a result, I get involved in the recruitment process. I.e. to look for someone who would then become a team mate.

This post is about questions I ask a prospective Python/Django backend engineer. A prospect with 2+ years of experience doing Python/Django backend development full-time.

The prospective engineer would be working in my team. With me. Not for me.

So the questions aim to help me to answer the fundamental "Would I want to work with this person?" question.

Below I go through what I look for in each question. Note that unlike a "hackerrank" style assessment, these questions look for qualitative attributes.

Answers do not have a completely correct/wrong answer. The way in which the candidate answers is more important than the answer itself.

Update 2021-06-06. These questions are Python/Django-specific. They intend to center discussion about Django-specific knowledge and experience. I have written another backend engineer interview script which is technology agnostic.

Self-intro

Allow the candidate to go over their career trajectory. Have them delve deep into some projects they are proud to have been part of. Allow them to expand on specific processes/tools used.

This will provide you with loose ends to relate the below questions with.

Q1. How do you use the _ character in Python?

Objective is to see how the candidate reacts.

Among the correct answers:

Used in snake case (i.e. function names and variable names)
Used to store a function return value which is going to go unused.
Used to "show" that a function is meant to be used as a "private" function.

Q1 Follow-up: in case "private function" use case is brought up

The followup is a trick question:

How does Python enforce private properties/functions?

The answer is: It does not.

Probe the candidate to make sure she knows what she'd use it for, in Python.

Look for honesty in the candidate's answer. An "I don't know" is more than acceptable. Better than making stuff up.

Q2. Which test data generation toolkit have you used?

Usually the candidate will have mentioned tests in CV or during their own self-intro. It's hard to have good tests without proper management of test data.

This is an open ended question about how the candidate handled test data. Answer demonstrates whether the candidate has written tests or not. Look for familiarity with tools in the Python/Django testing landscape.

Q2 Follow-up: How to improve unit tests speed in Django?

A mature application is one that grows beyond that ideal “micro service” or “side project” size. Along size and complexity, one common aspect is tests taking longer to run.

Adam Johnson has written a whole Speed Up Your Django Tests book on topic. Which I fully recommend.

Do not look for that level of detail though. A couple “quick ways” to improve speed is fine. I.e. low-hanging fruit. Examples:

Avoid using fixtures in tests.
Avoid creating more test data than the test requires.
Change the default Django password hasher. Etc.

The way the candidate answers this shows their “battle scars” in writing tests for Django applications.

Also look out for “quick hacks”. For example, changing the tests to run against sqlite rather than say Postgres or MariaDB. This should not be one of those “quick ways”.

Be ready, especially when it comes to allocated time, to discuss any disagreement. Not only in this question. If the candidate doesn't make it to the next round, they still would have learnt something.

Q3. Which packages you "carry around" in your Django projects?

Djangonauts tend to have some favourite frameworks that they end up using in all their Django projects.

Popular answers include DRF, django-extensions, anymail, allauth and the DJDT.

Q3 Follow-up: How do you decide whether to use a package or reinvent the wheel?

This again is an open ended question. Aim is to see the candidate's experience in making this sort of decision when building out a project. Good considerations include:

Does this package actually fit the use case? Or at least, is it very close to the use case?
Set up/installation complexity, especially given architecture of existing application. And other installed moving parts which might not play well with the package.
Security concerns.
Activity around package on Github, i.e. recent commits, contributors, responses to issues, etc.

Q4. Which code standards do you follow for a Django project? Tools?

Ideally any of these are mentioned: PEP 8, flake8, black, Django style guide, isort.

There is no correct answer. Just checking for "awareness".

Keeping in mind for example that black and Django's own style guide are not 100% compatible. The specifics are subjective.

Look for familiarity with a tool or a couple of them.

Q5. Which deployment tools have you used? To ship code to production?

Probe the candidate to know more whether they have been exposed to “ops” kind of work. Or at least gauge whether there’s an interest.

Feel free to discuss tools used at your current post, example AWS, Ansible, Python's Fabric library, Docker for local development, etc.

The motivation behind Q5 is that I work in environments where some “ops” skills alongside Python/Django skills are appreciated.

But you might be looking for a full-stack person. Then replace this question with another which shows the person's reaction about some FE-oriented work aspect.

On open-ended questions as these the candidate being opinionated is not a bad thing.

Discussing topics you do not have full agreement on is revealing. For both sides of the interview process. Which is what you want.

To conclude

Remember to encourage the candidate to ask questions throughout. This again enables the two-way information street we want.

I hope you find these useful.

Are there any other questions I should ask? Why? Feel free to add your suggestions as a comment.

This post was inspired by Jacob Kaplan-Moss's article: My interview kickoff script, annotated.

I suggest you give that article a look. Tread carefully. You might find yourself down in Jacob's interviewing series rabbit hole. Some excellent nuggets of knowledge shared down there.

A minimal Websockets setup with Django in production

J.V. Zammit — Tue, 04 Aug 2020 09:47:44 +0000

Use Case

I have a Handsontable implementation for an underlying database table. I.e. a “JavaScript data grid that looks and feels like a spreadsheet”.

A requirement came up. Obvious in hindsight.

Changes made to cells in one sheet should be reflected in the same sheet for other users. When the sheet is open in other browser tabs/windows.

This calls for "server side push" using web sockets. I.e. the server needs to push a notification to open "clients".

Another way to do this would be to have client browsers Ajax-polling for changes. But that would be wasteful. Let's only update the sheet when valid changes are saved!

What does minimal mean in this case?

This application is used by a team internally. Usage not exceeding ten concurrent users. The implications of this:

One process to serve web socket requests is enough.
No real performance testing was done.
No consideration of alternatives to daphne such as uvicorn or starlette. I picked up daphne because it came up "first on the list of alternatives". That's it!
No need to handle websocket interactions asynchronously in my case. You can read more about going sync vs async with websockets in Django channels here.

This configuration remains unchanged until problems crop up. Because “premature optimisation is the root of all evil”.

Blueprint: Before & After

Note: http request protocol below refers both to http and https. Same applies to ws and wss. Assume that for local development the protocol is unsecure, whilst secure in production.

Before introducing websockets, the web browser made an http request to Nginx. At this point Nginx serves the request using gunicorn, hitting Django¹.

After adding websockets in the mix, Nginx still serves http requests. But it's now able to serve ws requests by talking to daphne. In this case you can replace daphne with any other Websocket termination server:

The "new" item in the building blocks above is therefore daphne:

Daphne is a HTTP, HTTP2 and WebSocket protocol server for ASGI and ASGI-HTTP, developed to power Django Channels.

It supports automatic negotiation of protocols; there's no need for URL prefixing to determine WebSocket endpoints versus HTTP endpoints.

The daphne component can be replaced with alternatives as uvicorn or starlette.

The other item of note is that ws:// connections are an "open" connection. "Data" travels in both directions along the same socket. As opposed to http://.

Code changes

daphne is part of the Django Channels effort:

Channels augments Django to bring WebSocket, long-poll HTTP, task offloading and other async support to your code, using familiar Django design patterns and a flexible underlying framework that lets you not only customize behaviours but also write support for your own protocols and needs.

I've followed the docs. The installation and excellent tutorial sections helped me get everything to work locally.

For completeness' sake, these code changes are for an installation using these package versions:

channels==2.4.0
channels-redis==3.0.1
Django==3.0.8
redis==3.5.3

in a Python 3.7.5 virtualenv.

The changes needed:

settings.py changes. These:
- add channels to INSTALLED_APPS, and
- configure channels to route websocket requests to the main channels entrypoint
Routing code changes:
- main project-level routing entrypoint
- app level entrypoint(s), in this example using just one example myapp as app
The consumer that hosts all the event-handling and message sending logic our app needs to implement.

1. settings module changes

Added channels as the first app in my project's list of INSTALLED_APPS. Why first?

Please be wary of any other third-party apps that require an overloaded or replacement runserver command. Channels provides a separate runserver command and may conflict with it. An example of such a conflict is with whitenoise.runserver_nostatic from whitenoise. In order to solve such issues, try moving channels to the top of your INSTALLED_APPS or remove the offending app altogether.

Then added this new setting for channels app to use:

# CHANNELS
ASGI_APPLICATION = 'proj.routing.application'
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            'hosts': [('127.0.0.1', 6379)],
        },
    },
}

Note that I already had redis installed for caching and the application's existing task queue with Huey.

2. Routing changes

I usually call the "default" app proj. This makes it obvious that the app is a container for project-wide items. As is the case with this new routing module. It contains the ProtocolTypeRouter that serves as main entry point for the ASGI application. proj/routing.py contents:

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
import myapp.routing

application = ProtocolTypeRouter({
    # (http->django views is added by default)
    'websocket': AuthMiddlewareStack(
        URLRouter(
            myapp.routing.websocket_urlpatterns
        )
    ),
})

myapp is the test app used for this exmaple. The top-level router above contains a reference to a myapp.routing module. This URLRouter routes http or websocket type connections via their HTTP path. myapp/routing.py contains the below:

from django.urls import re_path

from myapp import consumers

websocket_urlpatterns = [
    re_path(r'ws/sheet/(?P<sheet_name>\w+)/$', consumers.SheetConsumer),
]

Note how channels allows us to structure our web socket URLs in the already familiar format we're used for standard urls.py configuration².

3. The consumer

The final module that needs adding is the consumer. In channels, consumers:

Structure your code as a series of functions to be called whenever an event happens, rather than making you write an event loop.

Allow you to write synchronous or async code and deals with handoffs and threading for you.

myapp/consumers.py implements a SheetConsumer class which extends WebsocketConsumer:

import json
from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer


class SheetConsumer(WebsocketConsumer):

    def connect(self):
        self.sheet_name = self.scope['url_route']['kwargs']['sheet_name']
        self.sheet_group_name = 'sheet_%s' % self.sheet_name

        # Join sheet group
        async_to_sync(self.channel_layer.group_add)(
            self.sheet_group_name,
            self.channel_name
        )
        self.accept()

    def disconnect(self, close_code):
        # Leave sheet group
        async_to_sync(self.channel_layer.group_discard)(
            self.sheet_group_name,
            self.channel_name
        )

    # Receive message from WebSocket
    def receive(self, text_data):
        text_data_json = json.loads(text_data)

        # Send sheet_name to sheet group
        async_to_sync(self.channel_layer.group_send)(
            self.sheet_group_name,
            {
                'type': 'refresh_sheet',
                'sheet_name': text_data_json['sheet_name'],
                'object_id': text_data_json['object_id'],
                'column_index': text_data_json['column_index'],
                'new_value': text_data_json['new_value'],
                'broadcaster_id': text_data_json['broadcaster_id'],
            }
        )

    # Receive message from sheet group
    def refresh_sheet(self, event):
        # Send sheet_name to WebSocket
        self.send(text_data=json.dumps({
            'sheet_name': event['sheet_name'],
            'object_id': event['object_id'],
            'column_index': event['column_index'],
            'new_value': event['new_value'],
            'broadcaster_id': event['broadcaster_id'],
        }))

The above is based on the Write your first consumer tutorial section. Instead of chat messages, the data is about a sheet's cell updates. Updates that need to be applied for the same sheet open in other browser tabs/windows.

I'm passing the user ID of the authenticated user in broadcaster_id. To be able to tell which user "triggered" the websocket message being "broadcasted".

Give it a try locally

This is another great feature of channels. Not even your usual manage.py runserver workflow needs to change. Just note the new item in your default runserver output when it starts:

$ ./manage.py runserver
Watching for file changes with StatReloader
Performing system checks...

System check identified no issues (0 silenced).
August 01, 2020 - 16:07:41
Django version 3.0.8, using settings 'proj.settings.local'
Starting ASGI/Channels version 2.4.0 development server at http://127.0.0.1:8000/    << THIS!
Quit the server with CONTROL-C.

The socket handshakes are also shown in the output:

WebSocket HANDSHAKING /ws/sheet/sheet1/ [127.0.0.1:65181]
WebSocket CONNECT /ws/sheet/sheet1/ [127.0.0.1:65181]

Awesome! Let's deploy!

Deployment notes

Not so fast 😊

I followed the channels documentation here alongside Django's own documentation on deploying ASGI applications here. But I applied ~~two~~ three tweaks that I'd rather explain.

daphne command tweak

I experienced the CRITICAL Listen failure: [Errno 88] Socket operation on non-socket exception described here.

This was fixed by following the suggestion to remove the -fd 0 switch. This switch is suggested by default in the channels docs.

In the current use case I do not need to use this switch. Because I do not need to bind multiple Daphne instances to the same port my production instance. In case I do, I will need to change my structure (see next section) to have daphne called directly from supervisor. Rather than via bash script.

asgi.py tweak

I had implemented proj/asgi.py as described in the channels docs here. This works fine locally. But it led to this exception described here when executing web socket requests in production. I changed the proj/asgi.py as described in this stackoverflow answer which makes it have this content:

import os
import django
from channels.routing import get_default_application

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

django.setup()

application = get_default_application()

This change replaces usage of django.core.asgi.get_asgi_application with channels.routing.get_default_application. The stackoverflow answer above is supported as per channels' docs here.

I do not know whether this fixes things properly. Should I should have done something else? It appears to be a small incompatibility between channels and Django docs. channels docs suggest creating asgi.py from scratch. While Django 3.0.8 auto-created asgi.py.

If you have a better resolution to this please let me know (comment below).

redis version

Recall that my configuration is using channels_redis as backing store.

Since my production application runs on Ubuntu 18.04 LTS, default apt-get redis version was 4.0.1.

This resulted in this weird BZPOPMIN - "ERR unknown command 'BZPOPMIN'" error. This is because redis version 5 or higher is needed.

Therefore please upgrade redis for your configuration. In my case I've followed the quickstart docs, especially the "Installing Redis more properly" section.

On to the production config files!

Deployment - resulting configuration

My configuration's components:

An executable bash script runs daphne. I use this to be able to run and test daphne directly in the Django project's virtualenv.
A supervisor conf file to have this bash script process managed by supervisor.
Nginx, of course.

Bash script

start_daphne.bash contents. Remember to chmod +x your bash script.

#!/bin/bash

NAME="myproject-daphne"  # Name of the application
DJANGODIR=/home/ubuntu/webapp/myproject/proj  # Django project directory
DJANGOENVDIR=/home/ubuntu/webapp/myprojectenv  # Django project env

echo "Starting $NAME as `whoami`"

# Activate the virtual environment
cd $DJANGODIR
source /home/ubuntu/webapp/myprojectenv/bin/activate
source /home/ubuntu/webapp/myproject/proj/.env
export PYTHONPATH=$DJANGODIR:$PYTHONPATH

# Start daphne
exec ${DJANGOENVDIR}/bin/daphne -u /home/ubuntu/webapp/myprojectenv/run/daphne.sock --access-log - --proxy-headers proj.asgi:application

Supervisor

File located at: /etc/supervisor/conf.d/daphne.conf. Remember to create the log file directories.

; ================================
;  daphne supervisor
; ================================

[program:daphne]
command = /home/ubuntu/webapp/start_daphne.bash  ; Command to start app

user = ubuntu   ; User to run as
numprocs=1

autostart=true
autorestart=true

redirect_stderr=true
stdout_logfile = /home/ubuntu/webapp/logs/daphne/access.log  ; Where to write access log messages
stderr_logfile = /home/ubuntu/webapp/logs/daphne/error.log  ; Where to write error log messages
stdout_logfile_maxbytes=50MB
stderr_logfile_maxbytes=50MB
stdout_logfile_backups=10
stderr_logfile_backups=10
environment=LANG=en_US.UTF-8,LC_ALL=en_US.UTF-8  ; Set UTF-8 as default encoding

Nginx

Nginx configuration references I used: channels deployment docs and this answer on stackoverflow. Follow those links to understand what I did.

Relevant Nginx config contents:

upstream ws_server {
  server unix:/home/ubuntu/webapp/myprojectenv/run/daphne.sock fail_timeout=0;
}

upstream gunicorn_server {
  server unix:/home/ubuntu/webapp/myprojectenv/run/gunicorn.sock fail_timeout=0;
}

...

server {
  ...

  location /ws/ {
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_redirect off;
    proxy_pass http://ws_server;
  }

  location / {
    ...

    if (!-f $request_filename) {
        proxy_pass http://gunicorn_server;
        break;
    }
  }
}

Note the newly-added ws_server-related parts.

A note Markup/Javascript code

This is not configuration as such. But as you can see the whole tutorial did not tackle ws and wss usage. One reason is that in this project's case the SSL certificate part is not handled by Nginx. Since the project is using Cloudflare SSL, Cloudflare takes care of it even "before Nginx".

The only ws vs wss logic I have is done at client-side level. This allows the same code to use the correct protocol locally and in production. The code sets up the connection depending on the current http protocol in use:

...

if (window.location.protocol == 'https:') {
  wsProtocol = 'wss://'
} else {wsProtocol = 'ws://'}

sheetSocket = new WebSocket(
  wsProtocol + window.location.host
  + '/ws/sheet/' + sheetName + '/'
);

...

Conclusion

Please let me know (in the comments below) whether anything I've done is wrong or I can improve it.

This was my first experience with Websockets and Django together. And it was pleasant one. The few "conflicting docs" issues described above, although blocking, were kinda expected.

Credits: Diagram above drawn using excalidraw.com.

In practice, in some cases, Nginx is able to serve the request itself. Example, for static files or resources cached an Nginx level. Avoiding going into this to keep things simple. ↩
At the time of writing re_path() is used due to limitations in URLRouter. ↩

Huey as a minimal task queue for Django

J.V. Zammit — Wed, 08 Jul 2020 10:23:27 +0000

Are you considering adding a task queue to your Django project? Then this article should be useful to you.

When I was "younger" task queue with Django project meant celery task queue.

Now that I'm "older" there are simpler alternatives. The simplest I found was Huey. But the ideas presented here apply to evaluating all task queues for your Django project.

Background

Frustrated with celery and django-celery

In December 2019 I was taking a Django project from Python 2 to 3. This project relied on celery and its integration for Django for asynchronous task processing. Github project link here. This codebase's work was mostly done back in 2012-2015.

django-celery was the first problematic part. "Problematic" because it was not being actively maintained. Python 3 support was only "planned" at the time.

Besides, by testing celery with my Python3 setup I realised how "heavy" it is. "Heavy" in terms of dependencies. billiard, kombu, etc.

I had used celery for a very long time. A decade! But this nudged me into looking for alternatives.

My use cases for using an async task queue were:

Handle tasks async. For example, sending an email out. Tasks that you should not handle within the request-response cycle.
Scheduled tasks.
Retrying of failed tasks. For example, reading from an API fails due to a network error. I want that task retried for a few times.
Simple locking. More on what I mean by "simple" further down.

The above were (are) handled nicely by celery. The scheduled tasks part relied entirely on django-celery.

Another factor that pushed me "off the celery train" was something in my last long-term gig. The increasing reputation that celery is "heavyweight".

Finally, celery provides a whole lot more than the above basic set of use cases I need.

Is celery heavyweight?

A colleague achieved significant gains in task execution time by moving off celery. To dramatiq. This was on a Python 3 project I didn't work directly on. By significant I mean ~50% throughput. In this case, “tasks” were about handling a simple message that wrote one row to the database, at most. With no real processing of that message.

At about the same time the above happened, I was listening to the DjangoChat podcast.

The below is a transcript from the “Caching” episode from November 2019. Transcript here. The podcast folks were discussing caches. And brought up the usual suspects; Memcached and Redis. On mentioning Redis, Carlton Gibson calls out how easy it is to add a queue when you have Redis in place. And how much of an “overkill” celery is. Emphaisis in the below quote is mine:

have Redis? Yeah. You want to you want to use a queue. So let's take a good queue package. So, you know, everyone always talks about celery, but celery is overkill for, you know, the majority of use cases. So what's a good package? Well, there's one called django-q, which I love and have fun with. That's nice and simple. And that's got a Redis back end. So you pip install, right or, you know, apt install Redis. And then you pip install django-q into your project, you know, a little bit of settings, magic, and you're up and running [..]

Packages Considered

I did not compare packages. To compare would mean installing each one, run the same task and measure. I had what was left of a "free" day to switch over from celery to another package and have things deployed by day's end.

I considered these packages to see if I could adopt a more lightweight alternative to celery. Before I continue, by "lightweight" I mean, "lightweight in terms of":

package size and dependencies
code that I would need to rework to transition from celery to this task queue
works with Redis, a pre-existing component in my stack

The packages I ~~compared~~ considered:

Package	Link	How I got to know about it
`dramatiq`	Link	Ex-colleague's experience, described above.
`django-q`	Link	Recommended during the DjangoChat podcast.
`huey`	Link	Recommended by Adam Johnson, Django blogger I follow.

I decided to move on with Huey. It was not a clear-cut decision.

My mindset was not about installing the best. It was about installing a minimal package. That removes my dependency on celery/django-celery. And allows me to continue taking that project's codebase to Python 3.

Why not dramatiq? I did not go with dramatiq because of two reasons:

It required installation of another package. The "Advanced Python Scheduler", APS, to allow scheduling of tasks.
django-dramatiq, while maintained by the author of dramatiq itself, is "yet another package".

After my experience with django-celery any extra package scared me. I did not want to end up being unable to use a main library due to a smaller accompanying library not being maintained.

In comparison, for Huey I would only need to pip install huey. The Django integration part is part of the package. Docs here.

In addition, dramatiq is intended for rabbitmq. I see no need for advanced messaging such as that enabled by rabbitmq right now.¹

Why not django-q? It is actively maintained. And targeted for Django. And offers a lot of features. But by the looks of it, it offered many features I was not going to be using. At the cost of being less lightweight than Huey.

The above does not mean dramatiq or django-q are not great packages. Far from it. I have them in mind in case the use case changes.

Huey

Huey's Django integration provides:

Configuration of huey via the Django settings module.

Running the consumer as a Django management command.

Auto-discovery of tasks.py modules to simplify task importing.

Properly manage database connections.

Sweet.

Code changes

Installed huey? The Setting Things Up section in Huey’s Django integration guide covers what you need to do from then on.

Instead of importing the task or periodic_task decorators from the main Huey package, import from huey.contrib.djhuey:

from huey.contrib.djhuey import periodic_task, task

Huey also offers function decorators for tasks that execute queries, which automatically close the database connection:

from huey.contrib.djhuey import db_periodic_task, db_task

To set a task performing a db_task therefore:

from huey.contrib import djhuey as huey

@huey.db_task()
def save_data_and_send_email():
    ...

What about settings files? Main settings:

# HUEY
HUEY = {
    'name': 'mydjangoproject',
    'url': 'redis://localhost:6379/?db=1',
}

Seriously, that's enough to have things running!

A note about how things run locally by default:

When settings.DEBUG = True, tasks will be executed synchronously just like regular function calls. The purpose of this is to avoid running both Redis and an additional consumer process while developing or running tests.

This is probably the reason why run_huey doesn't autoreload.

To have Huey use redis to broker tasks locally:

ensure redis service is available locally (duh!)
set immediate and immediate_use_memory to False:

HUEY['immediate_use_memory'] = False
HUEY['immediate'] = False

More details on debug and synchronous execution with Huey here.

Passing Parameters

Note that, like in celery's case, Huey uses pickle to serialize messages (docs).

So pay attention to pass variable types that play well with pickle when passing parameters.

Thus avoid passing a Django model instance or queryset as parameter, as shown below:

from huey.contrib import djhuey as huey

@huey.task()
def notify_user(user):
    send_email(user.email, ...)

my_user = User.objects.get(id=user_id)
send_email(my_user)()  # enqueue task

Instead pass the object id, which is an int:

from huey.contrib import djhuey as huey

@huey.task()
def notify_user(user_id):
    user = User.objects.get(id=user_id)
    send_email(user.email, ...)

send_email(my_user.id)()  # enqueue task

The serializer can be overridden. But going into that takes us beyond minimal for today.

Periodic tasks

This is an example of a periodic task. It calls a Django management command to clear expired user sessions every 2 hours:

from huey import crontab
from huey.contrib import djhuey as huey

@huey.periodic_task(crontab(hour='*/2'))
def clear_expired_sessions():
    from django.core.management import call_command
    return call_command('clearsessions')

One minimal aspect of Django+Huey is when it comes to deployment (tackled further down).

manage.py run_huey takes care of executing both:

tasks that your code adds to the queue while it's running, e.g. send an email triggered by a user action
scheduled (aka "periodic") tasks

Both the task consumer and scheduler are run by the same run_huey process.

Retries

Example: I want to fetch API data on the first of the month at 2:30PM:

from huey import crontab
from huey.contrib import djhuey as huey

@huey.db_periodic_task(
    crontab(day='1', hour='14', minute='30'),
    retries=2, retry_delay=10)
@huey.lock_task('sync_gsuite_data')
def fetch_api_data():
    ...

The above retries the function twice, with an interval of 10 seconds.

The one aspect that's missing in this setup is a hook to retry only in case of specific exception. I want an exception caused by a network error to be retried. But not an exception due to a ZeroDivisionError, for example.

huey provides signals that allow you to inspect an exception on various types of events. For example, you could use the below code to notify admins when a Huey task fails without being retried:

import traceback

from django.core.mail import mail_admins

from huey import signals
from huey.contrib import djhuey as huey


@huey.signal(signals.SIGNAL_ERROR)
def task_error(signal, task, exc):
    if task.retries > 0:
        return  # do not notify when task is to be retried
    subject = f'Task [{task.name}] failed'
    message = f"""Task ID: {task.id}
Args: {task.args}
Kwargs: {task.kwargs}
Exception: {exc}
{traceback.format_exc()}"""
    mail_admins(subject, message)

Sidenote 1: Keep in mind signals are executed synchronously by the consumer as it processes tasks.

Sidenote 2: this is just to demonstrate what can be done. A more standard way to do this is to attach an email handler at loglevel ERROR to the huey consumer logger.

But I would prefer the hook that dramatiq provides to determine whether a task should be retried in this style:

def should_retry(retries_so_far, exception):
    return retries_so_far < 3 and isinstance(exception, HttpTimeout)


@dramatiq.actor(retry_when=should_retry)
def count_words(url):
    ...

See? I'm veering off "minimal" and into better features. Let's move on.

Simple locking

To quote huey's author himself:

A simple lock ensures that one task cannot be executed in parallel.

Example use case: a report generation task that runs every 10 minutes, but occasionally it can take 15 minutes to complete. You want to ensure that it does not start a stampede. So you use a lock to ensure that only one instance of the task can run at a time.

Example code from huey's docs on locking tasks:

@huey.periodic_task(crontab(minute='*/10'))
@huey.lock_task('reports-lock')  # Goes *after* the task decorator.
def generate_report():
    run_report()

Take a look around

A nice thing about Huey is its docs. Being a relatively small package, you don't easily get lost.

The docs feel "cohesive". Unlike how I used to feel with celery's docs. So feel free to take a look around for other features not mentioned in this tutorial.

The troubleshooting and common pitfalls section saved me a headscratch or two.

Deployment

In my Ubuntu setup I use supervisor for process management. For example supervisor manages the gunicorn which binds the Django application to Nginx.

The bash scipt

To run Huey, supervisor runs a bash script that runs Huey:

start_huey.bash

#!/bin/bash

NAME="mydjangoproject-huey"  # Name of the application
DJANGODIR=/home/ubuntu/webapp/mydjangoproject/proj  # Django project directory
DJANGOENVDIR=/home/ubuntu/webapp/mydjangoproject_env  # Django project virtualenv

echo "Starting $NAME as `whoami`"

# Activate the virtual environment
cd $DJANGODIR
source /home/ubuntu/webapp/mydjangoproject_env/bin/activate
source /home/ubuntu/webapp/mydjangoproject/proj/.env
export PYTHONPATH=$DJANGODIR:$PYTHONPATH

# Start Huey
exec ${DJANGOENVDIR}/bin/python manage.py run_huey

Test the bash script above by running it.

It has to be executable, i.e. chmod +x start_huey.bash if it's not.

Output should be similar to this:

Starting mydjangoproject-huey as ubuntu
[2020-07-01 16:26:54,455] INFO:huey.consumer:MainThread:Huey consumer started with 1 thread, PID 12113 at 2020-07-01 14:26:54.455715
[2020-07-01 16:26:54,456] INFO:huey.consumer:MainThread:Scheduler runs every 1 second(s).
[2020-07-01 16:26:54,456] INFO:huey.consumer:MainThread:Periodic tasks are enabled.
[2020-07-01 16:26:54,456] INFO:huey.consumer:MainThread:The following commands are available:
+ myapp.tasks.send_email
[...]

supervisor conf file

File located at: /etc/supervisor/conf.d/huey.conf

; ================================
;  huey supervisor
; ================================

[program:huey]
command = /home/ubuntu/webapp/start_huey.bash  ; Command to start huey

user=ubuntu
numprocs=1
stdout_logfile=/home/ubuntu/webapp/logs/huey/worker.log
stderr_logfile=/home/ubuntu/webapp/logs/huey/error.log
stdout_logfile_maxbytes=50MB
stderr_logfile_maxbytes=50MB
stdout_logfile_backups=10
stderr_logfile_backups=10 
autostart=true
autorestart=true
startsecs=10

; Need to wait for currently executing tasks to finish at shutdown.
; Increase this if you have very long running tasks.
stopwaitsecs = 2

; Causes supervisor to send the termination signal (SIGTERM) to the whole process group.
stopasgroup=true

A lot of the conf file above is supervisor-specific.

The point of this section is to show you how simple it is to have huey with Django run reliably on an Ubuntu instance.

Conclusion

What do you think about this? Can it be better? Can it be more minimal?

If you want to know more on dramatiq, its author was interviewed in December 2017 on the Python podcast. Great episode! ↩

DEV Community: J.V. Zammit

A minimal Django testing style guide

Terminology

Where do I place tests?

Naming test classes

Naming test functions

Creating test data

Testing views

Unit vs integration, again

How to structure your view tests

Testing configuration

Diving deeper

Test runners other than TestCase

Mocking

Coverage

Speeding tests up

References / Further Reading

Backend Engineer Interview Script

1. Server response time

Caching

Database Query Optimisation

2. Authentication

3. Task queues and async work

4. REST API design

To Conclude

Django Engineer Interview Script

Self-intro

Q1. How do you use the _ character in Python?

Q1 Follow-up: in case "private function" use case is brought up

Q2. Which test data generation toolkit have you used?

Q2 Follow-up: How to improve unit tests speed in Django?

Q3. Which packages you "carry around" in your Django projects?

Q3 Follow-up: How do you decide whether to use a package or reinvent the wheel?

Q4. Which code standards do you follow for a Django project? Tools?

Q5. Which deployment tools have you used? To ship code to production?

To conclude

A minimal Websockets setup with Django in production

Use Case

What does minimal mean in this case?

Blueprint: Before & After

Code changes

1. settings module changes

2. Routing changes

3. The consumer

Give it a try locally

Deployment notes

daphne command tweak

asgi.py tweak

redis version

Deployment - resulting configuration

Bash script

Supervisor

Nginx

A note Markup/Javascript code

Conclusion

Huey as a minimal task queue for Django

Background

Frustrated with celery and django-celery

Is celery heavyweight?

Packages Considered

Huey

Code changes

Passing Parameters

Periodic tasks

Retries

Simple locking

Take a look around

Deployment

The bash scipt

supervisor conf file

Conclusion

Test runners other than `TestCase`