This is the multi-page printable view of this section. Click here to print.

Return to the regular view of this page.

Activities

Shorter activities completed in-class or online

1 - Django Channels Tutorial

Our copy of the Django Channels tutorial, adapted for uv and your locallibrary project.

Channels allows you to use WebSockets and other non-HTTP protocols in your Django site. For example you might want to use WebSockets to allow a page on your site to immediately receive updates from your Django server without using HTTP long-polling or other expensive techniques.

In this tutorial we will build a simple chat server, where you can join an online room, post messages to the room, and have others in the same room see those messages immediately.

  1. Part 1: Basic Setup
  2. Part 2: Implement a Chat Server
  3. Part 3: Rewrite Chat Server as Asynchronous
  4. Part 4: Automated Testing

How this copy differs from the original

This is a copy of the official Channels tutorial, changed to match the way we have worked in our fork of the MDN Django Tutorial:

  • You don’t create a new project. The chat app goes into your existing locallibrary repository, next to catalog. Everywhere the original says mysite, this copy says locallibrary_config.
  • Dependencies are managed with uv. pip install becomes uv add, and every command is run with uv run python ….
  • Edits to existing files are additions. The original shows complete settings.py and urls.py contents because it assumes a new, empty project. Here you only add lines, so you keep what you built for catalog.
  • Part 4’s tests use a separate database file. The original’s test settings would delete your development database; this copy’s don’t.
  • Part 4 doesn’t need a manual chromedriver install. Selenium downloads the right driver itself.

This copy was checked on 2026-09-29 with Channels 4.3.2, Daphne 4.2.3, channels-redis 4.3.0, Selenium 4.49, Django 6.1.1, and Python 3.14.


The original tutorial is Copyright (c) Django Software Foundation and individual contributors, and is used under its BSD 3-Clause license.

1.1 - Tutorial Part 1: Basic Setup

Add Channels and Daphne to locallibrary and create the chat app’s index page.

In this tutorial we will build a simple chat server. It will have two pages:

  • An index view that lets you type the name of a chat room to join.
  • A room view that lets you see messages posted in a particular chat room.

The room view will use a WebSocket to communicate with the Django server and listen for any messages that are posted.

We assume that you have completed our fork of the MDN Django Tutorial so far, and that you have your locallibrary repository: a uv project with manage.py, pyproject.toml, and uv.lock at its root, the locallibrary_config project package, and the catalog app. Every command in this tutorial is run from that root folder.

You can check which version of Django your project uses by running the following command in your terminal:

uv run python -m django --version

Installing Channels and Daphne

Add Channels, together with the Daphne server, to your project:

uv add "channels[daphne]"

Just like uv add "django" did when you set up your project, this records the dependency in pyproject.toml, installs it into .venv, and pins the exact versions in uv.lock. Commit both files.

You can check that they’re installed by running the following command:

uv run python -c "import channels; import daphne; print(channels.__version__, daphne.__version__)"

This tutorial was checked with Channels 4.3.2 and Daphne 4.2.3, on Django 6.1 and Python 3.14.

Redis

This tutorial also uses Docker to install and run Redis. We use Redis as the backing store for the channel layer, which is an optional component of the Channels library that we use in the tutorial. Install Docker from its official website - there are official runtimes for Mac OS and Windows that make it easy to use, and packages for many Linux distributions where it can run natively.

Creating the Chat app

You don’t need to create a project: you already have one. We will put the code for the chat server in its own app, alongside catalog.

Make sure you’re in the same directory as manage.py and type this command:

uv run python manage.py startapp chat

That’ll create a directory chat next to catalog, which is laid out like this:

chat/
    __init__.py
    admin.py
    apps.py
    migrations/
        __init__.py
    models.py
    tests.py
    views.py

For the purposes of this tutorial, we will only be working with chat/views.py and chat/__init__.py. So remove all other files from the chat directory.

After removing unnecessary files, the chat directory should look like:

chat/
    __init__.py
    views.py

We need to tell our project that the chat app is installed. Edit the locallibrary_config/settings.py file and add 'chat' to the end of the INSTALLED_APPS setting, after catalog. Don’t remove anything that’s already there. It’ll look like this:

# locallibrary_config/settings.py
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    # Add our new application
    'catalog.apps.CatalogConfig', # This object was created for us in /catalog/apps.py
    'chat',
]

Add the index view

We will now create the first view, an index view that lets you type the name of a chat room to join.

Create a templates directory in your chat directory. Within the templates directory you have just created, create another directory called chat, and within that create a file called index.html to hold the template for the index view.

Your chat directory should now look like:

chat/
    __init__.py
    templates/
        chat/
            index.html
    views.py

Put the following code in chat/templates/chat/index.html:

<!-- chat/templates/chat/index.html -->
<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Chat Rooms</title>
    <link rel="icon" href="https://fav.mhciael.com/💬">
</head>

<body>
    <form>
        What chat room would you like to enter?<br>
        <input id="room-name-input" type="text" size="100"><br>
        <button id="room-name-submit" type="submit">Enter</button>
    </form>
    <script>
        document.querySelector('#room-name-input').focus();
        const form = document.querySelector('form');

        form.addEventListener('submit', function (e) {
            e.preventDefault();
            var roomName = document.querySelector('#room-name-input').value;
            window.location.pathname = '/chat/' + roomName + '/';
        });
    </script>
</body>

</html>

Create the view function for the room view. Put the following code in chat/views.py:

# chat/views.py
from django.shortcuts import render


def index(request):
    return render(request, "chat/index.html")

To call the view, we need to map it to a URL - and for this we need a URLconf.

To create a URLconf in the chat directory, create a file called urls.py. Your app directory should now look like:

chat/
    __init__.py
    templates/
        chat/
            index.html
    urls.py
    views.py

In the chat/urls.py file include the following code:

# chat/urls.py
from django.urls import path

from . import views


urlpatterns = [
    path("", views.index, name="chat-index"),
]

The next step is to point the root URLconf at the chat.urls module. Your locallibrary_config/urls.py already imports include() and already includes catalog.urls. Add the chat app in the same style, right after the catalog block:

# locallibrary_config/urls.py
# (after the existing catalog block)
urlpatterns += [
    path('chat/', include('chat.urls')),
]

Leave the admin, catalog, redirect, and static-files entries as they are.

Let’s verify that the index view works. Run the following command:

uv run python manage.py runserver

You’ll see output like the following on the command line:

Watching for file changes with StatReloader
Performing system checks...

System check identified no issues (0 silenced).
September 29, 2026 - 09:28:06
Django version 6.1.1, using settings 'locallibrary_config.settings'
Starting development server at http://127.0.0.1:8000/
Quit the server with CONTROL-C.

Unlike in the original tutorial, you won’t see a warning about unapplied migrations, because you already ran migrate for catalog.

Go to http://127.0.0.1:8000/chat/ 🔗 in your browser and you should see the text “What chat room would you like to enter?” along with a text input to provide a room name. (The site root still redirects to catalog/, so type the /chat/ path yourself.)

Type in “lobby” as the room name and press enter. You should be redirected to the room view at http://127.0.0.1:8000/chat/lobby/ but we haven’t written the room view yet, so you’ll get a “Page not found” error page.

Go to the terminal where you ran the runserver command and press Control-C to stop the server.

Integrate the Channels library

So far we’ve just created a regular Django app; we haven’t used the Channels library at all. Now it’s time to integrate Channels.

Let’s start by creating a routing configuration for Channels. A Channels routing configuration is an ASGI application that is similar to a Django URLconf, in that it tells Channels what code to run when an HTTP request is received by the Channels server.

Start by replacing the contents of the locallibrary_config/asgi.py file with the following code:

# locallibrary_config/asgi.py
import os

from channels.routing import ProtocolTypeRouter
from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "locallibrary_config.settings")

application = ProtocolTypeRouter(
    {
        "http": get_asgi_application(),
        # Just HTTP for now. (We can add other protocols later.)
    }
)

Now add the Daphne library to the list of installed apps, in order to enable an ASGI version of the runserver command.

Edit the locallibrary_config/settings.py file and add 'daphne' to the top of the INSTALLED_APPS setting. It’ll look like this:

# locallibrary_config/settings.py
INSTALLED_APPS = [
    'daphne',
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    # Add our new application
    'catalog.apps.CatalogConfig', # This object was created for us in /catalog/apps.py
    'chat',
]

You’ll also need to point Daphne at the root routing configuration. Edit the locallibrary_config/settings.py file again and add the following to the bottom of it:

# locallibrary_config/settings.py
# Daphne
ASGI_APPLICATION = "locallibrary_config.asgi.application"

With Daphne now in the installed apps, it will take control of the runserver command, replacing the standard Django development server with the ASGI compatible version.

Let’s ensure that the Channels development server is working correctly. Run the following command:

uv run python manage.py runserver

You’ll see output like the following on the command line:

Watching for file changes with StatReloader
Performing system checks...

System check identified no issues (0 silenced).
September 29, 2026 - 09:28:06
Django version 6.1.1, using settings 'locallibrary_config.settings'
Starting ASGI/Daphne version 4.2.3 development server at http://127.0.0.1:8000/
Quit the server with CONTROL-C.

Notice the line beginning with Starting ASGI/Daphne …. This indicates that the Daphne development server has taken over from the Django development server.

Go to http://127.0.0.1:8000/chat/ 🔗 in your browser and you should still see the index page that we created before. Your catalog pages at http://127.0.0.1:8000/catalog/ 🔗 still work too: Daphne serves them the same way.

Go to the terminal where you ran the runserver command and press Control-C to stop the server.

This tutorial continues in Tutorial 2.

1.2 - Tutorial Part 2: Implement a Chat Server

Add the room view, a WebSocket consumer, and a Redis-backed channel layer.

This tutorial begins where Tutorial 1 left off. We’ll get the room page working so that you can chat with yourself and others in the same room.

Add the room view

We will now create the second view, a room view that lets you see messages posted in a particular chat room.

Create a new file chat/templates/chat/room.html. Your app directory should now look like:

chat/
    __init__.py
    templates/
        chat/
            index.html
            room.html
    urls.py
    views.py

Create the view template for the room view in chat/templates/chat/room.html:

<!-- chat/templates/chat/room.html -->
<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Chat Room</title>
    <link rel="icon" href="https://fav.mhciael.com/💬">
</head>

<body>
    <form>
        <textarea id="chat-log" cols="100" rows="20"></textarea><br>
        <input id="chat-message-input" type="text" size="100"><br>
        <button id="chat-message-submit" type="submit" value="Send">Send</button>
        {{ room_name|json_script:"room-name" }}
    </form>
    <script>
        const roomName = JSON.parse(document.getElementById('room-name').textContent);

        // Use wss:// when the page was loaded over https://, otherwise ws://
        const wsScheme = window.location.protocol === 'https:' ? 'wss://' : 'ws://';

        const chatSocket = new WebSocket(
            wsScheme
            + window.location.host
            + '/ws/chat/'
            + roomName
            + '/'
        );

        chatSocket.onmessage = function (e) {
            const data = JSON.parse(e.data);
            document.querySelector('#chat-log').value += (data.message + '\n');
        };

        chatSocket.onclose = function (e) {
            console.error('Chat socket closed unexpectedly');
        };

        document.querySelector('#chat-message-input').focus();


        document.querySelector('form').addEventListener('submit', (e) => {
            e.preventDefault();
            const messageInputDom = document.querySelector('#chat-message-input');
            const message = messageInputDom.value;
            chatSocket.send(JSON.stringify({
                'message': message
            }));
            messageInputDom.value = '';
        });
    </script>
</body>

</html>

Create the view function for the room view in chat/views.py:

# chat/views.py
from django.shortcuts import render


def index(request):
    return render(request, "chat/index.html")


def room(request, room_name):
    return render(request, "chat/room.html", {"room_name": room_name})

Create the route for the room view in chat/urls.py:

# chat/urls.py
from django.urls import path

from . import views

urlpatterns = [
    path("", views.index, name="chat-index"),
    path("<str:room_name>/", views.room, name="chat-room"),
]

Start the Channels development server:

uv run python manage.py runserver

Go to http://127.0.0.1:8000/chat/ 🔗 in your browser to see the index page.

Type in “lobby” as the room name and press enter. You should be redirected to the room page at http://127.0.0.1:8000/chat/lobby 🔗 which now displays an empty chat log.

Type the message “hello” and press enter. Nothing happens. In particular the message does not appear in the chat log. Why?

The room view is trying to open a WebSocket to the URL ws://127.0.0.1:8000/ws/chat/lobby/ but we haven’t created a consumer that accepts WebSocket connections yet. If you open your browser’s JavaScript console, you should see an error that looks like:

WebSocket connection to 'ws://127.0.0.1:8000/ws/chat/lobby/' failed: Unexpected response code: 500

Write your first consumer

When Django accepts an HTTP request, it consults the root URLconf to lookup a view function, and then calls the view function to handle the request. Similarly, when Channels accepts a WebSocket connection, it consults the root routing configuration to lookup a consumer, and then calls various functions on the consumer to handle events from the connection.

We will write a basic consumer that accepts WebSocket connections on the path /ws/chat/ROOM_NAME/ that takes any message it receives on the WebSocket and echos it back to the same WebSocket.

Create a new file chat/consumers.py. Your app directory should now look like:

chat/
    __init__.py
    consumers.py
    templates/
        chat/
            index.html
            room.html
    urls.py
    views.py

Put the following code in chat/consumers.py:

# chat/consumers.py
import json

from channels.generic.websocket import WebsocketConsumer


class ChatConsumer(WebsocketConsumer):
    def connect(self):
        self.accept()

    def disconnect(self, close_code):
        pass

    def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json["message"]

        self.send(text_data=json.dumps({"message": message}))

This is a synchronous WebSocket consumer that accepts all connections, receives messages from its client, and echos those messages back to the same client. For now it does not broadcast messages to other clients in the same room.

We need to create a routing configuration for the chat app that has a route to the consumer. Create a new file chat/routing.py. Your app directory should now look like:

chat/
    __init__.py
    consumers.py
    routing.py
    templates/
        chat/
            index.html
            room.html
    urls.py
    views.py

Put the following code in chat/routing.py:

# chat/routing.py
from django.urls import re_path

from . import consumers

websocket_urlpatterns = [
    re_path(r"ws/chat/(?P<room_name>\w+)/$", consumers.ChatConsumer.as_asgi()),
]

We call the as_asgi() classmethod in order to get an ASGI application that will instantiate an instance of our consumer for each user-connection. This is similar to Django’s as_view(), which plays the same role for per-request Django view instances.

(Note we use re_path() due to limitations in URLRouter.)

The next step is to point the main ASGI configuration at the chat.routing module. Replace the contents of locallibrary_config/asgi.py with the code below. The highlighted lines are what changed since you edited this file in Part 1:

  • Imports: AuthMiddlewareStack and AllowedHostsOriginValidator are new, and URLRouter joins ProtocolTypeRouter on the existing channels.routing line.
  • django_asgi_app: get_asgi_application() is now called before chat.routing is imported, and its result is saved in django_asgi_app. The comment above it explains why the order matters.
  • from chat.routing import websocket_urlpatterns: this import is new. It deliberately sits below the get_asgi_application() call rather than at the top of the file with the other imports.
  • "http": now uses the saved django_asgi_app instead of calling get_asgi_application() again.
  • "websocket": this key is new and replaces the # Just HTTP for now comment.
# locallibrary_config/asgi.py
import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "locallibrary_config.settings")
# Initialize Django ASGI application early to ensure the AppRegistry
# is populated before importing code that may import ORM models.
django_asgi_app = get_asgi_application()

from chat.routing import websocket_urlpatterns

application = ProtocolTypeRouter(
    {
        "http": django_asgi_app,
        "websocket": AllowedHostsOriginValidator(
            AuthMiddlewareStack(URLRouter(websocket_urlpatterns))
        ),
    }
)

This root routing configuration specifies that when a connection is made to the Channels development server, the ProtocolTypeRouter will first inspect the type of connection. If it is a WebSocket connection (ws:// or wss://), the connection will be given to the AuthMiddlewareStack.

The AuthMiddlewareStack will populate the connection’s scope with a reference to the currently authenticated user, similar to how Django’s AuthenticationMiddleware populates the request object of a view function with the currently authenticated user. (Scopes will be discussed later in this tutorial.) Then the connection will be given to the URLRouter.

The URLRouter will examine the HTTP path of the connection to route it to a particular consumer, based on the provided url patterns.

Let’s verify that the consumer for the /ws/chat/ROOM_NAME/ path works. Django’s session framework needs the database, which you already migrated while building catalog. Run migrate anyway to be sure, and then start the Channels development server:

uv run python manage.py migrate

You should see:

Operations to perform:
  Apply all migrations: admin, auth, catalog, contenttypes, sessions
Running migrations:
  No migrations to apply.

Then start the server:

uv run python manage.py runserver

Go to the room page at http://127.0.0.1:8000/chat/lobby 🔗 which now displays an empty chat log.

Type the message “hello” and press enter. You should now see “hello” echoed in the chat log.

However if you open a second browser tab to the same room page at http://127.0.0.1:8000/chat/lobby 🔗 and type in a message, the message will not appear in the first tab. For that to work, we need to have multiple instances of the same ChatConsumer be able to talk to each other. Channels provides a channel layer abstraction that enables this kind of communication between consumers.

Go to the terminal where you ran the runserver command and press Control-C to stop the server.

Enable a channel layer

A channel layer is a kind of communication system. It allows multiple consumer instances to talk with each other, and with other parts of Django.

A channel layer provides the following abstractions:

  • A channel is a mailbox where messages can be sent to. Each channel has a name. Anyone who has the name of a channel can send a message to the channel.

  • A group is a group of related channels. A group has a name. Anyone who has the name of a group can add/remove a channel to the group by name and send a message to all channels in the group. It is not possible to enumerate what channels are in a particular group.

Every consumer instance has an automatically generated unique channel name, and so can be communicated with via a channel layer.

In our chat application we want to have multiple instances of ChatConsumer in the same room communicate with each other. To do that we will have each ChatConsumer add its channel to a group whose name is based on the room name. That will allow ChatConsumers to transmit messages to all other ChatConsumers in the same room.

We will use a channel layer that uses Redis as its backing store. To start a Redis server on port 6379, run the following command in a separate terminal (press Control-C to stop it):

docker run --rm -p 6379:6379 redis:7

We need to install channels_redis so that Channels knows how to interface with Redis. Run the following command:

uv add channels-redis

Commit the updated pyproject.toml and uv.lock again.

Before we can use a channel layer, we must configure it. Where Redis lives is configuration that differs between machines (your laptop vs. a deployed server), so it belongs in your env file rather than hard-coded in settings.py. You set up .env, .env.example, and django-environ back in Prep 02.

At that point your env file only held the DJANGO_SUPERUSER_* values, so now you’ll add two new variables: REDIS_URL, which tells Django how to reach the Redis server you just started, and REDIS_SOCKET_TIMEOUT, which sets how many seconds to wait on Redis before giving up.

A Redis URL has the form redis://HOST:PORT/DB:

  • redis:// is the scheme (use rediss://, with two s’s, for a TLS connection to a hosted Redis)
  • HOST is the machine Redis runs on. Your Docker container (or redis-server) is on your own machine, so use 127.0.0.1
  • PORT is the port Redis listens on. The -p 6379:6379 in the docker run command above publishes Redis on port 6379
  • DB is the Redis database number. A fresh Redis server has databases 0–15; we’ll use 0
  1. Open .env (in the root of your project, next to manage.py) and add these lines below the existing DJANGO_SUPERUSER_* lines:

    DJANGO_SUPERUSER_USERNAME="me"
    DJANGO_SUPERUSER_PASSWORD="me"
    DJANGO_SUPERUSER_EMAIL="me@me.me"
    REDIS_URL="redis://127.0.0.1:6379/0"
    REDIS_SOCKET_TIMEOUT=10
    
  2. Add the same REDIS_URL and REDIS_SOCKET_TIMEOUT lines to .env.example, so anyone who copies it to create their own .env gets a working value.

  3. Commit .env.example (but never .env; it should already be listed in your .gitignore).

If you ran Redis on a different port (e.g. docker run --rm -p 6380:6379 redis:7 because something else was using 6379), change only the port in your .env, for example REDIS_URL="redis://127.0.0.1:6380/0". Your settings.py won’t need to change. Likewise, if you still see Redis timeouts, try raising REDIS_SOCKET_TIMEOUT in your .env.

Next, edit the locallibrary_config/settings.py file and add a CHANNEL_LAYERS setting to the bottom, below the ASGI_APPLICATION line you added in Part 1. Instead of hard-coding the host and port, read the URL and timeout from your env file with the env object defined near the top of settings.py. It should look like:

# locallibrary_config/settings.py
# Channels
ASGI_APPLICATION = "locallibrary_config.asgi.application"
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [
                {
                    "address": env("REDIS_URL"),
                    "socket_timeout": env.int("REDIS_SOCKET_TIMEOUT"),
                }
            ],
        },
    },
}

Let’s make sure that the channel layer can communicate with Redis. Open a Django shell and run the following commands:

uv run python manage.py shell
import channels.layers
channel_layer = channels.layers.get_channel_layer()
from asgiref.sync import async_to_sync
async_to_sync(channel_layer.send)('test_channel', {'type': 'hello'})
async_to_sync(channel_layer.receive)('test_channel')

You should see:

{'type': 'hello'}

(The number of objects the shell imports automatically depends on your models; that’s the same auto-import you’ve seen in the shell before.)

Type Control-D (or exit()) to exit the Django shell.

Now that we have a channel layer, let’s use it in ChatConsumer. Put the following code in chat/consumers.py, replacing the old code:

# chat/consumers.py
import json

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


class ChatConsumer(WebsocketConsumer):
    def connect(self):
        self.room_name = self.scope["url_route"]["kwargs"]["room_name"]
        self.room_group_name = f"chat_{self.room_name}"

        # Join room group
        async_to_sync(self.channel_layer.group_add)(
            self.room_group_name, self.channel_name
        )

        self.accept()

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

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

        # Send message to room group
        async_to_sync(self.channel_layer.group_send)(
            self.room_group_name, {"type": "chat.message", "message": message}
        )

    # Receive message from room group
    def chat_message(self, event):
        message = event["message"]

        # Send message to WebSocket
        self.send(text_data=json.dumps({"message": message}))

When a user posts a message, a JavaScript function will transmit the message over WebSocket to a ChatConsumer. The ChatConsumer will receive that message and forward it to the group corresponding to the room name. Every ChatConsumer in the same group (and thus in the same room) will then receive the message from the group and forward it over WebSocket back to JavaScript, where it will be appended to the chat log.

Several parts of the new ChatConsumer code deserve further explanation:

  • self.scope["url_route"]["kwargs"]["room_name"]

    • Obtains the 'room_name' parameter from the URL route in chat/routing.py that opened the WebSocket connection to the consumer.
    • Every consumer has a scope that contains information about its connection, including in particular any positional or keyword arguments from the URL route and the currently authenticated user if any.
  • self.room_group_name = f"chat_{self.room_name}"

    • Constructs a Channels group name directly from the user-specified room name, without any quoting or escaping.
    • Group names may only contain alphanumerics, hyphens, underscores, or periods. Therefore this example code will fail on room names that have other characters.
  • async_to_sync(self.channel_layer.group_add)(...)

    • Joins a group.
    • The async_to_sync(...) wrapper is required because ChatConsumer is a synchronous WebsocketConsumer but it is calling an asynchronous channel layer method. (All channel layer methods are asynchronous.)
    • Group names are restricted to ASCII alphanumerics, hyphens, and periods only and are limited to a maximum length of 100 in the default backend. Since this code constructs a group name directly from the room name, it will fail if the room name contains any characters that aren’t valid in a group name or exceeds the length limit.
  • self.accept()

    • Accepts the WebSocket connection.
    • If you do not call accept() within the connect() method then the connection will be rejected and closed. You might want to reject a connection for example because the requesting user is not authorized to perform the requested action.
    • It is recommended that accept() be called as the last action in connect() if you choose to accept the connection.
  • async_to_sync(self.channel_layer.group_discard)(...)

    • Leaves a group.
  • async_to_sync(self.channel_layer.group_send)

    • Sends an event to a group.
    • An event has a special 'type' key corresponding to the name of the method that should be invoked on consumers that receive the event. This translation is done by replacing . with _, thus in this example, chat.message calls the chat_message method.

Let’s verify that the new consumer for the /ws/chat/ROOM_NAME/ path works. To start the Channels development server, run the following command:

uv run python manage.py runserver

Open a browser tab to the room page at http://127.0.0.1:8000/chat/lobby/. Open a second browser tab to the same room page.

In the second browser tab, type the message “hello” and press enter. You should now see “hello” echoed in the chat log in both the second browser tab and in the first browser tab.

You now have a basic fully-functional chat server!

This tutorial continues in Tutorial 3.

1.3 - Tutorial Part 3: Rewrite Chat Server as Asynchronous

Rewrite ChatConsumer as an asynchronous consumer.

This tutorial begins where Tutorial 2 left off. We’ll rewrite the consumer code to be asynchronous rather than synchronous to improve its performance.

Rewrite the consumer to be asynchronous

The ChatConsumer that we have written is currently synchronous. Synchronous consumers are convenient because they can call regular synchronous I/O functions such as those that access Django models without writing special code. However asynchronous consumers can provide a higher level of performance since they don’t need to create additional threads when handling requests.

ChatConsumer only uses async-native libraries (Channels and the channel layer) and in particular it does not access synchronous code. Therefore it can be rewritten to be asynchronous without complications.

Let’s rewrite ChatConsumer to be asynchronous. Put the following code in chat/consumers.py:

# chat/consumers.py
import json

from channels.generic.websocket import AsyncWebsocketConsumer


class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope["url_route"]["kwargs"]["room_name"]
        self.room_group_name = f"chat_{self.room_name}"

        # Join room group
        await self.channel_layer.group_add(self.room_group_name, self.channel_name)

        await self.accept()

    async def disconnect(self, close_code):
        # Leave room group
        await self.channel_layer.group_discard(self.room_group_name, self.channel_name)

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

        # Send message to room group
        await self.channel_layer.group_send(
            self.room_group_name, {"type": "chat.message", "message": message}
        )

    # Receive message from room group
    async def chat_message(self, event):
        message = event["message"]

        # Send message to WebSocket
        await self.send(text_data=json.dumps({"message": message}))

This new code for ChatConsumer is very similar to the original code, with the following differences:

  • ChatConsumer now inherits from AsyncWebsocketConsumer rather than WebsocketConsumer.
  • All methods are async def rather than just def.
  • await is used to call asynchronous functions that perform I/O.
  • async_to_sync is no longer needed when calling methods on the channel layer.

Let’s verify that the consumer for the /ws/chat/ROOM_NAME/ path still works. Make sure Redis is still running, then start the Channels development server:

uv run python manage.py runserver

Open a browser tab to the room page at http://127.0.0.1:8000/chat/lobby/. Open a second browser tab to the same room page.

In the second browser tab, type the message “hello” and press enter. You should now see “hello” echoed in the chat log in both the second browser tab and in the first browser tab.

Now your chat server is fully asynchronous!

This tutorial continues in Tutorial 4.

1.4 - Tutorial Part 4: Automated Testing

End-to-end tests for the chat server with Selenium and ChannelsLiveServerTestCase.

This tutorial begins where Tutorial 3 left off. We’ve built a simple chat server and now we’ll create some automated tests for it.

Testing the views

To ensure that the chat server keeps working, we will write some tests.

We will write a suite of end-to-end tests using Selenium to control a Chrome web browser. These tests will ensure that:

  • when a chat message is posted then it is seen by everyone in the same room
  • when a chat message is posted then it is not seen by anyone in a different room

Install the Chrome web browser, if you do not already have it.

Install Selenium as a development dependency, like the testing tools in the MDN tutorial, so that it is never installed on your production server. Run the following command:

uv add --dev selenium

Commit the updated pyproject.toml and uv.lock.

You don’t need to install chromedriver yourself. The first time the tests start Chrome, Selenium finds a matching chromedriver or downloads one automatically.

Create a new file chat/tests.py. Your app directory should now look like:

chat/
    __init__.py
    consumers.py
    routing.py
    templates/
        chat/
            index.html
            room.html
    tests.py
    urls.py
    views.py

Put the following code in chat/tests.py:

# chat/tests.py
from channels.testing import ChannelsLiveServerTestCase
from selenium import webdriver
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.wait import WebDriverWait


class ChatTests(ChannelsLiveServerTestCase):
    serve_static = True  # emulate StaticLiveServerTestCase

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        try:
            # NOTE: Selenium finds or downloads a matching chromedriver automatically
            cls.driver = webdriver.Chrome()
        except:
            super().tearDownClass()
            raise

    @classmethod
    def tearDownClass(cls):
        cls.driver.quit()
        super().tearDownClass()

    def test_when_chat_message_posted_then_seen_by_everyone_in_same_room(self):
        try:
            self._enter_chat_room("room_1")

            self._open_new_window()
            self._enter_chat_room("room_1")

            self._switch_to_window(0)
            self._post_message("hello")
            WebDriverWait(self.driver, 2).until(
                lambda _: "hello" in self._chat_log_value,
                "Message was not received by window 1 from window 1",
            )
            self._switch_to_window(1)
            WebDriverWait(self.driver, 2).until(
                lambda _: "hello" in self._chat_log_value,
                "Message was not received by window 2 from window 1",
            )
        finally:
            self._close_all_new_windows()

    def test_when_chat_message_posted_then_not_seen_by_anyone_in_different_room(self):
        try:
            self._enter_chat_room("room_1")

            self._open_new_window()
            self._enter_chat_room("room_2")

            self._switch_to_window(0)
            self._post_message("hello")
            WebDriverWait(self.driver, 2).until(
                lambda _: "hello" in self._chat_log_value,
                "Message was not received by window 1 from window 1",
            )

            self._switch_to_window(1)
            self._post_message("world")
            WebDriverWait(self.driver, 2).until(
                lambda _: "world" in self._chat_log_value,
                "Message was not received by window 2 from window 2",
            )
            self.assertTrue(
                "hello" not in self._chat_log_value,
                "Message was improperly received by window 2 from window 1",
            )
        finally:
            self._close_all_new_windows()

    # === Utility ===

    def _enter_chat_room(self, room_name):
        self.driver.get(self.live_server_url + "/chat/")
        ActionChains(self.driver).send_keys(room_name, Keys.ENTER).perform()
        WebDriverWait(self.driver, 2).until(
            lambda _: room_name in self.driver.current_url
        )

    def _open_new_window(self):
        self.driver.execute_script('window.open("about:blank", "_blank");')
        self._switch_to_window(-1)

    def _close_all_new_windows(self):
        while len(self.driver.window_handles) > 1:
            self._switch_to_window(-1)
            self.driver.execute_script("window.close();")
        if len(self.driver.window_handles) == 1:
            self._switch_to_window(0)

    def _switch_to_window(self, window_index):
        self.driver.switch_to.window(self.driver.window_handles[window_index])

    def _post_message(self, message):
        ActionChains(self.driver).send_keys(message, Keys.ENTER).perform()

    @property
    def _chat_log_value(self):
        return self.driver.find_element(
            by=By.CSS_SELECTOR, value="#chat-log"
        ).get_property("value")

Our test suite extends ChannelsLiveServerTestCase rather than Django’s usual suites for end-to-end tests (StaticLiveServerTestCase or LiveServerTestCase) so that URLs inside the Channels routing configuration like /ws/chat/ROOM_NAME/ will work inside the suite.

We are using sqlite3, which for testing is run as an in-memory database, and therefore the tests will not run correctly. We need to tell our project that the test database should be a file instead. Edit the DATABASES setting in locallibrary_config/settings.py and add a TEST entry, keeping the ENGINE and NAME lines you already have:

# locallibrary_config/settings.py
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': BASE_DIR / 'db.sqlite3',
        'TEST': {
            'NAME': BASE_DIR / 'test_db.sqlite3',
        },
    }
}

Django creates test_db.sqlite3 while the tests run, and it should never be committed. If you started your .gitignore from GitHub’s Python template, its *.sqlite3 line already ignores it. Otherwise, add this line:

test_db.sqlite3

The tests use your real channel layer, so Redis must be running (see Part 2). To run the tests, run the following command:

uv run python manage.py test chat.tests

You should see output that looks like:

Creating test database for alias 'default'...
Found 2 test(s).
System check identified no issues (0 silenced).
Not Found: /favicon.ico
..
----------------------------------------------------------------------
Ran 2 tests in 8.953s

OK
Destroying test database for alias 'default'...

Chrome windows will open and close by themselves while the tests run. The Not Found: /favicon.ico line is Chrome asking for an icon the chat pages don’t have; you can ignore it.

You now have a tested chat server!

What’s next?

Congratulations! You’ve fully implemented a chat server, made it performant by writing it in asynchronous style, and written automated tests to ensure it won’t break.

This is the end of the tutorial. At this point you should know enough to start an app of your own that uses Channels and start fooling around. As you need to learn new tricks, come back to the rest of the Channels documentation.

2 - Django Ninja Quickstart

Our copy of the Django Ninja Quick Start, adapted for uv and the chat app in your locallibrary project.

In three short pages you’ll build a small chat API for the chat rooms in your locallibrary: people can browse rooms, filter and paginate them, read a room’s messages, sign in, and post a message that shows up live in the chat page. Along the way you’ll see most of what makes Django Ninja pleasant to use: type hints that validate input, schemas generated from your models, interactive docs, auth, and routers.

This guide assumes you already know Django: projects, apps, models and migrations. It won’t explain those; it only shows the Django Ninja parts.

Page What you’ll do
1. First API Install, create an API, write an endpoint, open the interactive docs
2. Models & Schemas Save chat rooms and messages to the database, then full CRUD for the Room model, plus pagination and filtering
3. Auth, Routers & Next Steps Split the API into a router, require login, let users post messages to a room

Before you start

We assume that you have completed preps 01-06 (our fork of the MDN Django Tutorial through Part 8) and all four parts of our Channels tutorial. You have your locallibrary repository: a uv project with manage.py, pyproject.toml, and uv.lock at its root, the locallibrary_config project package, the catalog app, and the asynchronous chat app. Every command in this guide is run from that root folder.

The chat app works, but it forgets everything: a message exists only while it travels through Redis to the browsers in the room. An API that lists rooms and messages needs them in the database. So before the Django Ninja parts, page 2 adds Room and Message models to the chat app, saves every message to the database, and shows a room’s history when you open it.

How this copy differs from the original

This is a copy of the official Django Ninja Quick Start, changed to match the way we have worked in our fork of the MDN Django Tutorial and our Channels tutorial:

  • You don’t create a new project or app. The API goes into your existing locallibrary repository. Everywhere the original says mysite, this copy says locallibrary_config, and the original’s events app is your chat app.
  • Chat rooms instead of events. The original builds an API for an Event model, and users register for events. This copy builds the same API for a Room model, and users post Messages to rooms. Each piece of the original has a counterpart here: EventFilter’s upcoming filter becomes RoomFilter’s recent filter, the 409 for an event that clashes with another becomes a 409 for a room name that’s taken, and the nested RegistrationOut schema becomes a nested MessageOut schema.
  • You add models to the chat app first. Page 2 starts with Room and Message models, saves messages from the WebSocket consumer, and shows a room’s saved messages when you open it. The original has nothing like this, because it starts from an empty app.
  • Posted messages go to the chat page. On page 3, a message posted through the API is also sent to everyone in the room over the channel layer, so it appears in the chat page right away.
  • Dependencies are managed with uv. pip install becomes uv add, and every command is run with uv run python ….
  • Edits to existing files are additions. The original shows complete urls.py contents because it assumes a new, empty project. Here you only add lines, so you keep what you built for catalog and chat.

This copy follows the Django Ninja documentation as of 2026-10-19 (Django Ninja 1.7.1), on Django 6.1, Channels 4.3, and Python 3.14. Once you finish, continue with our Django Ninja Tutorial.


The original Quick Start is Copyright (c) 2025 Vitaliy Kucheryaviy, and is used under its MIT license.

2.1 - Quick Start Part 1: First API

Install Django Ninja, create an API in locallibrary, write an endpoint, and open the interactive docs.

Install

Add Django Ninja to your project:

uv add django-ninja

Just like uv add "channels[daphne]" did in the Channels tutorial, this records the dependency in pyproject.toml, installs it into .venv, and pins the exact versions in uv.lock. Commit both files.

Django Ninja works with any Django project. The original Quick Start creates a new project and an events app at this point. You don’t: you already have a project, locallibrary_config, and the chat app plays the part of events for the rest of this guide.

Django Ninja itself doesn’t need to be in INSTALLED_APPS, but adding "ninja" lets Django’s runserver and collectstatic serve the interactive docs’ assets locally instead of from a CDN. Edit locallibrary_config/settings.py and add 'ninja' to the end of INSTALLED_APPS, after chat. Don’t remove anything that’s already there:

# locallibrary_config/settings.py
INSTALLED_APPS = [
    'daphne',
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    # Add our new application
    'catalog.apps.CatalogConfig', # This object was created for us in /catalog/apps.py
    'chat',
    'ninja',
]

Create the API

An API starts with a NinjaAPI instance. Create locallibrary_config/api.py next to urls.py:

# locallibrary_config/api.py
from ninja import NinjaAPI

api = NinjaAPI()


@api.get("/hello")
def hello(request, name: str = "world"):
    return {"message": f"Hello, {name}!"}

Then mount it in locallibrary_config/urls.py, like any other set of views. That file already includes catalog.urls and chat.urls. Add the API in the same style, at the end of the file:

# locallibrary_config/urls.py
# (after the existing blocks)
# Mount the Django Ninja API
from .api import api

urlpatterns += [
    path('api/', api.urls),
]

Leave the admin, catalog, chat, redirect, and static-files entries as they are.

Run the server:

uv run python manage.py runserver

This is the Daphne development server from the Channels tutorial. It serves your Django Ninja API the same way it serves your catalog pages. Redis doesn’t need to be running for this page; only the chat’s WebSockets use it.

Open http://127.0.0.1:8000/api/hello?name=Ninja 🔗:

{"message": "Hello, Ninja!"}

A few things happened with no extra code:

  • @api.get("/hello") registered a GET endpoint at /api/hello.
  • The request argument is a regular Django HttpRequest, so request.user, sessions and middleware all work as usual.
  • name: str = "world" became an optional query parameter with a default value.
  • The returned dict was serialized to JSON.

Type hints are validation

Every argument’s type hint says what the endpoint accepts. Add an endpoint with two required integers to locallibrary_config/api.py:

# locallibrary_config/api.py
@api.get("/add")
def add(request, a: int, b: int):
    return {"result": a + b}

/api/add?a=2&b=3 🔗 returns {"result": 5}, and a and b arrive in your function as real ints, not strings. Invalid input never reaches your code. Try /api/add?a=2&b=three 🔗:

{
    "detail": [
        {
            "type": "int_parsing",
            "loc": ["query", "b"],
            "msg": "Input should be a valid integer, unable to parse string as an integer"
        }
    ]
}

The response has status code 422, and the error points to exactly which parameter failed and why. The same works for float, bool, date, datetime, UUID, enums, and more. See Query Parameters for the full list.

Interactive docs

Now open http://127.0.0.1:8000/api/docs 🔗:

Swagger UI

Django Ninja generates an OpenAPI schema from your code, including parameter types, required fields and responses, and serves it with Swagger UI. Click an endpoint, then Try it out to call it from the browser. The raw schema is at /api/openapi.json, ready for client generators and API tools.

The docs are never out of date, because they are built from the same type hints that validate your requests. See OpenAPI & Interactive Docs to customize them.

Next

So far the API only returns hand-built dicts. Next, give your chat rooms and messages a home in the database, connect the API to them, and build a real CRUD API: Models & Schemas.

2.2 - Quick Start Part 2: Models & Schemas

Save chat rooms and messages to the database, then build a CRUD API for rooms with pagination and filtering.

On this page you’ll give your chat rooms and messages a home in the database, then turn the Room model into a complete CRUD API, and add pagination and filtering. Along the way you’ll see how schemas are generated from your models.

The models

Nothing Ninja-specific here, just regular Django models. The original Quick Start has a single Event model. Your chat needs two: a Room, and the Messages posted in it.

In Part 1 of the Channels tutorial, you deleted chat/models.py because the chat didn’t store anything. Create it again, with this code:

# chat/models.py
from django.conf import settings
from django.db import models


class Room(models.Model):
    name = models.CharField(max_length=50, unique=True)
    topic = models.CharField(max_length=200, blank=True, default="")
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.name


class Message(models.Model):
    room = models.ForeignKey(Room, related_name="messages", on_delete=models.CASCADE)
    author = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        null=True,
        blank=True,
        related_name="chat_messages",
        on_delete=models.SET_NULL,
    )
    content = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        ordering = ["created_at", "id"]

    def __str__(self):
        return f"{self.author or 'anonymous'} in {self.room}: {self.content[:50]}"
  • A room’s name is the name in its URL, like lobby in /chat/lobby/, so two rooms can’t share one: unique=True. It’s also part of the room’s Channels group name (chat_lobby), and group names must be shorter than 100 characters, so 50 leaves plenty of room.
  • topic is optional. blank=True lets forms and the admin leave it empty, and default="" gives new rooms an empty topic.
  • author is optional too, because anyone can chat without logging in: AuthMiddlewareStack in your asgi.py puts an anonymous user in the scope for them. If a user is deleted, their messages stay, with no author (SET_NULL).
  • related_name="chat_messages" lets you write user.chat_messages. Plain messages would read as if it had to do with Django’s messages framework, which is also installed.
  • Meta.ordering keeps every list of messages oldest first, the way a chat log reads.

You also deleted the chat/migrations/ folder in Part 1 of the Channels tutorial. Naming the app tells makemigrations to create it again:

uv run python manage.py makemigrations chat
uv run python manage.py migrate

You should see:

Migrations for 'chat':
  chat/migrations/0001_initial.py
    + Create model Room
    + Create model Message
Operations to perform:
  Apply all migrations: admin, auth, catalog, chat, contenttypes, sessions
Running migrations:
  Applying chat.0001_initial... OK

Your Channels tutorial also deleted chat/admin.py. Create it again, so you can look at rooms and messages in the admin site:

# chat/admin.py
from django.contrib import admin

from .models import Message, Room

admin.site.register(Room)
admin.site.register(Message)

Save messages from the consumer

The ChatConsumer you wrote in Part 3 of the Channels tutorial is asynchronous. As that page’s note said, an asynchronous consumer can still use Django models, through the model methods that start with a, like aget_or_create() and acreate(), as long as it calls channels.db.aclose_old_connections() before it uses the database. Put the following code in chat/consumers.py. The highlighted lines are new:

# chat/consumers.py
import json

from channels.db import aclose_old_connections
from channels.generic.websocket import AsyncWebsocketConsumer

from .models import Message, Room


class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope["url_route"]["kwargs"]["room_name"]
        self.room_group_name = f"chat_{self.room_name}"

        # Find the room in the database, or create it the first time anyone joins
        await aclose_old_connections()
        self.room, _ = await Room.objects.aget_or_create(name=self.room_name)

        # Join room group
        await self.channel_layer.group_add(self.room_group_name, self.channel_name)

        await self.accept()

    async def disconnect(self, close_code):
        # Leave room group
        await self.channel_layer.group_discard(self.room_group_name, self.channel_name)

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

        # Save the message before anyone sees it
        await aclose_old_connections()
        user = self.scope["user"]
        await Message.objects.acreate(
            room=self.room,
            author=user if user.is_authenticated else None,
            content=message,
        )

        # Send message to room group
        await self.channel_layer.group_send(
            self.room_group_name, {"type": "chat.message", "message": message}
        )

    # Receive message from room group
    async def chat_message(self, event):
        message = event["message"]

        # Send message to WebSocket
        await self.send(text_data=json.dumps({"message": message}))
  • aget_or_create() returns a (room, created) tuple. The first person to open /chat/lobby/ creates the lobby room, and everyone after that joins the same row. The consumer keeps the room in self.room for as long as the WebSocket is open.
  • self.scope["user"] is the user that AuthMiddlewareStack found, the same as request.user in a view. If you’re logged in, for example at /accounts/login/ from Prep 06, your messages get you as their author. Otherwise they’re saved with no author.
  • The message is saved before it’s sent to the group. If saving fails, nobody sees a message that won’t be there when they reload the page.

Show a room’s history

Saved messages are only useful if you can see them. Update the room view in chat/views.py to load the room’s messages:

# chat/views.py
from django.shortcuts import render

from .models import Room


def index(request):
    return render(request, "chat/index.html")


def room(request, room_name):
    room = Room.objects.filter(name=room_name).first()
    chat_messages = room.messages.all() if room else []
    return render(
        request,
        "chat/room.html",
        {"room_name": room_name, "chat_messages": chat_messages},
    )

The context variable is called chat_messages, not messages. Django’s messages framework already puts a messages variable in every template’s context, and yours would hide it.

The view doesn’t create the room: opening the page isn’t enough to make a room, but connecting the WebSocket is. If the room doesn’t exist yet, there’s no history to show.

In chat/templates/chat/room.html, change only the textarea line, so that it starts with one line per saved message:

<!-- chat/templates/chat/room.html -->
<textarea id="chat-log" cols="100" rows="20">{% for chat_message in chat_messages %}{{ chat_message.content }}
{% endfor %}</textarea><br>

That’s the same format the JavaScript in the page uses when a new message arrives: the message, then a newline. The line break inside the for loop is that newline, so keep it exactly where it is, and don’t add spaces or line breaks anywhere else between the textarea tags. In particular, the {% endfor %} line must start at the very beginning of the line, even though the rest of the file is indented: any spaces before it would end up at the start of every message but the first.

Try it

Make sure Redis is running (see Part 2 of the Channels tutorial), then start the server:

uv run python manage.py runserver

Open http://127.0.0.1:8000/chat/lobby/ 🔗 in two tabs, and send “hello” and “anyone here?” from one of them. Both show up in both tabs, as before. Now reload a tab: the messages are still there.

Go to http://127.0.0.1:8000/admin/chat/message/ 🔗 and log in with your superuser. Your messages are listed there, and the lobby room is under Rooms. Messages you sent while logged in show your username as their author.

Finally, run the Channels tutorial’s tests again, to make sure the chat still works end to end:

uv run python manage.py test chat.tests

The rest of this page is about Django Ninja again. Join two or three more rooms, such as books and random, so the API has a few rooms to work with.

Schemas

A schema describes the shape of data going in or out of your API. Django Ninja schemas are Pydantic models: a class with type-annotated fields.

# locallibrary_config/api.py
from datetime import datetime

from ninja import Schema


class RoomIn(Schema):
    name: str
    topic: str = ""


class RoomOut(Schema):
    id: int
    name: str
    topic: str
    created_at: datetime

RoomIn is what clients send, and RoomOut is what they get back. They’re separate so a client can’t set id or created_at (or, later, fields you’d rather keep private). topic has a default, so it’s optional in RoomIn.

A schema doesn’t have to match a model. You’ll use plain Schema classes for anything that isn’t a table: login payloads, computed results, aggregates.

The same, generated from the model

When a schema mirrors a model, as these do, it only repeats what the model already says. ModelSchema reads the fields from the model instead:

# locallibrary_config/api.py
from ninja import ModelSchema

from chat.models import Room


class RoomIn(ModelSchema):
    class Meta:
        model = Room
        fields = ["name", "topic"]


class RoomOut(ModelSchema):
    class Meta:
        model = Room
        fields = ["id", "name", "topic", "created_at"]

These two classes behave the same as the hand-written ones, with the types and the topic default taken from the model. You also get the model’s constraints: name is limited to 50 characters because of max_length=50. When you add or change a model field, the schema follows.

The rest of this Quick Start uses the ModelSchema versions. You can mix both kinds freely, and a ModelSchema can also declare extra fields of its own (see ModelSchema).

Read: list and detail

Use response= to tell Django Ninja what an endpoint returns:

# locallibrary_config/api.py
from django.shortcuts import get_object_or_404


@api.get("/rooms", response=list[RoomOut])
def list_rooms(request):
    return Room.objects.all()


@api.get("/rooms/{room_id}", response=RoomOut)
def get_room(request, room_id: int):
    return get_object_or_404(Room, id=room_id)
  • You return querysets and model instances directly. Django Ninja reads the fields listed in the schema and serializes them to JSON. There’s no serializer to call.
  • {room_id} in the path becomes the room_id: int argument, converted and validated like the query parameters on the previous page.
  • Django’s get_object_or_404 works as usual: a missing room is a 404 JSON response.
  • Only the fields in RoomOut are sent back, even if the object has more. The schema is also the contract shown in the docs.

GET /api/rooms/1:

{
    "id": 1,
    "name": "lobby",
    "topic": "",
    "created_at": "2026-10-19T14:02:11.482Z"
}

Create, update, delete

A schema-typed argument is read from the JSON request body. The highlighted lines are the ones explained below:

# locallibrary_config/api.py
@api.post("/rooms", response={201: RoomOut})
def create_room(request, payload: RoomIn):
    return Room.objects.create(**payload.dict())


@api.put("/rooms/{room_id}", response=RoomOut)
def update_room(request, room_id: int, payload: RoomIn):
    room = get_object_or_404(Room, id=room_id)
    for attr, value in payload.dict().items():
        setattr(room, attr, value)
    room.save()
    return room


@api.delete("/rooms/{room_id}", response={204: None})
def delete_room(request, room_id: int):
    get_object_or_404(Room, id=room_id).delete()
  • payload: RoomIn means “parse the body as RoomIn”. By the time your function runs, the data is valid and typed.
  • response={201: RoomOut} maps a status code to a schema. With only one code listed, that’s the code every response gets, so create_room returns 201 and delete_room returns 204 with an empty body.
  • update_room mixes a path parameter and a body in one signature. Django Ninja works out where each argument comes from.
  • Deleting a room also deletes its messages, because of on_delete=models.CASCADE on Message.room.

Send a bad body, such as {"topic": 42}, and the error lists every problem at once:

{
    "detail": [
        {"type": "missing", "loc": ["body", "payload", "name"], "msg": "Field required"},
        {"type": "string_type", "loc": ["body", "payload", "topic"], "msg": "Input should be a valid string"}
    ]
}

Open /api/docs again. All five room endpoints are there, with request and response schemas, and you can create a room from the browser with Try it out. A room you create there is a real room: open /chat/<its name>/ and chat in it.

More than one status code

Try creating a room called lobby with Try it out. You get a 500 error, not a helpful answer: the name column is unique, so the database refuses the second lobby with an IntegrityError.

When an endpoint can answer in different ways, list every status code in response and return Status(code, data) to pick one. Replace create_room with a version that creates a room, or reports a clash with an existing one:

# locallibrary_config/api.py
from ninja import Schema, Status


class Conflict(Schema):
    message: str
    existing_id: int


@api.post("/rooms", response={201: RoomOut, 409: Conflict})
def create_room(request, payload: RoomIn):
    existing = Room.objects.filter(name=payload.name).first()
    if existing:
        return Status(409, {"message": "A room with that name already exists", "existing_id": existing.id})
    return Status(201, Room.objects.create(**payload.dict()))

Each response is validated against the schema for its code, and both appear in the interactive docs. See Responses for more.

(Renaming a room to a name that’s taken with PUT has the same problem. Fixing update_room the same way is a good exercise.)

Pagination

A list endpoint that returns every row won’t last long in production. Add one decorator:

# locallibrary_config/api.py
from ninja.pagination import paginate


@api.get("/rooms", response=list[RoomOut])
@paginate
def list_rooms(request):
    return Room.objects.all()

The view still returns a plain queryset. @paginate adds limit and offset query parameters, slices the queryset in the database, and wraps the result:

{
    "items": [{"id": 1, "name": "lobby", ...}],
    "count": 3
}

Page-number and cursor pagination, or your own class, are one argument away. See Pagination.

Filtering

Filters are also a schema: a FilterSchema, read from the query string. Each field maps to an ORM lookup:

# locallibrary_config/api.py
from datetime import timedelta
from typing import Annotated

from django.db.models import Q
from django.utils import timezone
from ninja import FilterLookup, FilterSchema, Query

...


class RoomFilter(FilterSchema):
    search: Annotated[str | None, FilterLookup("name__icontains")] = None
    topic: Annotated[str | None, FilterLookup("topic__icontains")] = None
    recent: bool | None = None

    def filter_recent(self, value: bool) -> Q:
        week_ago = timezone.now() - timedelta(days=7)
        return Q(created_at__gte=week_ago) if value else Q(created_at__lt=week_ago)


@api.get("/rooms", response=list[RoomOut])
@paginate
def list_rooms(request, filters: Query[RoomFilter]):
    return filters.filter(Room.objects.order_by("name"))
  • search and topic map to icontains lookups on the room’s name and topic.
  • recent needs custom logic, so a filter_recent method returns a Q object: rooms created in the last week, or, with recent=false, the ones before that.
  • Filters left out of the request are skipped, and everything is validated. ?recent=maybe gets a 422, not a server error.

Now these all work, and combine with pagination:

/api/rooms?search=lob
/api/rooms?recent=true&topic=books
/api/rooms?recent=true&limit=10&offset=20

See Filtering for combining filters with OR, filtering on related models, and more.

What you have so far

Saved chat messages, plus about 60 lines of code that give you a paginated, filterable, validated and documented CRUD API:

Method Path
GET /api/rooms list, with search, topic, recent, limit, offset
POST /api/rooms create
GET /api/rooms/{room_id} detail
PUT /api/rooms/{room_id} update
DELETE /api/rooms/{room_id} delete

Right now, though, anyone can create or delete rooms, and the API can’t show or post messages. Next: Auth, Routers & Next Steps.

2.3 - Quick Start Part 3: Auth, Routers & Next Steps

Move the endpoints into the chat app, require login for writes, and let signed-in users post messages to a room.

The rooms API works, but it all lives in one file and anyone can change data. On this page you’ll move the endpoints into the chat app, require login for writes, and let signed-in users post messages that show up in the chat page right away.

Move the endpoints into a router

A Router is a group of endpoints that you mount on the API under a prefix, much like Django’s include(). Each app can own its endpoints and the project just wires them together.

First, move the schemas to chat/schemas.py, unchanged: RoomIn, RoomOut, Conflict and RoomFilter. Then chat/api.py gets the endpoints, with @api replaced by @router and /rooms dropped from the paths:

# chat/api.py
from ninja import Router

router = Router(tags=["rooms"])


@router.get("/", response=list[RoomOut])
@paginate
def list_rooms(request, filters: Query[RoomFilter]):
    ...


@router.get("/{room_id}", response=RoomOut)
def get_room(request, room_id: int):
    ...

# ... and the rest

The chat app now has both chat/views.py, with the views for its HTML pages, and chat/api.py, with its API endpoints. They don’t interfere: chat/urls.py routes to the views, and the API routes to the router.

locallibrary_config/api.py is now only wiring. The hello and add endpoints from page 1 were only for practice, so they go too:

# locallibrary_config/api.py
from ninja import NinjaAPI

from chat.api import router as rooms_router

api = NinjaAPI()

api.add_router("/rooms", rooms_router)

The router’s / becomes /api/rooms/ and /{room_id} becomes /api/rooms/{room_id}. Note the trailing slash on the list: from now on, list and create rooms at /api/rooms/. A GET of /api/rooms still works, because Django redirects it to /api/rooms/, but a POST to it fails, because Django can’t redirect a POST. Thanks to tags=["rooms"], the interactive docs now group these endpoints under a rooms heading. Routers can be nested, and each one can carry its own auth, tags and other settings. See Routers.

Messages

The original Quick Start adds a model here for “user X is going to event Y”. Your chat already has its counterpart, “user X said Y in room Z”: the Message model from page 2. All it needs is schemas. Add them to chat/schemas.py, with Message added to its model import, and the room nested inside each message:

# chat/schemas.py
class MessageIn(ModelSchema):
    class Meta:
        model = Message
        fields = ["content"]


class MessageOut(ModelSchema):
    room: RoomOut

    class Meta:
        model = Message
        fields = ["id", "author", "content", "created_at"]

Declaring room: RoomOut tells Django Ninja to serialize the related Room with RoomOut, instead of returning only its id. author isn’t declared, so it comes back the default way for a foreign key: as the user’s id, or null for a message posted by someone who wasn’t logged in.

Require login

Turn on auth for the whole API in locallibrary_config/api.py:

# locallibrary_config/api.py
from ninja import NinjaAPI
from ninja.security import django_auth

from chat.api import router as rooms_router

api = NinjaAPI(auth=django_auth)

api.add_router("/rooms", rooms_router)

Every endpoint now requires a logged-in user, so you have to opt the public ones back out. Here’s the complete chat/api.py:

# chat/api.py
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from django.shortcuts import get_object_or_404
from ninja import Query, Router, Status
from ninja.errors import HttpError
from ninja.pagination import paginate

from .models import Message, Room
from .schemas import Conflict, MessageIn, MessageOut, RoomFilter, RoomIn, RoomOut

router = Router(tags=["rooms"])


@router.get("/", response=list[RoomOut], auth=None)
@paginate
def list_rooms(request, filters: Query[RoomFilter]):
    return filters.filter(Room.objects.order_by("name"))


@router.get("/mine", response=list[MessageOut])
def my_messages(request):
    return Message.objects.filter(author=request.auth).select_related("room")


@router.get("/{room_id}", response=RoomOut, auth=None)
def get_room(request, room_id: int):
    return get_object_or_404(Room, id=room_id)


@router.post("/", response={201: RoomOut, 409: Conflict})
def create_room(request, payload: RoomIn):
    existing = Room.objects.filter(name=payload.name).first()
    if existing:
        return Status(409, {"message": "A room with that name already exists", "existing_id": existing.id})
    return Status(201, Room.objects.create(**payload.dict()))


@router.put("/{room_id}", response=RoomOut)
def update_room(request, room_id: int, payload: RoomIn):
    room = get_object_or_404(Room, id=room_id)
    for attr, value in payload.dict().items():
        setattr(room, attr, value)
    room.save()
    return room


@router.delete("/{room_id}", response={204: None})
def delete_room(request, room_id: int):
    get_object_or_404(Room, id=room_id).delete()


@router.get("/{room_id}/messages", response=list[MessageOut], auth=None)
@paginate
def list_messages(request, room_id: int):
    room = get_object_or_404(Room, id=room_id)
    return room.messages.select_related("room")


@router.post("/{room_id}/messages", response={201: MessageOut})
def post_message(request, room_id: int, payload: MessageIn):
    room = get_object_or_404(Room, id=room_id)
    if not payload.content.strip():
        raise HttpError(400, "Messages can't be blank")
    message = Message.objects.create(room=room, author=request.auth, content=payload.content)
    # Send it to everyone in the room's chat page, the way ChatConsumer.receive does
    async_to_sync(get_channel_layer().group_send)(
        f"chat_{room.name}", {"type": "chat.message", "message": message.content}
    )
    return message

What’s new:

  • NinjaAPI(auth=django_auth) requires a logged-in Django user for every endpoint, so new endpoints are protected unless you say otherwise. django_auth uses the regular Django session, the same one the admin and your catalog’s login pages use.
  • auth=None opts the three read endpoints back out, so anyone can browse rooms and read their messages, just as anyone can open a chat page. Auth can be set on the whole API, on a router (Router(auth=...)) or on a single endpoint, and the most specific one wins.
  • request.auth is whatever the authenticator returned. For django_auth that’s the User. my_messages lists the messages you posted, from the chat page while logged in or through the API, each with its room nested inside.
  • HttpError(400, "...") stops the request and returns that status with a JSON {"detail": "..."} body. There’s no need to build error responses by hand.
  • list_messages returns the messages oldest first, thanks to Meta.ordering on Message, and pages them with @paginate. select_related("room") fetches each message’s room in the same query, since MessageOut nests it.
  • post_message sends the new message to the room’s group on the channel layer, exactly like ChatConsumer.receive does. Every ChatConsumer in that room receives the chat.message event, and its chat_message method sends the message to its browser. The view is synchronous and channel layer methods are asynchronous, so it calls them through async_to_sync, like your first, synchronous consumer did in Part 2 of the Channels tutorial. That means Redis must be running for post_message to work.

An anonymous request to a protected endpoint gets a 401:

$ curl http://127.0.0.1:8000/api/rooms/mine
{"detail": "Unauthorized"}

To try it out, make sure Redis is running, log in at /admin/ (or at /accounts/login/, from Prep 06), and open /api/docs. Swagger UI reuses your session and sends the CSRF token for you, so you can post a message with Try it out.

In a second tab, open the chat page for the room you’ll post to, such as http://127.0.0.1:8000/chat/lobby/ 🔗. Find its id with GET /api/rooms/?search=lobby. Then, back in the docs, POST /api/rooms/1/messages (with your room’s id) and the body {"content": "hello from the API"}. The message appears in the chat tab right away, and the response has the nested schema:

{
    "room": {
        "id": 1,
        "name": "lobby",
        "topic": "",
        "created_at": "2026-10-19T14:02:11.482Z"
    },
    "id": 7,
    "author": 1,
    "content": "hello from the API",
    "created_at": "2026-10-19T15:21:40.913Z"
}

Your message and user ids will be different. Post {"content": " "} and you get 400 {"detail": "Messages can't be blank"}.

What you’ve built

In three pages, and without writing any serializers, URL patterns or docs by hand, you have a chat that remembers its messages, and an API with:

  • typed, validated path, query and body parameters,
  • schemas generated from Django models, including a nested one,
  • full CRUD, pagination and filtering,
  • session auth with public and protected endpoints,
  • proper error responses,
  • messages posted through the API that reach the chat page live, and
  • interactive OpenAPI docs that always match the code.

Where to go next

Our Django Ninja Tutorial builds a larger project step by step, in your locallibrary next to everything you’ve built so far. The Guide covers each feature in depth. Here’s where to read more about everything used in this Quick Start:

You used Read more
NinjaAPI, urls.py NinjaAPI, URLs & Reverse
@api.get, @api.post, … Operations
name: str, room_id: int Query Parameters, Path Parameters
payload: RoomIn Request Body, plus Form Data and File Uploads
ModelSchema, nested schemas ModelSchema, Schemas
response={201: ...} Responses
@paginate Pagination
FilterSchema Filtering
Router, add_router Routers
auth=django_auth, request.auth Authentication, CSRF
HttpError Errors & Exception Handling
/api/docs OpenAPI & Interactive Docs
async_to_sync, group_send Channel Layers in the Channels docs

Features you haven’t seen yet: async views, throttling, API versioning, custom renderers and a test client.

3 - Django Ninja Tutorial

Our copy of the Django Ninja tutorial, adapted for uv and your locallibrary project.

This tutorial builds one realistic API from start to finish: TaskFlow, a task tracker for teams. Users sign up, create projects, invite members with roles, and track tasks with assignees, labels, statuses, due dates, comments and file attachments. Each part adds a feature to the same codebase and shows the Django Ninja tools that make it work.

What you’ll build

By the end of Part 8, TaskFlow has these endpoints:

Method URL What it does
POST /taskflow/api/auth/signup Create an account
POST /taskflow/api/auth/token, /taskflow/api/auth/refresh Get and refresh JWT access tokens
GET, POST /taskflow/api/projects/ List your projects, create one
GET, PUT, DELETE /taskflow/api/projects/{project_id} Read, update and delete a project
POST /taskflow/api/projects/{project_id}/labels Add a label to a project
GET, POST /taskflow/api/projects/{project_id}/members List and add members
GET, POST /taskflow/api/projects/{project_id}/tasks/ Filter, search, sort and paginate tasks, create one
GET, PATCH /taskflow/api/projects/{project_id}/tasks/{task_id} Read a task, update part of it
POST /taskflow/api/projects/{project_id}/tasks/{task_id}/status Move a task through its workflow
GET, POST /taskflow/api/projects/{project_id}/tasks/{task_id}/comments List comments with a cursor, add one
PUT, DELETE /taskflow/api/projects/{project_id}/tasks/{task_id}/comments/{comment_id} Edit or delete a comment
GET, POST /taskflow/api/projects/{project_id}/tasks/{task_id}/attachments List attachments, upload a file
DELETE /taskflow/api/projects/{project_id}/tasks/{task_id}/attachments/{attachment_id} Delete an attachment

Every endpoint except the auth ones needs a token, and users only ever see the projects they’re members of. A second version of the API runs next to the first at /taskflow/api/v2/, and a pytest suite covers the rules that matter.

Before you start

  • You know Django: projects, apps, models, migrations and settings. The tutorial doesn’t explain those. It focuses on the Django Ninja parts.
  • You’ve done our Django Ninja Quickstart, so NinjaAPI, schemas and the interactive docs are familiar. That also means you’ve finished preps 01-06 and our Channels tutorial, and you have your locallibrary repository: a uv project with manage.py, pyproject.toml, and uv.lock at its root, the locallibrary_config project package, and the catalog and chat apps. Every command in this tutorial is run from that root folder.
  • You have Python 3.12 or newer. The code uses modern type hints such as int | None and list[int]. Your locallibrary uses Python 3.14.

The only packages TaskFlow needs are Django and Django Ninja, and both are already in your pyproject.toml: Django since Prep 01, and Django Ninja since the Quickstart. You can check:

uv run python -c "import django, ninja; print(django.__version__, ninja.__version__)"

Everything else, including JSON Web Tokens, is written with the standard library. Part 7 adds pytest and pytest-django as development tools. The database is the same SQLite db.sqlite3 your catalog and chat use, and all of TaskFlow’s views are synchronous.

How the parts work

  • Each part starts with a What you’ll learn box and takes about 15 minutes to read and type.
  • Each part builds on the code of the one before, so work through them in order. Changed files are shown in full, or as excerpts with ... where the rest of the file stays the same.
  • The example requests and responses are real output from the original tutorial. Your projects, labels, tasks, comments and attachments get new tables, so if you follow along exactly, their ids will match the ones on the pages. Your user ids won’t, because your database already has users. Part 2 explains what to do about that.
  • Go deeper links at the end of each section point to the Django Ninja Guide page that covers the feature in full.

The parts

  1. Project setup & layout: one router per app, schemas from models, and task URLs nested under projects.
  2. Relations & computed fields: assignees and labels as nested output, computed and annotated fields, and fewer queries.
  3. Validation & workflows: status and priority choices, field and model validators, PATCH updates and status transitions.
  4. Auth & permissions: JWT tokens from scratch, project memberships with roles, and object-level rules for comments.
  5. Filtering, search, ordering & pagination: a FilterSchema for tasks, safe sort orders, and page-number and cursor pagination.
  6. File uploads & forms: multipart uploads with form fields, file validation, media files and absolute URLs.
  7. Testing: a pytest suite with fixtures and TestClient, covering validation, permissions, filters and uploads.
  8. OpenAPI polish & versioning: descriptions, examples and deprecations in the docs, and a version 2 next to version 1.

Ready? Start with Part 1: Project setup & layout.

How this copy differs from the original

This is a copy of the official Django Ninja tutorial, changed to match the way we have worked in our fork of the MDN Django Tutorial, our Channels tutorial, and our Django Ninja Quickstart:

  • You don’t create a new project. The projects, tasks, and accounts apps go into your existing locallibrary repository, next to catalog and chat. Everywhere the original edits taskflow/settings.py or taskflow/urls.py, this copy edits locallibrary_config/settings.py or locallibrary_config/urls.py, and only adds lines.
  • taskflow is a plain package. In the original, taskflow is the Django project package. Here it’s a small package next to manage.py that holds only taskflow/api.py, so imports like from taskflow.api import api work exactly as in the original.
  • TaskFlow lives at /taskflow/api/, not /api/. Your Quickstart API already uses /api/. Every TaskFlow URL in this copy starts with /taskflow/api/, and version 2 is at /taskflow/api/v2/. Paths inside TestClient calls are relative to the API, so they’re the same as in the original.
  • Each TaskFlow NinjaAPI sets urls_namespace=. That keeps TaskFlow’s URL names apart from the Quickstart API’s. Part 8 explains why.
  • Dependencies are managed with uv. pip install becomes uv add (uv add --dev for pytest), and every command is run with uv run.
  • You never delete db.sqlite3. The original suggests it in Part 2 so the ids match. Yours holds your catalog and chat data, so this copy looks users up by username where the original uses their ids.
  • Part 6 adds media/ to your .gitignore. The original assumes it’s already there.
  • Part 7’s tests use your test_db.sqlite3 file. That’s the TEST database setting you added in Part 4 of the Channels tutorial.

This copy follows the Django Ninja documentation as of 2026-10-19 (Django Ninja 1.7.1), on Django 6.1 and Python 3.14.


The original tutorial is Copyright (c) 2025 Vitaliy Kucheryaviy, and is used under its MIT license.

3.1 - Part 1: Project setup & layout

One router per app, schemas from models, and task URLs nested under projects.

Over eight parts, you’ll build TaskFlow, a task tracker for teams: projects, members, tasks, labels, comments and file attachments. This first part sets up a project layout that stays tidy as the API grows. Each app gets its own router and schemas, and one small module wires them together.

Create the apps

You don’t create a new project: TaskFlow goes into your existing locallibrary repository, next to catalog and chat. Django and Django Ninja are already in your pyproject.toml from the Django Ninja Quickstart, so there’s nothing to install. From the folder with manage.py, create the two apps:

uv run python manage.py startapp projects
uv run python manage.py startapp tasks

TaskFlow doesn’t use Django’s template views, so you can delete views.py and tests.py from both apps. The tests come back in Part 7, in their own place.

In the original tutorial, taskflow is the name of the Django project package, the folder with settings.py and urls.py. Yours is called locallibrary_config, and you’ll keep it. TaskFlow still gets a taskflow package of its own, though, to hold the module that wires its routers together. It’s a plain Python package, not an app, so create it by hand: a taskflow folder next to manage.py, with an empty __init__.py file inside.

Register the two apps in locallibrary_config/settings.py, at the end of INSTALLED_APPS. 'ninja' is already there from the Quickstart. Don’t remove anything:

# locallibrary_config/settings.py
INSTALLED_APPS = [
    'daphne',
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'catalog.apps.CatalogConfig',
    'chat',
    'ninja',
    'projects',
    'tasks',
]

'ninja' is optional. With it, runserver serves the Swagger UI assets locally instead of from a CDN.

When this part is done, the new parts of your locallibrary will look like this:

catalog/            # (unchanged)
chat/               # (unchanged)
locallibrary_config/
├── settings.py
└── urls.py         # mounts the TaskFlow API next to the Quickstart's
taskflow/
├── __init__.py
└── api.py          # NinjaAPI: wires the routers together
projects/
├── api.py          # Router with the project endpoints
├── models.py
└── schemas.py
tasks/
├── api.py          # Router with the task endpoints
├── models.py
└── schemas.py
manage.py

The pattern is the same for every app: models in models.py, schemas in schemas.py, endpoints in api.py. The taskflow package holds only the wiring.

Models

A project, and a task that belongs to one:

# projects/models.py
from django.db import models


class Project(models.Model):
    name = models.CharField(max_length=100)
    description = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.name
# tasks/models.py
from django.db import models

from projects.models import Project


class Task(models.Model):
    project = models.ForeignKey(Project, related_name="tasks", on_delete=models.CASCADE)
    title = models.CharField(max_length=200)
    description = models.TextField(default="")
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.title
$ uv run python manage.py makemigrations
Migrations for 'projects':
  projects/migrations/0001_initial.py
    + Create model Project
Migrations for 'tasks':
  tasks/migrations/0001_initial.py
    + Create model Task
$ uv run python manage.py migrate

Tasks will get assignees, labels, a status and more in later parts. For now, a title is enough.

Schemas

Each app keeps its schemas in schemas.py. As in the Quick Start, there’s an In schema for what clients send and an Out schema for what they get back:

# projects/schemas.py
from ninja import ModelSchema

from .models import Project


class ProjectIn(ModelSchema):
    class Meta:
        model = Project
        fields = ["name", "description"]


class ProjectOut(ModelSchema):
    class Meta:
        model = Project
        fields = ["id", "name", "description", "created_at"]

ProjectIn is generated entirely from the model. Neither name nor description has a default, so both are required.

The task schemas follow the same pattern:

# tasks/schemas.py
from ninja import ModelSchema

from .models import Task


class TaskIn(ModelSchema):
    class Meta:
        model = Task
        fields = ["title", "description"]


class TaskOut(ModelSchema):
    class Meta:
        model = Task
        fields = ["id", "project", "title", "description", "created_at"]

Task.description has default="", so TaskIn makes it optional, with an empty string as the default. TaskIn has no project field. The project comes from the URL, as you’ll see below. In TaskOut, the project foreign key is serialized as the project’s id. Part 2 replaces ids like this with nested objects where that helps.

Go deeper: ModelSchema

The projects router

Each app’s endpoints live on a Router in the app’s api.py. Here’s full CRUD for projects:

# projects/api.py
from django.shortcuts import get_object_or_404
from ninja import Router, Status

from .models import Project
from .schemas import ProjectIn, ProjectOut

router = Router(tags=["projects"])


@router.get("/", response=list[ProjectOut])
def list_projects(request):
    return Project.objects.order_by("id")


@router.post("/", response={201: ProjectOut})
def create_project(request, payload: ProjectIn):
    return Status(201, Project.objects.create(**payload.dict()))


@router.get("/{project_id}", response=ProjectOut)
def get_project(request, project_id: int):
    return get_object_or_404(Project, id=project_id)


@router.put("/{project_id}", response=ProjectOut)
def update_project(request, project_id: int, payload: ProjectIn):
    project = get_object_or_404(Project, id=project_id)
    for attr, value in payload.dict().items():
        setattr(project, attr, value)
    project.save()
    return project


@router.delete("/{project_id}", response={204: None})
def delete_project(request, project_id: int):
    get_object_or_404(Project, id=project_id).delete()
    return Status(204, None)
  • The router doesn’t know where it will be mounted. Its paths are relative (/, /{project_id}), and the prefix is set when the project wires it in.
  • tags=["projects"] applies to every operation on the router. The interactive docs group endpoints by tag, so each app gets its own section.
  • Status(201, ...) picks the status code from the ones declared in response=. Status(204, None) sends an empty response.

Go deeper: Routers, Responses

Wire the routers together

The taskflow/api.py module creates the NinjaAPI and mounts each app’s router under a prefix:

# taskflow/api.py
from ninja import NinjaAPI

api = NinjaAPI(title="TaskFlow API", urls_namespace="taskflow")

api.add_router("/projects", "projects.api.router")
api.add_router("/projects/{project_id}/tasks", "tasks.api.router")

add_router() accepts the router object or its dotted import path. With the string form, taskflow/api.py has no imports from the apps, and each router module is imported only when add_router() runs. As the apps start importing each other’s models, that helps you avoid circular imports.

Your locallibrary_config/urls.py already mounts the Quickstart’s API at api/. Mount TaskFlow’s API once, like any other set of views, in the same style as the other blocks in that file. Add this to the end of the file:

# locallibrary_config/urls.py
# (after the existing blocks)
# Mount the TaskFlow API from the Django Ninja Tutorial
from taskflow.api import api as taskflow_api

urlpatterns += [
    path('taskflow/api/', taskflow_api.urls),
]

Both APIs are named api in their own modules, so this import renames TaskFlow’s to taskflow_api. Leave the admin, catalog, chat, redirect, static-files, and Quickstart api/ entries as they are.

The project routes now live at /taskflow/api/projects/ and /taskflow/api/projects/{project_id}. In the original tutorial they’re at /api/projects/. In this copy every TaskFlow URL starts with /taskflow/api/ instead, because /api/ already belongs to your Quickstart API.

Go deeper: The NinjaAPI Instance

Tasks under a project

Tasks always belong to a project, so their URLs sit under it: /taskflow/api/projects/{project_id}/tasks/. The second add_router() call above sets that up. Its prefix contains a {project_id} placeholder, and every operation on the tasks router can read it:

# tasks/api.py
from django.shortcuts import get_object_or_404
from ninja import Path, Router, Status

from projects.models import Project

from .models import Task
from .schemas import TaskIn, TaskOut

router = Router(tags=["tasks"])


@router.get("/", response=list[TaskOut])
def list_tasks(request, project_id: Path[int]):
    project = get_object_or_404(Project, id=project_id)
    return project.tasks.order_by("id")


@router.post("/", response={201: TaskOut})
def create_task(request, project_id: Path[int], payload: TaskIn):
    project = get_object_or_404(Project, id=project_id)
    return Status(201, Task.objects.create(project=project, **payload.dict()))


@router.get("/{task_id}", response=TaskOut)
def get_task(request, project_id: Path[int], task_id: int):
    return get_object_or_404(Task, id=task_id, project_id=project_id)
  • project_id: Path[int] reads the value from the URL prefix and converts it to an int.
  • task_id is part of the operation’s own path (/{task_id}), so a plain int is enough, like project_id in the projects router.
  • create_task takes the project from the URL, not from the body. A client can’t create a task in one project by posting it to another project’s URL.
  • get_task filters on both ids, so /projects/3/tasks/2 returns 404 when task 2 belongs to project 1.

Why not a nested router?

You could get the same URLs by nesting routers inside the projects app:

router.add_router("/{project_id}/tasks", tasks_router)

That works, but the projects app would then have to import the tasks app. Mounting both routers in taskflow/api.py keeps each app self-contained, and all the URL structure is in one file. Nested routers make more sense for sub-routers of a single app.

Go deeper: Path Parameters, Routers

Try it

uv run python manage.py runserver

The quickest way to send the requests below is the interactive docs at http://127.0.0.1:8000/taskflow/api/docs: click an endpoint, then Try it out. (You’ll see more of that page at the end of this section.)

Create a project with POST /taskflow/api/projects/. The body is {"name": "Website redesign", "description": "New marketing site"}, and the response is 201:

{
    "id": 1,
    "name": "Website redesign",
    "description": "New marketing site",
    "created_at": "2026-09-29T09:30:27.635Z"
}

A second project, {"name": "Mobile app", "description": "iOS and Android"}, gets id 2:

{
    "id": 2,
    "name": "Mobile app",
    "description": "iOS and Android",
    "created_at": "2026-09-29T09:30:27.636Z"
}

Leave out the required fields, as in {}, and you get a 422 that points at each missing one:

{
    "detail": [
        {
            "type": "missing",
            "loc": ["body", "payload", "name"],
            "msg": "Field required"
        },
        {
            "type": "missing",
            "loc": ["body", "payload", "description"],
            "msg": "Field required"
        }
    ]
}

Now add two tasks to project 1. POST /taskflow/api/projects/1/tasks/ with {"title": "Draft the sitemap"}, then again with {"title": "Pick a color palette", "description": "Two options"}. GET /taskflow/api/projects/1/tasks/ lists them:

[
    {
        "id": 1,
        "project": 1,
        "title": "Draft the sitemap",
        "description": "",
        "created_at": "2026-09-29T09:30:27.647Z"
    },
    {
        "id": 2,
        "project": 1,
        "title": "Pick a color palette",
        "description": "Two options",
        "created_at": "2026-09-29T09:30:27.649Z"
    }
]

Unknown ids, whether a missing project or a task under the wrong project, return a 404. GET /taskflow/api/projects/3/tasks/2:

{
    "detail": "Not Found: No Task matches the given query."
}

The message after Not Found only appears with DEBUG = True. In production, the body is just {"detail": "Not Found"}.

Open http://127.0.0.1:8000/taskflow/api/docs. The page title is TaskFlow API, and the endpoints are grouped under projects and tasks. Here’s the full list, with where each parameter comes from:

GET    /taskflow/api/projects/                                    ['projects']
POST   /taskflow/api/projects/                                    ['projects']
GET    /taskflow/api/projects/{project_id}                        ['projects']  project_id (path)
PUT    /taskflow/api/projects/{project_id}                        ['projects']  project_id (path)
DELETE /taskflow/api/projects/{project_id}                        ['projects']  project_id (path)
GET    /taskflow/api/projects/{project_id}/tasks/                 ['tasks']  project_id (path)
POST   /taskflow/api/projects/{project_id}/tasks/                 ['tasks']  project_id (path)
GET    /taskflow/api/projects/{project_id}/tasks/{task_id}        ['tasks']  project_id (path), task_id (path)

Go deeper: OpenAPI & Interactive Docs, Errors & Exception Handling

Recap

  • Each app owns a Router in api.py and its schemas in schemas.py. taskflow/api.py only creates the NinjaAPI and mounts routers.
  • add_router() sets the prefix and accepts a dotted path, which helps avoid circular imports. Router(tags=[...]) groups each app’s endpoints in the docs.
  • ModelSchema generates schemas from models. A model field without a default is required in the schema, and default="" makes it optional.
  • A router mounted under /projects/{project_id}/tasks reads the prefix parameter with project_id: Path[int]. Without Path, it would become a query parameter.
  • Status(code, body) picks the response status code.

Next

Tasks are still flat records. In Part 2: Relations & computed fields, you’ll add assignees and labels, return nested objects, compute task counts, and keep the number of queries under control.

3.2 - Part 2: Relations & computed fields

Assignees and labels as nested output, computed and annotated fields, and fewer queries.

In Part 1, a task was a flat record whose project came back as a bare id. Real tasks have an assignee and labels, and clients want to see who and what those are without making a request for each one. In this part you’ll add those relations, return them as nested objects, compute a few values the model doesn’t store, and check how many queries each list costs.

Assignees and labels

Labels belong to a project, so each project has its own set, like “bug” or “design”. The Label model goes in the projects app:

# projects/models.py
from django.db import models


class Project(models.Model):
    name = models.CharField(max_length=100)
    description = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.name


class Label(models.Model):
    project = models.ForeignKey(Project, related_name="labels", on_delete=models.CASCADE)
    name = models.CharField(max_length=50)
    color = models.CharField(max_length=7, default="#808080")

    class Meta:
        ordering = ["name"]

    def __str__(self):
        return self.name

A task gets an optional assignee, which is a Django user, and any number of labels:

# tasks/models.py
from django.conf import settings
from django.db import models

from projects.models import Label, Project


class Task(models.Model):
    project = models.ForeignKey(Project, related_name="tasks", on_delete=models.CASCADE)
    title = models.CharField(max_length=200)
    description = models.TextField(default="")
    assignee = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        null=True,
        blank=True,
        related_name="assigned_tasks",
        on_delete=models.SET_NULL,
    )
    labels = models.ManyToManyField(Label, blank=True, related_name="tasks")
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.title
$ uv run python manage.py makemigrations
Migrations for 'projects':
  projects/migrations/0002_label.py
    + Create model Label
Migrations for 'tasks':
  tasks/migrations/0002_task_assignee_task_labels.py
    + Add field assignee to task
    + Add field labels to task
$ uv run python manage.py migrate

TaskFlow has no user endpoints yet; sign-up and login come in Part 4. For now, create two users in the shell. Django’s shell imports your models, User included, automatically:

$ uv run python manage.py shell
>>> User.objects.create_user("alice", first_name="Alice", last_name="Martin")
<User: alice>
>>> User.objects.create_user("bob")
<User: bob>
>>> User.objects.get(username="alice").id, User.objects.get(username="bob").id
(4, 5)

Nested output, ids on input

By default, ModelSchema turns a foreign key into the related object’s id and a many-to-many field into a list of ids. That’s the right shape for input, since a client picks an assignee by id. For output, clients usually want the name as well, so they don’t have to look it up. The task schemas therefore use ids going in and nested objects coming out:

# tasks/schemas.py
from django.contrib.auth.models import User
from ninja import ModelSchema

from projects.schemas import LabelOut

from .models import Task


class UserOut(ModelSchema):
    display_name: str

    class Meta:
        model = User
        fields = ["id", "username"]

    @staticmethod
    def resolve_display_name(user):
        return user.get_full_name() or user.username


class TaskIn(ModelSchema):
    assignee_id: int | None = None
    label_ids: list[int] = []

    class Meta:
        model = Task
        fields = ["title", "description"]


class TaskOut(ModelSchema):
    assignee: UserOut | None
    labels: list[LabelOut]

    class Meta:
        model = Task
        fields = ["id", "project", "title", "description", "created_at"]
  • TaskOut declares assignee and labels with other schemas as their types. Django Ninja reads task.assignee and serializes it with UserOut, or as null when it’s empty. It reads task.labels, a related manager, and turns it into a list of LabelOut. You don’t need to call .all().
  • TaskIn takes assignee_id and label_ids, both optional. assignee_id matches the column name Django uses for the foreign key, so it can go straight into Task.objects.create().
  • display_name isn’t a field on User. resolve_display_name computes it from the object being serialized: the full name if there is one, or else the username. A resolver must be a @staticmethod named resolve_<field>.

LabelOut lives next to its model in projects/schemas.py, with a LabelIn for creating labels:

# projects/schemas.py
from ninja import ModelSchema

from .models import Label, Project


class LabelIn(ModelSchema):
    class Meta:
        model = Label
        fields = ["name", "color"]


class LabelOut(ModelSchema):
    class Meta:
        model = Label
        fields = ["id", "name", "color"]


class ProjectIn(ModelSchema):
    class Meta:
        model = Project
        fields = ["name", "description"]


class ProjectOut(ModelSchema):
    labels: list[LabelOut]
    task_count: int
    unassigned_task_count: int

    class Meta:
        model = Project
        fields = ["id", "name", "description", "created_at"]

color has a model default, so LabelIn makes it optional. The new ProjectOut fields are covered in Computed project fields below.

Go deeper: Schemas, ModelSchema

Creating tasks with relations

The ids in TaskIn come from the client, so check them before you use them. An unknown assignee_id would otherwise fail inside the database with an IntegrityError, and a label from another project doesn’t belong on this task:

# tasks/api.py
from django.contrib.auth.models import User
from django.shortcuts import get_object_or_404
from ninja import Path, Router, Status
from ninja.errors import HttpError

from projects.models import Project

from .models import Task
from .schemas import TaskIn, TaskOut

router = Router(tags=["tasks"])


def check_relations(project, payload):
    if payload.assignee_id is not None and not User.objects.filter(id=payload.assignee_id).exists():
        raise HttpError(400, f"User {payload.assignee_id} does not exist")
    labels = list(project.labels.filter(id__in=payload.label_ids))
    if len(labels) != len(set(payload.label_ids)):
        raise HttpError(400, "Unknown label id for this project")
    return labels


@router.get("/", response=list[TaskOut])
def list_tasks(request, project_id: Path[int]):
    project = get_object_or_404(Project, id=project_id)
    return project.tasks.select_related("assignee").prefetch_related("labels").order_by("id")


@router.post("/", response={201: TaskOut})
def create_task(request, project_id: Path[int], payload: TaskIn):
    project = get_object_or_404(Project, id=project_id)
    labels = check_relations(project, payload)
    task = Task.objects.create(project=project, **payload.dict(exclude={"label_ids"}))
    task.labels.set(labels)
    return Status(201, task)


@router.get("/{task_id}", response=TaskOut)
def get_task(request, project_id: Path[int], task_id: int):
    return get_object_or_404(Task, id=task_id, project_id=project_id)
  • check_relations returns the Label objects. It looks them up through project.labels, the reverse relation, so labels from other projects are never found.
  • HttpError(400, ...) stops the request with a {"detail": ...} body. Part 3 covers validation and errors in more depth.
  • A many-to-many relation needs a saved row, so it’s set after create(). payload.dict(exclude={"label_ids"}) leaves out the one key the model doesn’t accept.
  • The select_related and prefetch_related calls in list_tasks are explained in Counting queries.

Labels are created on their project with POST /taskflow/api/projects/{project_id}/labels. That endpoint is create_label at the end of projects/api.py, shown in full in the next section.

Try it

Create a project with POST /taskflow/api/projects/ and {"name": "Website redesign", "description": "New marketing site"}, then add two labels to it: POST /taskflow/api/projects/1/labels with {"name": "bug", "color": "#d73a4a"}, then with {"name": "design"}. The second label gets the default color:

{
    "id": 2,
    "name": "design",
    "color": "#808080"
}

Now create a task. POST /taskflow/api/projects/1/tasks/ with {"title": "Draft the sitemap", "assignee_id": 1, "label_ids": [2]}, using alice’s id in place of 1, returns 201, with ids in and objects out:

{
    "assignee": {
        "display_name": "Alice Martin",
        "id": 1,
        "username": "alice"
    },
    "labels": [
        {
            "id": 2,
            "name": "design",
            "color": "#808080"
        }
    ],
    "id": 1,
    "project": 1,
    "title": "Draft the sitemap",
    "description": "",
    "created_at": "2026-09-28T10:45:47.345Z"
}

Add two more: {"title": "Fix broken footer links", "assignee_id": 2, "label_ids": [1, 2]}, with bob’s id in place of 2, and {"title": "Pick a color palette"} with neither field. The last one comes back with "assignee": null and "labels": []. Bob has no full name, so his display_name is "bob".

Bad ids get a 400. {"title": "Ghost", "assignee_id": 42} returns this, as long as none of your users has id 42:

{
    "detail": "User 42 does not exist"
}

To see the label check, create a second project, {"name": "Mobile app", "description": "iOS and Android"}, and give it a label, {"name": "ios"}. That label gets id 3. Back on project 1, {"title": "Wrong label", "label_ids": [3]} returns:

{
    "detail": "Unknown label id for this project"
}

Go deeper: Errors & Exception Handling

Computed project fields

A project list is more useful with a count of tasks next to each project. You could compute that in a resolver with project.tasks.count(), but that runs one query per project. Let the database compute it for the whole list with annotate():

# projects/api.py
from django.db.models import Count, Q
from django.shortcuts import get_object_or_404
from ninja import Router, Status

from .models import Project
from .schemas import LabelIn, LabelOut, ProjectIn, ProjectOut

router = Router(tags=["projects"])


def projects_with_counts():
    return Project.objects.annotate(
        task_count=Count("tasks"),
        unassigned_task_count=Count("tasks", filter=Q(tasks__assignee=None)),
    ).prefetch_related("labels")


@router.get("/", response=list[ProjectOut])
def list_projects(request):
    return projects_with_counts().order_by("id")


@router.post("/", response={201: ProjectOut})
def create_project(request, payload: ProjectIn):
    project = Project.objects.create(**payload.dict())
    return Status(201, projects_with_counts().get(id=project.id))


@router.get("/{project_id}", response=ProjectOut)
def get_project(request, project_id: int):
    return get_object_or_404(projects_with_counts(), id=project_id)


@router.put("/{project_id}", response=ProjectOut)
def update_project(request, project_id: int, payload: ProjectIn):
    project = get_object_or_404(projects_with_counts(), id=project_id)
    for attr, value in payload.dict().items():
        setattr(project, attr, value)
    project.save()
    return project


@router.delete("/{project_id}", response={204: None})
def delete_project(request, project_id: int):
    get_object_or_404(Project, id=project_id).delete()
    return Status(204, None)


@router.post("/{project_id}/labels", response={201: LabelOut})
def create_label(request, project_id: int, payload: LabelIn):
    project = get_object_or_404(Project, id=project_id)
    return Status(201, project.labels.create(**payload.dict()))
  • An annotation becomes an attribute on each instance, so ProjectOut reads task_count and unassigned_task_count like any other field. The schema doesn’t need to know they came from annotate().
  • labels: list[LabelOut] on ProjectOut reads the reverse relation project.labels. ModelSchema never includes reverse relations on its own, so you declare them yourself.
  • Every view that returns a ProjectOut gets its project from projects_with_counts(). create_project fetches the new project again for that reason, and update_project keeps the annotated instance it loaded.
  • create_label creates the label through project.labels, which sets its project for you.

If a view returns a plain Project, you don’t get a silent 0. The response fails validation with Field required for task_count and unassigned_task_count, and the client gets a server error. It’s easy to spot in development.

GET /taskflow/api/projects/1, with the three tasks from above:

{
    "labels": [
        {
            "id": 1,
            "name": "bug",
            "color": "#d73a4a"
        },
        {
            "id": 2,
            "name": "design",
            "color": "#808080"
        }
    ],
    "task_count": 3,
    "unassigned_task_count": 1,
    "id": 1,
    "name": "Website redesign",
    "description": "New marketing site",
    "created_at": "2026-09-28T10:45:47.335Z"
}

Go deeper: Schemas, ModelSchema

Counting queries

Nested output is convenient, but each nested object comes from somewhere. In Part 1, list_tasks returned project.tasks.order_by("id"). With the new TaskOut, every task in that list costs one extra query for its assignee and one for its labels. This is the N+1 problem.

To see it, temporarily change the last line of list_tasks back to Part 1’s return project.tasks.order_by("id"). Then count the queries from the shell. ninja.testing.TestClient calls the API directly, without a running server. Part 7 uses it for the test suite.

$ uv run python manage.py shell
>>> from django.db import connection
>>> from django.test.utils import CaptureQueriesContext
>>> from ninja.testing import TestClient
>>> from taskflow.api import api
>>> client = TestClient(api)
>>> with CaptureQueriesContext(connection) as ctx:
...     response = client.get("/projects/1/tasks/")
...
>>> len(ctx.captured_queries)
7

That’s 7 queries for three tasks with Part 1’s queryset: one for the project, one for the tasks, two for the assignees (the third task has none), and three for the labels. Every extra task adds up to two more, so ten tasks take 21 or 22. The fix is to fetch the related rows up front, which is the version of list_tasks you typed earlier:

# tasks/api.py
@router.get("/", response=list[TaskOut])
def list_tasks(request, project_id: Path[int]):
    project = get_object_or_404(Project, id=project_id)
    return project.tasks.select_related("assignee").prefetch_related("labels").order_by("id")
  • select_related("assignee") joins the user table into the tasks query. Use it for foreign keys.
  • prefetch_related("labels") loads the labels of all the listed tasks in one extra query. Use it for many-to-many and reverse relations.

Restore that line, open a new shell, and run the same code again:

>>> len(ctx.captured_queries)
3

It’s 3 queries, for three tasks or ten: the project, the tasks joined with their assignees, and the labels. The projects list works the same way. projects_with_counts() computes both counts in the main query and adds .prefetch_related("labels"), so GET /taskflow/api/projects/ takes 2 queries however many projects there are.

Go deeper: Schemas, Responses

Recap

  • Declare a related field with another schema as its type (assignee: UserOut | None, labels: list[LabelOut]) to nest it in the output. Related managers are turned into lists for you.
  • Accept plain ids on input (assignee_id, label_ids), check them, and set many-to-many relations after create().
  • A @staticmethod named resolve_<field> computes a field in Python. An annotate() in the queryset computes it in the database, and the schema reads it like any other attribute.
  • Reverse relations such as project.labels are never generated by ModelSchema. Declare them on the schema.
  • Nested output can cause N+1 queries. Use select_related for foreign keys and prefetch_related for many-to-many and reverse relations, and count queries with CaptureQueriesContext.

Next

Tasks still have no status, and nothing stops a client from sending a blank title. In Part 3: Validation & workflows, you’ll add status and priority choices, field and model validators, and a status transition endpoint.

3.3 - Part 3: Validation & workflows

Status and priority choices, field and model validators, PATCH updates and status transitions.

So far, TaskFlow accepts almost anything a client sends: a title made of spaces is a valid task, and there’s no way to say a task is finished. In this part you’ll give tasks a status, a priority and a due date, validate input with field and model validators, add a PATCH endpoint for partial updates, and move tasks through a small workflow that rejects invalid status changes.

Status, priority and due date

Status and priority are fixed sets of values, so they’re Django choices. The allowed status changes go right next to them:

# tasks/models.py
from django.conf import settings
from django.db import models

from projects.models import Label, Project


class TaskStatus(models.TextChoices):
    TODO = "todo"
    IN_PROGRESS = "in_progress"
    DONE = "done"


class Priority(models.IntegerChoices):
    """1 = low, 2 = medium, 3 = high, 4 = urgent"""

    LOW = 1
    MEDIUM = 2
    HIGH = 3
    URGENT = 4


# The statuses a task may move to from each status
ALLOWED_TRANSITIONS = {
    TaskStatus.TODO: {TaskStatus.IN_PROGRESS},
    TaskStatus.IN_PROGRESS: {TaskStatus.TODO, TaskStatus.DONE},
    TaskStatus.DONE: {TaskStatus.IN_PROGRESS},
}


class Task(models.Model):
    project = models.ForeignKey(Project, related_name="tasks", on_delete=models.CASCADE)
    title = models.CharField(max_length=200)
    description = models.TextField(default="")
    status = models.CharField(max_length=20, choices=TaskStatus, default=TaskStatus.TODO)
    priority = models.IntegerField(choices=Priority, default=Priority.MEDIUM)
    due_date = models.DateField(null=True, blank=True)
    assignee = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        null=True,
        blank=True,
        related_name="assigned_tasks",
        on_delete=models.SET_NULL,
    )
    labels = models.ManyToManyField(Label, blank=True, related_name="tasks")
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.title
  • Priority is an integer, so that “most urgent first” is a plain order_by("-priority") later on.
  • ALLOWED_TRANSITIONS is the whole workflow: from todo you can only start a task, and a done task can only be reopened as in_progress. You’ll use it in Status transitions.
$ uv run python manage.py makemigrations
Migrations for 'tasks':
  tasks/migrations/0003_task_due_date_task_priority_task_status.py
    + Add field due_date to task
    + Add field priority to task
    + Add field status to task
$ uv run python manage.py migrate

The tasks you created in Part 2 get the model defaults. GET /taskflow/api/projects/1/tasks/1 now includes the new fields:

{
    "assignee": {
        "display_name": "Alice Martin",
        "id": 1,
        "username": "alice"
    },
    "labels": [
        {
            "id": 2,
            "name": "design",
            "color": "#808080"
        }
    ],
    "status": "todo",
    "priority": 2,
    "id": 1,
    "project": 1,
    "title": "Draft the sitemap",
    "description": "",
    "due_date": null,
    "created_at": "2026-09-28T10:53:40.574Z"
}

Validating input

ModelSchema doesn’t look at choices: a status field from Meta.fields is just a str, and priority is just an int. To have the schema accept only valid values, declare the field with the choices class as its type. TextChoices and IntegerChoices are Python enums, which Pydantic validates and OpenAPI describes.

The rest of the rules go into validators. Here’s the new tasks/schemas.py:

# tasks/schemas.py
from django.contrib.auth.models import User
from ninja import ModelSchema, Schema
from pydantic import field_validator, model_validator
from pydantic_core import PydanticCustomError

from projects.schemas import LabelOut

from .models import Priority, Task, TaskStatus


class UserOut(ModelSchema):
    display_name: str

    class Meta:
        model = User
        fields = ["id", "username"]

    @staticmethod
    def resolve_display_name(user):
        return user.get_full_name() or user.username


class TaskUpdate(ModelSchema):
    priority: Priority = Priority.MEDIUM

    class Meta:
        model = Task
        fields = ["title", "description", "priority", "due_date"]

    @field_validator("title", "description", "priority", check_fields=False)
    @classmethod
    def not_null(cls, value):
        if value is None:
            raise PydanticCustomError("not_null", "This field can't be null")
        return value

    @field_validator("title", check_fields=False)
    @classmethod
    def title_not_blank(cls, value: str) -> str:
        value = value.strip()
        if not value:
            raise PydanticCustomError("blank", "Title can't be blank")
        return value


class TaskIn(TaskUpdate):
    assignee_id: int | None = None
    label_ids: list[int] = []

    @model_validator(mode="after")
    def urgent_needs_due_date(self):
        if self.priority == Priority.URGENT and self.due_date is None:
            raise ValueError("Urgent tasks need a due date")
        return self


class StatusIn(Schema):
    status: TaskStatus


class TaskOut(ModelSchema):
    assignee: UserOut | None
    labels: list[LabelOut]
    status: TaskStatus
    priority: Priority

    class Meta:
        model = Task
        fields = ["id", "project", "title", "description", "due_date", "created_at"]
  • TaskUpdate holds the fields a client can edit: title, description, priority and due date. TaskIn extends it with the fields you only set on create. A ModelSchema subclass inherits Meta, fields and validators, so TaskIn has everything TaskUpdate has.
  • status isn’t in either input schema. New tasks always start as todo, and the only way to change the status is the transition endpoint below.
  • title_not_blank is a field validator. It runs after the type check, gets the value, and returns what should be stored, here the stripped title.
  • not_null does nothing on a normal request, because the type check already rejects null for these fields. It exists for the PATCH endpoint, explained in the next section.
  • urgent_needs_due_date is a model validator. With mode="after" it runs once every field is valid, so it can compare fields. It returns self.
  • StatusIn is the body of the transition endpoint. TaskOut declares status and priority with their enums so the response schema lists the allowed values too.

Error messages

A validator reports a problem by raising an exception, and Django Ninja turns it into a 422 response. What the client sees depends on what you raise:

  • ValueError("...") gives the type value_error, and Pydantic adds Value error, before your message.
  • PydanticCustomError(type, message) uses your type and your message exactly as written. Clients can match on the type.

Try it

POST /taskflow/api/projects/1/tasks/ with {"title": " Write launch announcement ", "priority": 3, "due_date": "2026-10-15"} returns 201, with the title stripped:

{
    "assignee": null,
    "labels": [],
    "status": "todo",
    "priority": 3,
    "id": 4,
    "project": 1,
    "title": "Write launch announcement",
    "description": "",
    "due_date": "2026-10-15",
    "created_at": "2026-09-28T10:53:40.592Z"
}

Invalid input gets a 422 with a list of errors. When several fields are wrong, you get all of them in one response. {"title": "", "priority": 5, "due_date": "next friday"} returns:

{
    "detail": [
        {
            "type": "enum",
            "loc": [
                "body",
                "payload",
                "priority"
            ],
            "msg": "Input should be 1, 2, 3 or 4",
            "ctx": {
                "expected": "1, 2, 3 or 4"
            }
        },
        {
            "type": "blank",
            "loc": [
                "body",
                "payload",
                "title"
            ],
            "msg": "Title can't be blank"
        },
        {
            "type": "date_from_datetime_parsing",
            "loc": [
                "body",
                "payload",
                "due_date"
            ],
            "msg": "Input should be a valid date or datetime, invalid character in year",
            "ctx": {
                "error": "invalid character in year"
            }
        }
    ]
}

loc is the path to each bad value: the request body, the view argument payload, then the field. type and msg of the title error come from the PydanticCustomError in title_not_blank.

The model validator runs only when every field is valid. {"title": "Fix checkout bug", "priority": 4} has no due date, so it gets an error for the whole payload, with the Value error, prefix that comes from ValueError:

{
    "detail": [
        {
            "type": "value_error",
            "loc": [
                "body",
                "payload"
            ],
            "msg": "Value error, Urgent tasks need a due date",
            "ctx": {
                "error": "Urgent tasks need a due date"
            }
        }
    ]
}

None of these requests reach the view, so nothing is saved.

In the interactive docs at /taskflow/api/docs, the schemas now include both enums. Priority uses the class docstring as its description, which is the easiest way to explain what the numbers mean:

{
    "description": "1 = low, 2 = medium, 3 = high, 4 = urgent",
    "enum": [
        1,
        2,
        3,
        4
    ],
    "title": "Priority",
    "type": "integer"
}

Go deeper: Schemas, Errors & Exception Handling

Partial updates with PATCH

Part 1’s PUT for projects replaces every field. For tasks, a client usually changes one thing at a time, such as the priority. That’s a PATCH, and PatchDict handles it. Here’s the full tasks/api.py with the two new endpoints:

# tasks/api.py
from django.contrib.auth.models import User
from django.shortcuts import get_object_or_404
from ninja import PatchDict, Path, Router, Status
from ninja.errors import HttpError

from projects.models import Project

from .models import ALLOWED_TRANSITIONS, Priority, Task
from .schemas import StatusIn, TaskIn, TaskOut, TaskUpdate

router = Router(tags=["tasks"])


def check_relations(project, payload):
    if payload.assignee_id is not None and not User.objects.filter(id=payload.assignee_id).exists():
        raise HttpError(400, f"User {payload.assignee_id} does not exist")
    labels = list(project.labels.filter(id__in=payload.label_ids))
    if len(labels) != len(set(payload.label_ids)):
        raise HttpError(400, "Unknown label id for this project")
    return labels


@router.get("/", response=list[TaskOut])
def list_tasks(request, project_id: Path[int]):
    project = get_object_or_404(Project, id=project_id)
    return project.tasks.select_related("assignee").prefetch_related("labels").order_by("id")


@router.post("/", response={201: TaskOut})
def create_task(request, project_id: Path[int], payload: TaskIn):
    project = get_object_or_404(Project, id=project_id)
    labels = check_relations(project, payload)
    task = Task.objects.create(project=project, **payload.dict(exclude={"label_ids"}))
    task.labels.set(labels)
    return Status(201, task)


@router.get("/{task_id}", response=TaskOut)
def get_task(request, project_id: Path[int], task_id: int):
    return get_object_or_404(Task, id=task_id, project_id=project_id)


@router.patch("/{task_id}", response=TaskOut)
def update_task(request, project_id: Path[int], task_id: int, payload: PatchDict[TaskUpdate]):
    task = get_object_or_404(Task, id=task_id, project_id=project_id)
    for attr, value in payload.items():
        setattr(task, attr, value)
    if task.priority == Priority.URGENT and task.due_date is None:
        raise HttpError(400, "Urgent tasks need a due date")
    task.save()
    return task


@router.post("/{task_id}/status", response=TaskOut)
def change_status(request, project_id: Path[int], task_id: int, payload: StatusIn):
    task = get_object_or_404(Task, id=task_id, project_id=project_id)
    if payload.status not in ALLOWED_TRANSITIONS[task.status]:
        raise HttpError(409, f"Can't move a task from {task.status} to {payload.status}")
    task.status = payload.status
    task.save()
    return task
  • PatchDict[TaskUpdate] makes every field of TaskUpdate optional and gives the view a plain dict with only the keys the client sent. The loop copies those onto the task and leaves the rest alone.
  • The field validators still run on the values that were sent, so a blank title is rejected on PATCH as well.
  • create_task didn’t change. payload.dict() now includes priority and due_date, and Task.objects.create() accepts them as they are.

PatchDict makes every field optional by also allowing null. For due_date, null is useful: it clears the date. For title, description and priority, the database column can’t hold NULL, and a null would reach task.save() and fail with an IntegrityError. The not_null validator stops it first. PATCH /taskflow/api/projects/1/tasks/4 with {"title": null} returns:

{
    "detail": [
        {
            "type": "not_null",
            "loc": [
                "body",
                "payload",
                "title"
            ],
            "msg": "This field can't be null"
        }
    ]
}

Rules that need the stored data

PATCH uses TaskUpdate, not TaskIn, so the urgent rule doesn’t run as a model validator there. It couldn’t work on a partial payload anyway: a validator only sees the request body, and {"priority": 4} doesn’t include the due date the task already has. So update_task checks the rule on the task itself, after the changes are applied and before anything is saved.

Task 4 has a due date, so PATCH /taskflow/api/projects/1/tasks/4 with {"priority": 4} returns 200 with "priority": 4, and the title and due date unchanged.

Clearing the due date of that urgent task with {"due_date": null} returns 400, and the task keeps its date:

{
    "detail": "Urgent tasks need a due date"
}

This error comes from HttpError, so it has the simple {"detail": "..."} shape rather than a list. Use validators for what the request alone can decide, and checks in the view for anything that depends on the database.

Go deeper: Request Body, ModelSchema

Status transitions

A status isn’t just another field to PATCH. Whether done is a valid new status depends on the current one. That’s why status has its own endpoint, POST /taskflow/api/projects/{project_id}/tasks/{task_id}/status:

# tasks/api.py
@router.post("/{task_id}/status", response=TaskOut)
def change_status(request, project_id: Path[int], task_id: int, payload: StatusIn):
    task = get_object_or_404(Task, id=task_id, project_id=project_id)
    if payload.status not in ALLOWED_TRANSITIONS[task.status]:
        raise HttpError(409, f"Can't move a task from {task.status} to {payload.status}")
    task.status = payload.status
    task.save()
    return task
  • StatusIn makes sure the value is one of the three statuses. The view only has to decide whether this change is allowed.
  • ALLOWED_TRANSITIONS[task.status] is the set of statuses the task can move to. task.status is a plain string when it’s loaded from the database, and it still finds the TaskStatus key, because a TextChoices member compares and hashes like its string value.
  • 409 Conflict says the request was well formed but clashes with the current state of the task. The message tells the client which move was refused.

Task 4 is still todo, so {"status": "done"} is refused:

{
    "detail": "Can't move a task from todo to done"
}

Starting it first with {"status": "in_progress"} returns 200 and the task with "status": "in_progress". After that, {"status": "done"} works. A status that doesn’t exist never reaches the view. {"status": "archived"} returns a 422 from the enum:

{
    "detail": [
        {
            "type": "enum",
            "loc": [
                "body",
                "payload",
                "status"
            ],
            "msg": "Input should be 'todo', 'in_progress' or 'done'",
            "ctx": {
                "expected": "'todo', 'in_progress' or 'done'"
            }
        }
    ]
}

Go deeper: Errors & Exception Handling

Recap

  • Declare TextChoices and IntegerChoices fields with the choices class as their type. The schema then accepts only valid values, and OpenAPI lists them.
  • A @field_validator checks and cleans one field. A @model_validator(mode="after") compares fields once they’re all valid. Validators on Meta.fields need check_fields=False.
  • Raise PydanticCustomError for your own error type and message, or ValueError for a quick value_error. Either way the client gets a 422 with every problem listed.
  • PatchDict[Schema] gives the view a dict of only the fields sent. It also accepts null for every field, so reject it where the column can’t be empty.
  • Validators only see the request. Check rules that depend on stored data, such as allowed status changes, in the view and raise HttpError.

Next

Anyone can still read and change every project. In Part 4: Auth & permissions, you’ll add JWT authentication, project memberships with roles, and comments that only their author can edit.

3.4 - Part 4: Auth & permissions

JWT tokens from scratch, project memberships with roles, and object-level rules for comments.

Right now anyone who can reach the API can read and change every project. In this part you’ll add JSON Web Tokens, written from scratch with the standard library, require them on every endpoint, and then decide what each user may do: project memberships with roles, queries scoped to the user’s projects, and comments that only their author can edit.

The accounts app

Users, tokens and the login endpoints get their own app:

uv run python manage.py startapp accounts

Delete views.py and tests.py again, and add the app to the end of INSTALLED_APPS:

# locallibrary_config/settings.py
INSTALLED_APPS = [
    'daphne',
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'catalog.apps.CatalogConfig',
    'chat',
    'ninja',
    'projects',
    'tasks',
    'accounts',
]

This accounts app has nothing to do with the accounts/ URLs (accounts/login/ and friends) that you added in Prep 06. Those come from django.contrib.auth.urls, keep working as before, and log users in with a session. TaskFlow’s accounts app logs users in with tokens instead. Both use the same User table, so a user can log in either way.

UserOut from Part 2 moves here too. The projects app needs it in this part, and projects/schemas.py can’t import from tasks/schemas.py, which already imports LabelOut from projects/schemas.py. Cut it from tasks/schemas.py, along with its User import, and import it from the new module instead, so TaskOut keeps working:

# tasks/schemas.py
from accounts.schemas import UserOut

Then create accounts/schemas.py:

# accounts/schemas.py
from django.contrib.auth.models import User
from ninja import ModelSchema, Schema


class UserOut(ModelSchema):
    display_name: str

    class Meta:
        model = User
        fields = ["id", "username"]

    @staticmethod
    def resolve_display_name(user):
        return user.get_full_name() or user.username


class SignupIn(Schema):
    username: str
    password: str


class TokenIn(Schema):
    username: str
    password: str


class RefreshIn(Schema):
    refresh: str


class TokenOut(Schema):
    access: str
    refresh: str

JSON Web Tokens from scratch

A JWT is three base64url strings joined by dots: a header, a payload of claims, and a signature over the first two. With HS256, the signature is an HMAC-SHA256 made with a secret key, here settings.SECRET_KEY. Anyone can read the claims, but no one can change them without the key.

That’s short enough to write yourself:

# accounts/tokens.py
import base64
import hashlib
import hmac
import json
import time

from django.conf import settings

ACCESS_TOKEN_LIFETIME = 15 * 60  # 15 minutes
REFRESH_TOKEN_LIFETIME = 7 * 24 * 60 * 60  # 7 days


def b64encode(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()


def b64decode(data: str) -> bytes:
    return base64.urlsafe_b64decode(data + "=" * (-len(data) % 4))


def sign(message: str) -> str:
    digest = hmac.new(settings.SECRET_KEY.encode(), message.encode(), hashlib.sha256).digest()
    return b64encode(digest)


def create_token(user_id: int, token_type: str, lifetime: int) -> str:
    header = b64encode(json.dumps({"alg": "HS256", "typ": "JWT"}).encode())
    claims = {"sub": str(user_id), "type": token_type, "exp": int(time.time()) + lifetime}
    payload = b64encode(json.dumps(claims).encode())
    return f"{header}.{payload}.{sign(f'{header}.{payload}')}"


def decode_token(token: str, token_type: str) -> dict | None:
    """Return the claims of a valid, unexpired token of the given type, else None."""
    try:
        header, payload, signature = token.split(".")
    except ValueError:
        return None
    if not hmac.compare_digest(signature.encode(), sign(f"{header}.{payload}").encode()):
        return None
    claims = json.loads(b64decode(payload))
    if claims["type"] != token_type or claims["exp"] < time.time():
        return None
    return claims


def create_token_pair(user_id: int) -> dict:
    return {
        "access": create_token(user_id, "access", ACCESS_TOKEN_LIFETIME),
        "refresh": create_token(user_id, "refresh", REFRESH_TOKEN_LIFETIME),
    }
  • JWTs use base64url without the = padding. b64decode adds the padding back before decoding.
  • sub (the user id) and exp (expiry, in Unix seconds) are standard claims. type is a claim of your own that tells access and refresh tokens apart, so a refresh token can’t be used to call the API.
  • decode_token checks the signature before it reads the payload, so every claim it looks at was signed by you. hmac.compare_digest compares in constant time, which doesn’t leak how many characters matched.
  • The header is signed, but never trusted: the code always verifies with HS256, whatever alg the token claims. That rules out the classic "alg": "none" attack.
  • Anything wrong, whether a malformed token, a bad signature, the wrong type or an expired token, returns None. The caller doesn’t need to know which.

The access token lives 15 minutes. The refresh token lives 7 days and can only be exchanged for a new pair.

Protecting the API

HttpBearer reads the Authorization: Bearer <token> header and passes the token to your authenticate() method. Whatever it returns becomes request.auth. None means the request is rejected with 401:

# accounts/auth.py
from django.contrib.auth.models import User
from ninja.security import HttpBearer

from .tokens import decode_token


class JWTAuth(HttpBearer):
    def authenticate(self, request, token):
        claims = decode_token(token, "access")
        if claims is None:
            return None
        return User.objects.filter(id=claims["sub"], is_active=True).first()

It returns the User, so every view can use request.auth as the current user. A deactivated user’s tokens stop working at once, even before they expire.

Set it on the NinjaAPI, so every operation requires a token, and mount the new router:

# taskflow/api.py
from ninja import NinjaAPI

from accounts.auth import JWTAuth

api = NinjaAPI(title="TaskFlow API", urls_namespace="taskflow", auth=JWTAuth())

api.add_router("/auth", "accounts.api.router")
api.add_router("/projects", "projects.api.router")
api.add_router("/projects/{project_id}/tasks", "tasks.api.router")

The auth endpoints themselves must work without a token. auth=None on their router opts them out:

# accounts/api.py
from django.contrib.auth import authenticate
from django.contrib.auth.models import User
from django.contrib.auth.password_validation import validate_password
from django.core.exceptions import ValidationError
from ninja import Router, Status
from ninja.errors import HttpError

from .schemas import RefreshIn, SignupIn, TokenIn, TokenOut, UserOut
from .tokens import create_token_pair, decode_token

router = Router(tags=["auth"], auth=None)


@router.post("/signup", response={201: UserOut})
def signup(request, payload: SignupIn):
    if User.objects.filter(username=payload.username).exists():
        raise HttpError(409, "This username is taken")
    try:
        validate_password(payload.password)
    except ValidationError as e:
        raise HttpError(400, " ".join(e.messages))
    user = User.objects.create_user(payload.username, password=payload.password)
    return Status(201, user)


@router.post("/token", response=TokenOut)
def obtain_token(request, payload: TokenIn):
    user = authenticate(username=payload.username, password=payload.password)
    if user is None:
        raise HttpError(401, "Invalid username or password")
    return create_token_pair(user.id)


@router.post("/refresh", response=TokenOut)
def refresh_token(request, payload: RefreshIn):
    claims = decode_token(payload.refresh, "refresh")
    if claims is None or not User.objects.filter(id=claims["sub"], is_active=True).exists():
        raise HttpError(401, "Invalid or expired refresh token")
    return create_token_pair(int(claims["sub"]))
  • auth resolves from the most specific level: operation, then router, then API. auth=None is how you turn it off for part of an API that sets it globally.
  • signup runs Django’s AUTH_PASSWORD_VALIDATORS and returns their messages as a 400.
  • obtain_token uses Django’s authenticate(), so it respects inactive users and any authentication backends you configure.

Try it

Alice and Bob from Part 2 have no passwords yet. Set them in the shell:

$ uv run python manage.py shell
>>> alice = User.objects.get(username="alice")
>>> alice.set_password("wonderland-42")
>>> alice.save()
>>> bob = User.objects.get(username="bob")
>>> bob.set_password("builder-42")
>>> bob.save()

Without a token, every endpoint except the three under /taskflow/api/auth/ now returns 401:

$ curl http://127.0.0.1:8000/taskflow/api/projects/
{"detail": "Unauthorized"}

Log in:

$ curl -X POST http://127.0.0.1:8000/taskflow/api/auth/token -H "Content-Type: application/json" -d '{"username": "alice", "password": "wonderland-42"}'
{"access": "eyJhbGciOiAiSFMyNTYiLCAidHlwIjogIkpXVCJ9.eyJzdWIiOiAiMSIsICJ0eXBlIjogImFjY2VzcyIsICJleHAiOiAxNzkwNTk0MTkwfQ.GaiXckAgDYsFg8sT9aw69lVVvTQicl6jgm8B6EOvkec", "refresh": "eyJhbGciOiAiSFMyNTYiLCAidHlwIjogIkpXVCJ9.eyJzdWIiOiAiMSIsICJ0eXBlIjogInJlZnJlc2giLCAiZXhwIjogMTc5MTE5ODA5MH0.2T6GPT9KOTlhyGKNlGgs74W1LmviG5UneZA_K4gwgvc"}

A wrong password gives 401 with {"detail": "Invalid username or password"}. The middle part of the access token is the payload, readable by anyone:

>>> from accounts.tokens import b64decode
>>> b64decode("eyJzdWIiOiAiMSIsICJ0eXBlIjogImFjY2VzcyIsICJleHAiOiAxNzkwNTk0MTkwfQ")
b'{"sub": "1", "type": "access", "exp": 1790594190}'

Your sub is alice’s id, not 1, so your token won’t match the one above.

Send the access token in the Authorization header. With TOKEN set to the access value, GET /taskflow/api/projects/ returns the projects again:

$ curl http://127.0.0.1:8000/taskflow/api/projects/ -H "Authorization: Bearer $TOKEN"

When it expires, post the refresh token as {"refresh": "..."} to /taskflow/api/auth/refresh for a new pair. An access token is refused there, because its type claim is wrong:

{
    "detail": "Invalid or expired refresh token"
}

In the interactive docs at /taskflow/api/docs, the operations now show a lock, and an Authorize button appears at the top. Paste an access token there (without the Bearer prefix) and Swagger UI sends it with every request you try. The three auth endpoints have no lock, because they have no auth.

TaskFlow ignores the session cookie that the Quickstart’s API uses. Even if you’re logged in to the admin or at /accounts/login/, TaskFlow only knows who you are from the bearer token.

Go deeper: Authentication, Authentication: auth at different levels

Memberships and roles

Authentication tells you who is calling. Authorization decides what they may do. In TaskFlow that depends on the project: a user can be the owner of one project and a plain member of another. That’s a Membership with a role. Add it between Project and Label, along with the settings import:

# projects/models.py
from django.conf import settings
from django.db import models

...

class Role(models.TextChoices):
    OWNER = "owner"
    ADMIN = "admin"
    MEMBER = "member"


class Membership(models.Model):
    project = models.ForeignKey(Project, related_name="memberships", on_delete=models.CASCADE)
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL, related_name="memberships", on_delete=models.CASCADE
    )
    role = models.CharField(max_length=10, choices=Role, default=Role.MEMBER)

    class Meta:
        constraints = [
            models.UniqueConstraint(fields=["project", "user"], name="unique_membership"),
        ]

    def __str__(self):
        return f"{self.user} in {self.project} ({self.role})"
$ uv run python manage.py makemigrations
Migrations for 'projects':
  projects/migrations/0003_membership.py
    + Create model Membership
$ uv run python manage.py migrate

The roles in TaskFlow:

Role Can
member see the project, work on its tasks, comment
admin also edit the project and create labels
owner also delete the project, add members, and delete anyone’s comments

Your existing projects have no members yet, so once the next step is in place nobody would see them. Make Alice their owner:

$ uv run python manage.py shell
>>> alice = User.objects.get(username="alice")
>>> for project in Project.objects.all():
...     Membership.objects.create(project=project, user=alice, role="owner")
...
<Membership: alice in Website redesign (owner)>
<Membership: alice in Mobile app (owner)>

One helper for every check

Almost every endpoint needs the same check: is this user a member of this project, and does their role allow the action? Put it in one function:

# projects/permissions.py
from django.shortcuts import get_object_or_404
from ninja.errors import HttpError

from .models import Membership


def get_project_or_404(user, project_id, roles=None):
    """Return the project if `user` is a member, with one of `roles` if given.

    Non-members get a 404, so they can't tell the project exists.
    Members without the right role get a 403.
    """
    membership = get_object_or_404(
        Membership.objects.select_related("project"), project_id=project_id, user=user
    )
    if roles and membership.role not in roles:
        raise HttpError(403, f"This needs the {' or '.join(roles)} role")
    return membership.project
  • One query finds the membership and the project together.
  • A non-member gets exactly the same 404 as for a project id that doesn’t exist. A 403 would confirm that project 1 exists.
  • A member who lacks the role gets a 403 that says which role is needed. They can already see the project, so there’s nothing to hide.

It’s a plain function, not a decorator: the view calls it and gets the project back, which it needs anyway.

Scoping projects

Now rewrite the projects endpoints so every query starts from the user’s memberships. Two schemas for managing members go at the end of projects/schemas.py, which now imports UserOut and Role:

# projects/schemas.py
from ninja import ModelSchema, Schema

from accounts.schemas import UserOut

from .models import Label, Project, Role

...

class MemberIn(Schema):
    username: str
    role: Role = Role.MEMBER


class MemberOut(Schema):
    user: UserOut
    role: Role
# projects/api.py
from django.contrib.auth.models import User
from django.db.models import Count, Q
from django.shortcuts import get_object_or_404
from ninja import Router, Status
from ninja.errors import HttpError

from .models import Project, Role
from .permissions import get_project_or_404
from .schemas import LabelIn, LabelOut, MemberIn, MemberOut, ProjectIn, ProjectOut

router = Router(tags=["projects"])


def projects_with_counts(user):
    return (
        Project.objects.filter(memberships__user=user)
        .annotate(
            task_count=Count("tasks"),
            unassigned_task_count=Count("tasks", filter=Q(tasks__assignee=None)),
        )
        .prefetch_related("labels")
    )


@router.get("/", response=list[ProjectOut])
def list_projects(request):
    return projects_with_counts(request.auth).order_by("id")


@router.post("/", response={201: ProjectOut})
def create_project(request, payload: ProjectIn):
    project = Project.objects.create(**payload.dict())
    project.memberships.create(user=request.auth, role=Role.OWNER)
    return Status(201, projects_with_counts(request.auth).get(id=project.id))


@router.get("/{project_id}", response=ProjectOut)
def get_project(request, project_id: int):
    return get_object_or_404(projects_with_counts(request.auth), id=project_id)


@router.put("/{project_id}", response=ProjectOut)
def update_project(request, project_id: int, payload: ProjectIn):
    project = get_project_or_404(request.auth, project_id, roles=[Role.OWNER, Role.ADMIN])
    for attr, value in payload.dict().items():
        setattr(project, attr, value)
    project.save()
    return projects_with_counts(request.auth).get(id=project.id)


@router.delete("/{project_id}", response={204: None})
def delete_project(request, project_id: int):
    get_project_or_404(request.auth, project_id, roles=[Role.OWNER]).delete()
    return Status(204, None)


@router.post("/{project_id}/labels", response={201: LabelOut})
def create_label(request, project_id: int, payload: LabelIn):
    project = get_project_or_404(request.auth, project_id, roles=[Role.OWNER, Role.ADMIN])
    return Status(201, project.labels.create(**payload.dict()))


@router.get("/{project_id}/members", response=list[MemberOut])
def list_members(request, project_id: int):
    project = get_project_or_404(request.auth, project_id)
    return project.memberships.select_related("user").order_by("id")


@router.post("/{project_id}/members", response={201: MemberOut})
def add_member(request, project_id: int, payload: MemberIn):
    project = get_project_or_404(request.auth, project_id, roles=[Role.OWNER])
    user = User.objects.filter(username=payload.username).first()
    if user is None:
        raise HttpError(400, f"User {payload.username} does not exist")
    if project.memberships.filter(user=user).exists():
        raise HttpError(409, f"{user.username} is already a member")
    return Status(201, project.memberships.create(user=user, role=payload.role))
  • projects_with_counts() takes the user and filters on their memberships, so list and get only ever see the user’s projects. The filter comes before annotate(), and a user has at most one membership per project, so the task counts stay correct.
  • The user who creates a project becomes its owner.
  • Update, delete and labels call get_project_or_404() with the roles they need.
  • MemberOut is a plain Schema, and list_members returns Membership objects. Django Ninja reads user and role from each one as attributes.

Try it

Sign up a third user with POST /taskflow/api/auth/signup. A weak password, as in {"username": "carol", "password": "carol"}, is rejected with the validators’ messages:

{
    "detail": "This password is too short. It must contain at least 8 characters. This password is too common."
}

{"username": "carol", "password": "sunflower-77"} returns 201 (with your next user id, not 3):

{
    "display_name": "carol",
    "id": 3,
    "username": "carol"
}

Log in as Carol. GET /taskflow/api/projects/ returns [], and GET /taskflow/api/projects/1 returns 404:

{
    "detail": "Not Found: No Project matches the given query."
}

The part after Not Found comes from Django’s Http404 message, and Django Ninja only adds it while DEBUG is on. With DEBUG = False, every 404 here is just {"detail": "Not Found"}, so it doesn’t matter which query didn’t find anything. Carol can create her own project, and she’s its owner.

As Alice, add Bob to project 1 with POST /taskflow/api/projects/1/members and {"username": "bob"}:

{
    "user": {
        "display_name": "bob",
        "id": 2,
        "username": "bob"
    },
    "role": "member"
}

Bob can now see the project and its tasks. He can’t rename it with PUT /taskflow/api/projects/1:

{
    "detail": "This needs the owner or admin role"
}

and DELETE /taskflow/api/projects/1 gives {"detail": "This needs the owner role"}.

Go deeper: Authentication, Errors & Exception Handling

Tasks and comments

Tasks belong to a project, so they’re covered by the same check. The new piece is comments, where the rule depends on the comment itself. Add the model to tasks/models.py:

# tasks/models.py
class Comment(models.Model):
    task = models.ForeignKey(Task, related_name="comments", on_delete=models.CASCADE)
    author = models.ForeignKey(
        settings.AUTH_USER_MODEL, related_name="comments", on_delete=models.CASCADE
    )
    body = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return f"Comment by {self.author} on {self.task}"
$ uv run python manage.py makemigrations
Migrations for 'tasks':
  tasks/migrations/0004_comment.py
    + Create model Comment
$ uv run python manage.py migrate

Add the comment schemas to tasks/schemas.py, and Comment to its model import:

# tasks/schemas.py
from ninja import ModelSchema, Schema
from pydantic import field_validator, model_validator
from pydantic_core import PydanticCustomError

from accounts.schemas import UserOut
from projects.schemas import LabelOut

from .models import Comment, Priority, Task, TaskStatus

...

class CommentIn(ModelSchema):
    class Meta:
        model = Comment
        fields = ["body"]


class CommentOut(ModelSchema):
    author: UserOut

    class Meta:
        model = Comment
        fields = ["id", "body", "created_at"]

Here’s the full tasks/api.py:

# tasks/api.py
from django.shortcuts import get_object_or_404
from ninja import PatchDict, Path, Router, Status
from ninja.errors import HttpError

from projects.models import Role
from projects.permissions import get_project_or_404

from .models import ALLOWED_TRANSITIONS, Priority, Task
from .schemas import CommentIn, CommentOut, StatusIn, TaskIn, TaskOut, TaskUpdate

router = Router(tags=["tasks"])


def get_task_or_404(user, project_id, task_id):
    project = get_project_or_404(user, project_id)
    return get_object_or_404(project.tasks, id=task_id)


def check_relations(project, payload):
    assignee_id = payload.assignee_id
    if assignee_id is not None and not project.memberships.filter(user_id=assignee_id).exists():
        raise HttpError(400, f"User {assignee_id} is not a member of this project")
    labels = list(project.labels.filter(id__in=payload.label_ids))
    if len(labels) != len(set(payload.label_ids)):
        raise HttpError(400, "Unknown label id for this project")
    return labels


@router.get("/", response=list[TaskOut])
def list_tasks(request, project_id: Path[int]):
    project = get_project_or_404(request.auth, project_id)
    return project.tasks.select_related("assignee").prefetch_related("labels").order_by("id")


@router.post("/", response={201: TaskOut})
def create_task(request, project_id: Path[int], payload: TaskIn):
    project = get_project_or_404(request.auth, project_id)
    labels = check_relations(project, payload)
    task = Task.objects.create(project=project, **payload.dict(exclude={"label_ids"}))
    task.labels.set(labels)
    return Status(201, task)


@router.get("/{task_id}", response=TaskOut)
def get_task(request, project_id: Path[int], task_id: int):
    return get_task_or_404(request.auth, project_id, task_id)


@router.patch("/{task_id}", response=TaskOut)
def update_task(request, project_id: Path[int], task_id: int, payload: PatchDict[TaskUpdate]):
    task = get_task_or_404(request.auth, project_id, task_id)
    for attr, value in payload.items():
        setattr(task, attr, value)
    if task.priority == Priority.URGENT and task.due_date is None:
        raise HttpError(400, "Urgent tasks need a due date")
    task.save()
    return task


@router.post("/{task_id}/status", response=TaskOut)
def change_status(request, project_id: Path[int], task_id: int, payload: StatusIn):
    task = get_task_or_404(request.auth, project_id, task_id)
    if payload.status not in ALLOWED_TRANSITIONS[task.status]:
        raise HttpError(409, f"Can't move a task from {task.status} to {payload.status}")
    task.status = payload.status
    task.save()
    return task


@router.get("/{task_id}/comments", response=list[CommentOut])
def list_comments(request, project_id: Path[int], task_id: int):
    task = get_task_or_404(request.auth, project_id, task_id)
    return task.comments.select_related("author").order_by("id")


@router.post("/{task_id}/comments", response={201: CommentOut})
def add_comment(request, project_id: Path[int], task_id: int, payload: CommentIn):
    task = get_task_or_404(request.auth, project_id, task_id)
    return Status(201, task.comments.create(author=request.auth, body=payload.body))


@router.put("/{task_id}/comments/{comment_id}", response=CommentOut)
def update_comment(
    request, project_id: Path[int], task_id: int, comment_id: int, payload: CommentIn
):
    task = get_task_or_404(request.auth, project_id, task_id)
    comment = get_object_or_404(task.comments, id=comment_id)
    if comment.author != request.auth:
        raise HttpError(403, "Only the author can edit a comment")
    comment.body = payload.body
    comment.save()
    return comment


@router.delete("/{task_id}/comments/{comment_id}", response={204: None})
def delete_comment(request, project_id: Path[int], task_id: int, comment_id: int):
    task = get_task_or_404(request.auth, project_id, task_id)
    comment = get_object_or_404(task.comments, id=comment_id)
    if comment.author != request.auth:
        # Anyone else needs to own the project
        get_project_or_404(request.auth, project_id, roles=[Role.OWNER])
    comment.delete()
    return Status(204, None)
  • get_task_or_404() goes through the project check first, then looks the task up in project.tasks. Every task and comment endpoint uses it, so none of them can reach another project’s data.
  • check_relations() now only accepts assignees who are members of the project. Part 2’s check for any existing user would let you assign tasks to strangers.
  • Comments are looked up through task.comments, so a comment id from another task is a 404.
  • The comment author comes from request.auth, never from the request body.
  • These are object-level rules: they depend on the comment, not only on the project role. Only the author may edit a comment. The author or the project owner may delete it, and for anyone else the view reuses get_project_or_404() with roles=[Role.OWNER].

Try it

As Bob, post a comment on task 2 with POST /taskflow/api/projects/1/tasks/2/comments and {"body": "The footer links point to the old domain."}:

{
    "author": {
        "display_name": "bob",
        "id": 2,
        "username": "bob"
    },
    "id": 1,
    "body": "The footer links point to the old domain.",
    "created_at": "2026-09-28T11:00:54.225Z"
}

Alice replies with a comment of her own, {"body": "Good catch, thanks!"} (id 2). She owns the project, but that doesn’t let her edit Bob’s words. PUT /taskflow/api/projects/1/tasks/2/comments/1 as Alice returns:

{
    "detail": "Only the author can edit a comment"
}

The same request as Bob returns 200 with the new body. Bob can’t delete Alice’s comment 2:

{
    "detail": "This needs the owner role"
}

But Alice, as the owner, can delete Bob’s comment 1: DELETE /taskflow/api/projects/1/tasks/2/comments/1 returns 204. And Carol, who isn’t a member, gets a 404 for the task’s comments, like for everything else in the project.

Go deeper: Authentication, Decorators

Recap

  • A JWT is a signed, readable payload. HS256 with hmac, hashlib, base64 and json fits in a module of about 50 lines. Verify the signature before reading the claims, and check the expiry and the token type.
  • HttpBearer reads the Authorization: Bearer header. What authenticate() returns becomes request.auth, and None means 401.
  • NinjaAPI(auth=...) protects every endpoint. auth=None on a router or an operation opts it out, as for login and refresh.
  • One helper, get_project_or_404(user, project_id, roles=...), answers 404 to non-members and 403 to members without the right role.
  • Start every query from the user’s memberships, and take ownership fields such as a comment’s author from request.auth.
  • Rules about a single object, such as who may edit a comment, go in the view, after the object is loaded.

Next

The task list returns everything at once. In Part 5: Filtering, search, ordering & pagination, you’ll filter tasks by status, assignee and label, search them, sort them, and paginate tasks and comments.

3.5 - Part 5: Filtering, search, ordering & pagination

A FilterSchema for tasks, safe sort orders, and page-number and cursor pagination.

GET /taskflow/api/projects/{project_id}/tasks/ returns every task of a project in one response. That’s fine for a handful, but a real team wants the tasks assigned to them, the overdue ones, or the ones mentioning “checkout”, and it doesn’t want a thousand of them at once. In this part you’ll add filters and search with a FilterSchema, let clients pick the sort order from a fixed list, and paginate tasks by page number and comments with a cursor.

Some tasks to work with

Filters are more interesting with more than four tasks. Add three to the “Website redesign” project, plus a few comments on task 2. Django’s shell imports your models automatically:

$ uv run python manage.py shell
>>> bob = User.objects.get(username="bob")
>>> fix = Task.objects.create(
...     project_id=1, title="Fix mobile menu overlap", assignee=bob, priority=4,
...     due_date="2026-10-01", description="The menu covers the logo on small screens",
... )
>>> fix.labels.add(1)
>>> Task.objects.create(project_id=1, title="Set up analytics", description="Page views and the signup funnel", due_date="2026-10-20")
<Task: Set up analytics>
>>> Task.objects.create(project_id=1, title="Compress hero images", assignee=bob, priority=3, due_date="2026-10-03")
<Task: Compress hero images>
>>> for body in ["Found three more on the pricing page.", "Fixed on staging.", "Looks good, closing."]:
...     Comment.objects.create(task_id=2, author=bob, body=body)
...
<Comment: Comment by bob on Fix broken footer links>
<Comment: Comment by bob on Fix broken footer links>
<Comment: Comment by bob on Fix broken footer links>

The original tutorial writes assignee_id=2 and author_id=2, which is bob’s id there. Looking bob up by his username works whatever his id is in your database.

The project now has seven tasks: 1 to 4 from the earlier parts and the new 5, 6 and 7. Task 2 has four comments, with ids 2 to 5. Nothing in this part changes a model, so there are no migrations.

A filter schema

A FilterSchema is a Schema whose fields are query parameters, with a .filter() method that turns them into a Q expression. Add one to tasks/schemas.py, together with the list of allowed sort orders:

# tasks/schemas.py
from datetime import date
from typing import Annotated, Literal

from ninja import FilterLookup, FilterSchema, ModelSchema, Schema
from pydantic import field_validator, model_validator
from pydantic_core import PydanticCustomError

from accounts.schemas import UserOut
from projects.schemas import LabelOut

from .models import Comment, Priority, Task, TaskStatus

...


class TaskFilter(FilterSchema):
    status: Annotated[list[TaskStatus] | None, FilterLookup("status__in")] = None
    assignee: Annotated[str | None, FilterLookup("assignee__username")] = None
    label: Annotated[str | None, FilterLookup("labels__name")] = None
    due_before: Annotated[date | None, FilterLookup("due_date__lt")] = None
    search: Annotated[
        str | None, FilterLookup(["title__icontains", "description__icontains"])
    ] = None


TaskOrdering = Literal[
    "created_at", "-created_at", "due_date", "-due_date", "priority", "-priority"
]
  • Every field defaults to None, and a None field is skipped. A request without query parameters is not filtered at all.
  • FilterLookup sets the ORM lookup for a field. Without it, the field name is used, as in Q(status=value). The public parameter names stay short (assignee, label), while the lookups follow relations (assignee__username, labels__name).
  • status is a list, so a client can repeat the parameter: ?status=todo&status=in_progress becomes Q(status__in=[...]). The items are TaskStatus values, so an unknown status is a 422.
  • search has a list of lookups. They’re combined with OR, so the term can appear in the title or the description. The different fields are combined with AND.
  • due_before is a date, parsed and validated like any other schema field.
  • TaskOrdering is used in the view below.

Go deeper: Filtering: custom lookups, Filtering: combining expressions

Filters, ordering and pages in the view

Here’s the new list_tasks, with the imports it needs:

# tasks/api.py
from django.shortcuts import get_object_or_404
from ninja import PatchDict, Path, Query, Router, Status
from ninja.errors import HttpError
from ninja.pagination import CursorPagination, PageNumberPagination, paginate

from projects.models import Role
from projects.permissions import get_project_or_404

from .models import ALLOWED_TRANSITIONS, Priority, Task
from .schemas import (
    CommentIn,
    CommentOut,
    StatusIn,
    TaskFilter,
    TaskIn,
    TaskOrdering,
    TaskOut,
    TaskUpdate,
)

...


@router.get("/", response=list[TaskOut])
@paginate(PageNumberPagination, page_size=20)
def list_tasks(
    request,
    project_id: Path[int],
    filters: Query[TaskFilter],
    ordering: TaskOrdering = "created_at",
):
    project = get_project_or_404(request.auth, project_id)
    tasks = filters.filter(project.tasks.select_related("assignee").prefetch_related("labels"))
    return tasks.order_by(ordering, "id")
  • Query[TaskFilter] reads the schema’s fields from the query string, so the parameters are ?status=...&search=..., not ?filters=....
  • filters.filter() applies the filters to the queryset it’s given. That queryset is already scoped to a project the user belongs to, so the filters can only narrow Part 4’s permissions, never widen them.
  • ordering is a plain query parameter typed with the Literal. Only the six listed values get through, which matters because the value goes straight into order_by(). Never pass a client’s string to order_by() unchecked. It would let them sort by any field, including ones they can’t see, such as a related user’s password hash.
  • "id" comes second in order_by(), so tasks with the same priority or due date always come back in the same order. Stable ordering matters once results are split into pages.
  • @paginate goes below the router decorator. The view still returns a queryset, and response still says list[TaskOut]. The next section covers what it changes.

Try it

The shell is quicker than curl for a series of queries. TestClient from Part 2 can send the token as a default header. With a small helper that prints only the titles:

$ uv run python manage.py shell
>>> from accounts.tokens import create_token_pair
>>> from ninja.testing import TestClient
>>> from taskflow.api import api
>>> token = create_token_pair(User.objects.get(username="alice").id)["access"]
>>> client = TestClient(api, headers={"Authorization": f"Bearer {token}"})
>>> def titles(query):
...     return [task["title"] for task in client.get(f"/projects/1/tasks/?{query}").json()["items"]]
...
>>> titles("assignee=bob")
['Fix broken footer links', 'Fix mobile menu overlap', 'Compress hero images']
>>> titles("status=in_progress&status=done")
['Write launch announcement']
>>> titles("label=bug")
['Fix broken footer links', 'Fix mobile menu overlap']
>>> titles("search=fix")
['Fix broken footer links', 'Fix mobile menu overlap']
>>> titles("search=SIGNUP")
['Set up analytics']
>>> titles("due_before=2026-10-10&ordering=due_date")
['Fix mobile menu overlap', 'Compress hero images']
>>> titles("status=todo&ordering=-priority")
['Fix mobile menu overlap', 'Compress hero images', 'Draft the sitemap', 'Fix broken footer links', 'Pick a color palette', 'Set up analytics']
  • search=SIGNUP finds “Set up analytics” through its description, because icontains ignores case.
  • Filtering on label=bug only selects the tasks. Their labels in the response still list every label, because prefetch_related loads them in a separate query.
  • With -priority, the four tasks with priority 2 come out in id order, thanks to the second order_by() field.
  • On SQLite, tasks without a due date come first with ordering=due_date. PostgreSQL puts them last. Combine the ordering with due_before, as above, to leave them out.

Values that don’t fit the types are rejected before the view runs. ?ordering=oldest returns a 422:

{
    "detail": [
        {
            "type": "literal_error",
            "loc": [
                "query",
                "ordering"
            ],
            "msg": "Input should be 'created_at', '-created_at', 'due_date', '-due_date', 'priority' or '-priority'",
            "ctx": {
                "expected": "'created_at', '-created_at', 'due_date', '-due_date', 'priority' or '-priority'"
            }
        }
    ]
}

And ?status=archived fails on the first list item. The loc includes filters, the name of the view parameter that holds the schema:

{
    "detail": [
        {
            "type": "enum",
            "loc": [
                "query",
                "filters",
                "status",
                0
            ],
            "msg": "Input should be 'todo', 'in_progress' or 'done'",
            "ctx": {
                "expected": "'todo', 'in_progress' or 'done'"
            }
        }
    ]
}

In the interactive docs, each filter shows up as its own query parameter, and ordering as a dropdown with the six values.

Go deeper: Filtering, Query Parameters

Page numbers

@paginate(PageNumberPagination, page_size=20) adds two query parameters, page (starting at 1) and page_size, and wraps the result in an object with the page’s items and the total count. The shell helper above already read ["items"]. With TOKEN set as in Part 4 (access tokens expire after 15 minutes, so get a fresh one from POST /taskflow/api/auth/token if you get a 401), ask for the third page of three tasks, sorted by priority:

$ curl "http://127.0.0.1:8000/taskflow/api/projects/1/tasks/?ordering=-priority&page=3&page_size=3" -H "Authorization: Bearer $TOKEN"
{
    "items": [
        {
            "assignee": null,
            "labels": [],
            "status": "todo",
            "priority": 2,
            "id": 6,
            "project": 1,
            "title": "Set up analytics",
            "description": "Page views and the signup funnel",
            "due_date": "2026-10-20",
            "created_at": "2026-09-28T11:08:55.867Z"
        }
    ],
    "count": 7
}
  • The paginator slices the queryset, so the database only returns the rows of the requested page. count is one extra COUNT query with the same filters. The filtered, paginated list takes 5 queries in total: the user from the token, the membership, the count, the tasks with their assignees, and the labels.
  • Without page_size, a page has 20 tasks. A client can ask for more, up to max_page_size, which defaults to 100. ?page_size=500 returns at most 100 tasks.
  • A page past the end, such as ?page=9, is not an error. It returns {"items": [], "count": 7}. ?page=0 is a 422, because page must be at least 1.
  • In the OpenAPI schema, the response is now PagedTaskOut, with items and count. Django Ninja generates it from list[TaskOut].

Page numbers are what most UIs need: “page 3 of 5”, with a total. Their weak point is data that changes while someone is paging. If a task is added at the top between two requests, every page shifts by one, and the client sees one task twice.

Go deeper: Pagination

Cursor pagination for comments

Comments grow at the end, and clients load them as a feed, “show more” by “show more”. That’s what CursorPagination is for. Instead of a page number, each response has a next link that points just past the last item seen. In tasks/api.py, add the decorator to list_comments and drop the .order_by("id"):

# tasks/api.py
@router.get("/{task_id}/comments", response=list[CommentOut])
@paginate(CursorPagination, ordering=("id",), page_size=20)
def list_comments(request, project_id: Path[int], task_id: int):
    task = get_task_or_404(request.auth, project_id, task_id)
    return task.comments.select_related("author")
  • ordering belongs to the paginator now. It sorts the queryset itself, and uses the first field to record the position in the cursor. That field should be unique, or close to it. The comment id is unique and grows with time, so the oldest comments come first.
  • The parameters are cursor and page_size. There is no page number and no count.

Ask for two comments at a time:

$ curl "http://127.0.0.1:8000/taskflow/api/projects/1/tasks/2/comments?page_size=2" -H "Authorization: Bearer $TOKEN"
{
    "previous": null,
    "next": "http://127.0.0.1:8000/taskflow/api/projects/1/tasks/2/comments?cursor=cD00&page_size=2",
    "results": [
        {
            "author": {
                "display_name": "Alice Martin",
                "id": 1,
                "username": "alice"
            },
            "id": 2,
            "body": "Good catch, thanks!",
            "created_at": "2026-09-28T11:08:55.863Z"
        },
        {
            "author": {
                "display_name": "bob",
                "id": 2,
                "username": "bob"
            },
            "id": 3,
            "body": "Found three more on the pricing page.",
            "created_at": "2026-09-28T11:08:55.867Z"
        }
    ]
}

The items are under results, not items. The next link keeps the other query parameters and adds a cursor. cD00 is base64 for p=4: the next page starts at comment id 4. Follow the link:

{
    "previous": "http://127.0.0.1:8000/taskflow/api/projects/1/tasks/2/comments?cursor=cD00JnI9VHJ1ZSZvPTE%3D&page_size=2",
    "next": null,
    "results": [
        {
            "author": {
                "display_name": "bob",
                "id": 2,
                "username": "bob"
            },
            "id": 4,
            "body": "Fixed on staging.",
            "created_at": "2026-09-28T11:08:55.869Z"
        },
        {
            "author": {
                "display_name": "bob",
                "id": 2,
                "username": "bob"
            },
            "id": 5,
            "body": "Looks good, closing.",
            "created_at": "2026-09-28T11:08:55.870Z"
        }
    ]
}
  • next is null on the last page, and previous leads back to comments 2 and 3.
  • The cursor is a position, not an offset. If someone adds a comment and the client follows the same next link again, it still gets comments 4 and 5, now with a next link to the new comment. Nothing is skipped or shown twice.
  • The paginator fetches one row more than the page size to know whether there’s a next page, so it needs no COUNT query. That keeps it fast on large tables, but a UI can’t show “page 3 of 5”.
  • The cursor is only encoded, not signed. A client can craft one, but it can only move the position inside the queryset the view returns, which is already limited to the task’s comments. A cursor that isn’t valid base64 returns a 422 with {"detail": [{"cursor": "Invalid Cursor"}]}.

Go deeper: Pagination

Recap

  • A FilterSchema declares filters as typed fields, and filters.filter(queryset) turns the fields that were sent into a Q expression. Use it with Query[...].
  • FilterLookup maps a short parameter name to any ORM lookup. A list of lookups is combined with OR, which is all a simple search needs.
  • Accept a sort order only from a Literal of allowed values, and add a unique field such as id as a tie-breaker.
  • @paginate(PageNumberPagination) wraps a view that returns a queryset. It slices the query and adds count. Clients get page numbers and totals.
  • @paginate(CursorPagination, ordering=...) gives next/previous links that stay correct while data is added, without a COUNT query. Use it for feeds such as comments.

Next

Tasks can have comments, but not files. In Part 6: File uploads & forms, you’ll attach files to tasks with multipart uploads, validate their size and type, and return absolute URLs for them.

3.6 - Part 6: File uploads & forms

Multipart uploads with form fields, file validation, media files and absolute URLs.

Tasks often come with files: a design mockup, a screenshot of a bug, a PDF from the client. In this part you’ll let members attach files to tasks. An upload is a multipart/form-data request instead of JSON, so you’ll combine a file parameter with form fields, validate the file’s size and type, store it with a plain FileField, and return an absolute URL for it.

Storing files

An attachment belongs to a task and remembers who uploaded it. Add the model at the end of tasks/models.py:

# tasks/models.py
class Attachment(models.Model):
    task = models.ForeignKey(Task, related_name="attachments", on_delete=models.CASCADE)
    file = models.FileField(upload_to="attachments/%Y/%m/")
    name = models.CharField(max_length=200)
    uploaded_by = models.ForeignKey(
        settings.AUTH_USER_MODEL, related_name="attachments", on_delete=models.CASCADE
    )
    uploaded_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.name
  • file is a FileField, not an ImageField. ImageField needs Pillow, and attachments aren’t only images.
  • upload_to puts files in a folder per month, such as attachments/2026/09/. The database stores only that relative path. The file itself goes to the storage, which is the local disk by default.
  • name is the display name. The client can choose it, or it defaults to the uploaded file’s name.
$ uv run python manage.py makemigrations
Migrations for 'tasks':
  tasks/migrations/0005_attachment.py
    + Create model Attachment
$ uv run python manage.py migrate

Django needs to know where to put uploads and under which URL to serve them. Add two settings below STATIC_URL in locallibrary_config/settings.py:

# locallibrary_config/settings.py
STATIC_URL = 'static/'

# User-uploaded files
MEDIA_URL = 'media/'
MEDIA_ROOT = BASE_DIR / 'media'

runserver doesn’t serve MEDIA_ROOT on its own. Your locallibrary_config/urls.py already uses Django’s static() helper to serve static files, in the block you added in the MDN tutorial. Add one line to that block, for media files:

# locallibrary_config/urls.py
# Use static() to add URL mapping to serve static files during development (only)
from django.conf import settings
from django.conf.urls.static import static

urlpatterns += static(settings.STATIC_URL, document_root=settings.STATIC_ROOT)
urlpatterns += static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)

static() returns no URL patterns when DEBUG is off. In production, the web server or a storage service such as S3 serves the files.

Uploaded files never belong in git. The original tutorial says its .gitignore already excludes media/, but the one you started from in Prep 01 doesn’t. Add this line to your .gitignore:

media/

Validating the upload

Django Ninja passes an uploaded file to your view as an UploadedFile, which is Django’s own class. It already has a name and a size. Validation rules for it can live in the type, like any other Pydantic rule. Add the new imports and code to tasks/schemas.py:

# tasks/schemas.py
from datetime import date
from pathlib import Path
from typing import Annotated, Literal

from ninja import FilterLookup, FilterSchema, ModelSchema, Schema, UploadedFile
from pydantic import AfterValidator, Field, field_validator, model_validator
from pydantic_core import PydanticCustomError

from accounts.schemas import UserOut
from projects.schemas import LabelOut

from .models import Attachment, Comment, Priority, Task, TaskStatus

...


MAX_UPLOAD_SIZE = 5 * 1024 * 1024  # 5 MB
ALLOWED_EXTENSIONS = {".pdf", ".png", ".jpg", ".jpeg", ".txt", ".md"}


def check_upload(file: UploadedFile) -> UploadedFile:
    if file.size > MAX_UPLOAD_SIZE:
        raise PydanticCustomError("file_too_large", "Files can be at most 5 MB")
    if Path(file.name).suffix.lower() not in ALLOWED_EXTENSIONS:
        raise PydanticCustomError("file_type", "Only PDF, PNG, JPEG and text files are allowed")
    return file


AttachmentFile = Annotated[UploadedFile, AfterValidator(check_upload)]


class AttachmentIn(Schema):
    name: str = Field("", max_length=200)
  • check_upload is a normal Pydantic validator. It runs after the file is parsed, and raises PydanticCustomError like the validators from Part 3, so a rejected file is a 422 with its own type and message.
  • AttachmentFile bundles the type and its rules. Any endpoint that takes an attachment can use it, and the view only ever sees a valid file.
  • The extension check uses the file name. The content_type of an upload comes from the client too, so neither proves what’s inside the file. What they do prevent is storing, say, an .html file that your media server would then serve as a web page.
  • Django has no size limit for uploaded files. DATA_UPLOAD_MAX_MEMORY_SIZE doesn’t count them, and FILE_UPLOAD_MAX_MEMORY_SIZE only decides when Django switches from memory to a temporary file. The check runs after the whole file has arrived, so in production also cap the request size in the web server, such as client_max_body_size in nginx.
  • AttachmentIn holds the form fields that come with the file. There is only one here, the optional display name.

Go deeper: File Uploads, Errors & Exception Handling

The upload endpoint

A file can’t travel inside a JSON body, so the whole request is multipart/form-data: one part per form field, and one for the file. Add the imports and two endpoints to tasks/api.py:

# tasks/api.py
from django.shortcuts import get_object_or_404
from ninja import File, Form, PatchDict, Path, Query, Router, Status
from ninja.errors import HttpError
from ninja.pagination import CursorPagination, PageNumberPagination, paginate

from projects.models import Role
from projects.permissions import get_project_or_404

from .models import ALLOWED_TRANSITIONS, Priority, Task
from .schemas import (
    AttachmentFile,
    AttachmentIn,
    AttachmentOut,
    CommentIn,
    CommentOut,
    StatusIn,
    TaskFilter,
    TaskIn,
    TaskOrdering,
    TaskOut,
    TaskUpdate,
)

...


@router.get("/{task_id}/attachments", response=list[AttachmentOut])
def list_attachments(request, project_id: Path[int], task_id: int):
    task = get_task_or_404(request.auth, project_id, task_id)
    return task.attachments.select_related("uploaded_by").order_by("id")


@router.post("/{task_id}/attachments", response={201: AttachmentOut})
def upload_attachment(
    request,
    project_id: Path[int],
    task_id: int,
    details: Form[AttachmentIn],
    file: File[AttachmentFile],
):
    task = get_task_or_404(request.auth, project_id, task_id)
    attachment = task.attachments.create(
        file=file, name=details.name or file.name, uploaded_by=request.auth
    )
    return Status(201, attachment)
  • Form[AttachmentIn] reads the schema’s fields from the form data, the same way Query[TaskFilter] read them from the query string in Part 5. The client sends a name field, not a details field.
  • File[AttachmentFile] reads the file part from request.FILES and runs check_upload. With a File parameter present, Django Ninja documents the whole request body as multipart/form-data, and the interactive docs show a file picker.
  • Passing the UploadedFile to create() is enough. The FileField saves it to the storage under upload_to, and the row gets the stored path.
  • The file name comes from the client, but Django keeps only its last part. A file sent as ../../evil.pdf is stored as attachments/2026/09/evil.pdf. If a file with the same name exists, the storage adds a random suffix, as in footer_42rBBEz.png, and the display name stays footer.png.
  • Both endpoints start with get_task_or_404 from Part 4, so only members of the project can list or upload files.

Go deeper: File Uploads, Form Data

Absolute file URLs

A FileField listed in Meta.fields is serialized as its URL, such as /media/attachments/2026/09/sitemap.pdf. That’s a path, not a URL. A mobile app or a frontend on another domain doesn’t know which host to put in front of it. The request does, so AttachmentOut builds the full URL in a resolver:

# tasks/schemas.py
class AttachmentOut(ModelSchema):
    uploaded_by: UserOut
    url: str

    class Meta:
        model = Attachment
        fields = ["id", "name", "uploaded_at"]

    @staticmethod
    def resolve_url(obj, context):
        return context["request"].build_absolute_uri(obj.file.url)
  • A resolver that takes a context argument receives {"request": ..., "response_status": ...} when the schema is used as a response. build_absolute_uri() adds the scheme and host of the current request.
  • url is a new field that doesn’t exist on the model. The model’s file field isn’t in Meta.fields, so the response has no relative path next to the absolute one.
  • Declared fields come first in the output, so the keys are uploaded_by, url, then id, name and uploaded_at.

Go deeper: Schemas: accessing context, Schemas: files and images

Try it

Start runserver and set TOKEN to an access token for alice, as in Part 4. With any small PDF saved as sitemap.pdf, upload it to task 1. curl -F sends a multipart/form-data request, and @ reads the file from disk:

$ curl http://127.0.0.1:8000/taskflow/api/projects/1/tasks/1/attachments \
    -H "Authorization: Bearer $TOKEN" \
    -F "name=Sitemap draft" \
    -F "file=@sitemap.pdf"
{
    "uploaded_by": {
        "display_name": "Alice Martin",
        "id": 1,
        "username": "alice"
    },
    "url": "http://127.0.0.1:8000/media/attachments/2026/09/sitemap.pdf",
    "id": 1,
    "name": "Sitemap draft",
    "uploaded_at": "2026-09-28T11:12:57.878Z"
}

Don’t set a Content-Type header yourself. curl -F sets it to multipart/form-data together with the boundary string that separates the parts. The URL works in the browser, and curl -I shows that Django serves it with the right type. Use the url from your own upload response, because the folder is named after the month you upload in:

$ curl -I http://127.0.0.1:8000/media/attachments/2026/09/sitemap.pdf
HTTP/1.1 200 OK
...
Content-Type: application/pdf
Content-Length: 2025
Content-Disposition: inline; filename="sitemap.pdf"
...

The media URL needs no token. Anyone who has the link can download the file. That’s fine for this tutorial, but for private files you’d return the file from an endpoint that checks permissions first, or use signed URLs from your storage service.

Now try a file type that isn’t allowed, such as -F "file=@setup.exe":

{
    "detail": [
        {
            "type": "file_type",
            "loc": [
                "file",
                "file"
            ],
            "msg": "Only PDF, PNG, JPEG and text files are allowed"
        }
    ]
}

A 6 MB scan.pdf gets "type": "file_too_large" with the message "Files can be at most 5 MB". Errors in the form fields and the file are collected together. A 201-character name with no file returns both:

{
    "detail": [
        {
            "type": "string_too_long",
            "loc": [
                "form",
                "name"
            ],
            "msg": "String should have at most 200 characters",
            "ctx": {
                "max_length": 200
            }
        },
        {
            "type": "missing",
            "loc": [
                "file",
                "file"
            ],
            "msg": "Field required"
        }
    ]
}

The first item of loc is where the value was read from: form for form fields, file for files. A JSON body sent to this endpoint gets the same missing error for file.

Deleting attachments

Deleting follows the rule from Part 4’s comments, with one difference: the uploader can delete their own file, and so can the project’s owners and admins. Add the endpoint to tasks/api.py:

# tasks/api.py
@router.delete("/{task_id}/attachments/{attachment_id}", response={204: None})
def delete_attachment(request, project_id: Path[int], task_id: int, attachment_id: int):
    task = get_task_or_404(request.auth, project_id, task_id)
    attachment = get_object_or_404(task.attachments, id=attachment_id)
    if attachment.uploaded_by != request.auth:
        get_project_or_404(request.auth, project_id, roles=[Role.OWNER, Role.ADMIN])
    attachment.file.delete(save=False)
    attachment.delete()
    return Status(204, None)
  • get_object_or_404(task.attachments, ...) looks the attachment up within the task. An attachment id from another task is a 404.
  • For anyone but the uploader, get_project_or_404 with roles does the role check, and raises a 403 for a plain member.
  • Deleting the row doesn’t delete the file. Django never removes files from the storage on its own, not even on a cascade. attachment.file.delete(save=False) removes it first. save=False skips saving the row, which is deleted on the next line anyway.

Bob is a member of the project, not an owner. Get a token for him (password builder-42) and put it in BOB_TOKEN. When he tries to delete alice’s sitemap:

$ curl -X DELETE http://127.0.0.1:8000/taskflow/api/projects/1/tasks/1/attachments/1 -H "Authorization: Bearer $BOB_TOKEN"
{
    "detail": "This needs the owner or admin role"
}

His own uploads, or alice deleting any attachment in her project, return 204 and remove the file from media/. Files deleted with the whole task, by the cascade, stay on disk. Cleaning those up takes a periodic job or a post_delete signal.

Go deeper: Authentication

Recap

  • A File[...] parameter makes the request multipart/form-data. Combine it with Form[Schema] to read the other fields, which are validated like a JSON body.
  • Annotated[UploadedFile, AfterValidator(...)] puts size and type rules into a reusable type. A bad file is a 422 with loc ["file", ...], next to any form errors.
  • Store uploads with a FileField and MEDIA_ROOT, and serve them with static() in development only.
  • A resolver with a context argument can read the request, and build_absolute_uri() turns a file’s path into a full URL.
  • Delete the stored file together with the row. Django won’t do it for you.

Next

TaskFlow now has most of its features, and each part checked them by hand. In Part 7: Testing, you’ll turn those checks into a pytest suite with Django Ninja’s TestClient, covering CRUD, validation, permissions, filters and uploads.

3.7 - Part 7: Testing

A pytest suite with fixtures and TestClient, covering validation, permissions, filters and uploads.

So far you’ve checked every feature by hand, with curl and the shell. In this part you’ll turn those checks into a pytest suite that runs in about half a second. Django Ninja’s TestClient calls your operations directly, so tests send JSON, form data, files and tokens without a running server.

Setting up pytest

The tests need two packages. They’re only used in development, so they don’t join Django and Django Ninja as runtime dependencies:

$ uv add --dev pytest pytest-django

Like Selenium in Part 4 of the Channels tutorial, they go into the development dependency group in pyproject.toml. Commit pyproject.toml and uv.lock.

pytest-django creates a test database for every run, applies your migrations to it, and wraps each test in a transaction that is rolled back afterwards. It needs to know your settings module. Create pytest.ini next to manage.py:

# pytest.ini
[pytest]
DJANGO_SETTINGS_MODULE = locallibrary_config.settings
testpaths = tests

Part 1 deleted the tests.py files from the apps. The tests go into a top-level tests/ folder instead, because most of them cross app boundaries: a task test needs users from accounts and a project from projects. testpaths tells pytest to look there. The folder needs no __init__.py. pytest-django puts the folder with manage.py on the import path, so from taskflow.api import api works in tests.

testpaths also keeps pytest away from catalog and chat. Their tests are written for Django’s own test runner, and chat/tests.py opens Chrome. Keep running those with uv run python manage.py test, which in turn doesn’t look in tests/, because that folder isn’t a Python package.

In the original tutorial, the SQLite test database lives in memory. Yours doesn’t: in Part 4 of the Channels tutorial you gave it its own file, test_db.sqlite3, in the TEST entry of DATABASES. pytest-django uses that setting too. Either way, your db.sqlite3 and the data from earlier parts are never touched.

Go deeper: Testing

Fixtures

Almost every test needs the same things: a few users, a project with an owner and a member, and a client that sends a token. pytest fixtures build them on demand. Fixtures in tests/conftest.py are available to every test file without an import:

# tests/conftest.py
import pytest
from django.contrib.auth.models import User
from ninja.testing import TestClient

from accounts.tokens import create_token_pair
from projects.models import Project, Role
from taskflow.api import api


@pytest.fixture(autouse=True)
def test_settings(settings, tmp_path):
    settings.MEDIA_ROOT = tmp_path  # uploads go to a temporary folder
    settings.PASSWORD_HASHERS = ["django.contrib.auth.hashers.MD5PasswordHasher"]


@pytest.fixture
def alice(db):
    return User.objects.create_user("alice", password="wonderland-42")


@pytest.fixture
def bob(db):
    return User.objects.create_user("bob", password="builder-42")


@pytest.fixture
def carol(db):
    return User.objects.create_user("carol", password="sunflower-77")


@pytest.fixture
def project(alice, bob):
    """Alice owns the project, bob is a member."""
    project = Project.objects.create(name="Website redesign", description="New marketing site")
    project.memberships.create(user=alice, role=Role.OWNER)
    project.memberships.create(user=bob, role=Role.MEMBER)
    return project


@pytest.fixture
def client_for():
    """Return a TestClient that is logged in as the given user."""

    def make_client(user):
        token = create_token_pair(user.id)["access"]
        return TestClient(api, headers={"Authorization": f"Bearer {token}"})

    return make_client
  • test_settings runs for every test because of autouse=True. It uses pytest-django’s settings fixture, which restores the original values after each test.
  • MEDIA_ROOT points to tmp_path, a fresh temporary folder per test, so upload tests don’t write into your media/ folder.
  • Django’s default password hasher is slow on purpose. Every create_user() and every login pays for it. The MD5 hasher is unsafe for real passwords, but fine for test users. On the machine used for this tutorial, the suite takes 51 seconds without that line and half a second with it.
  • The user fixtures take pytest-django’s db fixture, which allows database access. project depends on alice and bob, so asking for project creates all three.
  • client_for is a factory fixture. A test can call it for any user, even several times, as in client_for(alice) and client_for(bob). It creates the token with create_token_pair() from Part 4 instead of calling /auth/token, which keeps the tests of other endpoints independent of the login endpoint.
  • TestClient(api, headers=...) sends those headers with every request. You used the same client in the shell in Part 5.

TestClient doesn’t go through urls.py or Django’s middleware. It finds the operation in the NinjaAPI and calls it with a fake request. Paths are therefore relative to the API: /projects/, not /taskflow/api/projects/. Everything that is part of the operation still runs: authentication, validation, your view, the response schema and the error handlers. A path that matches no operation raises an exception instead of returning 404, so a typo in a test fails loudly.

Go deeper: Testing

Testing authentication

The auth tests use a plain client without a token, because getting one is what they test. Create tests/test_auth.py:

# tests/test_auth.py
from ninja.testing import TestClient

from accounts.tokens import create_token
from taskflow.api import api

client = TestClient(api)


def test_obtain_token(alice):
    response = client.post("/auth/token", json={"username": "alice", "password": "wonderland-42"})
    assert response.status_code == 200
    token = response.json()["access"]

    response = client.get("/projects/", headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 200


def test_wrong_password(alice):
    response = client.post("/auth/token", json={"username": "alice", "password": "nope"})
    assert response.status_code == 401
    assert response.json() == {"detail": "Invalid username or password"}


def test_missing_token():
    response = client.get("/projects/")
    assert response.status_code == 401


def test_expired_token(alice):
    token = create_token(alice.id, "access", lifetime=-1)
    response = client.get("/projects/", headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 401


def test_refresh_token_is_not_an_access_token(alice):
    response = client.post("/auth/token", json={"username": "alice", "password": "wonderland-42"})
    refresh = response.json()["refresh"]
    response = client.get("/projects/", headers={"Authorization": f"Bearer {refresh}"})
    assert response.status_code == 401
  • json= serializes the body, like a client sending Content-Type: application/json. response.json() decodes the answer.
  • headers= on a single request adds to the client’s default headers, or overrides them.
  • Waiting 15 minutes for a token to expire isn’t an option. create_token() takes the lifetime as an argument, so a negative one creates a token that expired a second ago.
  • test_missing_token needs no database. HttpBearer rejects a request without an Authorization header before authenticate() runs.

Go deeper: Testing

Projects and permissions

Most of the value of an API test suite is in the permission rules, because a missing check doesn’t show up when you try the happy path. Create tests/test_projects.py:

# tests/test_projects.py
from projects.models import Role


def test_create_project(alice, client_for):
    response = client_for(alice).post(
        "/projects/", json={"name": "Launch", "description": "Product launch"}
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Launch"
    assert data["task_count"] == 0
    assert alice.memberships.get(project_id=data["id"]).role == Role.OWNER


def test_list_only_my_projects(alice, carol, project, client_for):
    client_for(carol).post(
        "/projects/", json={"name": "Carol's garden", "description": "Vegetables"}
    )

    response = client_for(alice).get("/projects/")
    assert [p["name"] for p in response.json()] == ["Website redesign"]


def test_non_member_gets_404(carol, project, client_for):
    response = client_for(carol).get(f"/projects/{project.id}")
    assert response.status_code == 404


def test_member_cannot_delete(bob, project, client_for):
    response = client_for(bob).delete(f"/projects/{project.id}")
    assert response.status_code == 403
    assert response.json() == {"detail": "This needs the owner role"}


def test_owner_can_update(alice, project, client_for):
    response = client_for(alice).put(
        f"/projects/{project.id}", json={"name": "Website v2", "description": "Round two"}
    )
    assert response.status_code == 200
    project.refresh_from_db()
    assert project.name == "Website v2"


def test_add_member_twice(alice, bob, project, client_for):
    response = client_for(alice).post(f"/projects/{project.id}/members", json={"username": "bob"})
    assert response.status_code == 409
    assert response.json() == {"detail": "bob is already a member"}
  • The highlighted tests pin down the two sides of get_project_or_404 from Part 4: a non-member gets 404, a member without the right role gets 403.
  • Tests can check the database as well as the response. test_create_project confirms that the creator became the owner, and test_owner_can_update reloads the project with refresh_from_db().
  • Each test starts with an empty database. Carol’s project in test_list_only_my_projects is gone before the next test runs.

Tasks: validation, workflows and filters

The task tests share one more fixture, a task in the project. Create tests/test_tasks.py:

# tests/test_tasks.py
from datetime import date

import pytest
from django.core.files.uploadedfile import SimpleUploadedFile

from tasks.models import Priority, Task, TaskStatus


@pytest.fixture
def task(project, alice):
    return project.tasks.create(title="Draft the sitemap", assignee=alice)


def test_create_task(alice, bob, project, client_for):
    label = project.labels.create(name="design")
    response = client_for(alice).post(
        f"/projects/{project.id}/tasks/",
        json={"title": "  Pick fonts ", "assignee_id": bob.id, "label_ids": [label.id]},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Pick fonts"
    assert data["assignee"]["username"] == "bob"
    assert data["labels"] == [{"id": label.id, "name": "design", "color": "#808080"}]
    assert data["status"] == "todo"


def test_blank_title(alice, project, client_for):
    response = client_for(alice).post(f"/projects/{project.id}/tasks/", json={"title": "   "})
    assert response.status_code == 422
    assert response.json()["detail"] == [
        {"type": "blank", "loc": ["body", "payload", "title"], "msg": "Title can't be blank"}
    ]


def test_urgent_task_needs_due_date(alice, project, client_for):
    response = client_for(alice).post(
        f"/projects/{project.id}/tasks/", json={"title": "Fix login", "priority": Priority.URGENT}
    )
    assert response.status_code == 422
    assert response.json()["detail"][0]["msg"] == "Value error, Urgent tasks need a due date"


def test_assignee_must_be_member(alice, carol, project, client_for):
    response = client_for(alice).post(
        f"/projects/{project.id}/tasks/", json={"title": "Pick fonts", "assignee_id": carol.id}
    )
    assert response.status_code == 400


def test_non_member_cannot_see_tasks(carol, task, client_for):
    response = client_for(carol).get(f"/projects/{task.project_id}/tasks/{task.id}")
    assert response.status_code == 404


def test_patch_changes_only_sent_fields(alice, task, client_for):
    response = client_for(alice).patch(
        f"/projects/{task.project_id}/tasks/{task.id}", json={"due_date": "2026-10-15"}
    )
    assert response.status_code == 200
    task.refresh_from_db()
    assert task.due_date == date(2026, 10, 15)
    assert task.title == "Draft the sitemap"
  • Fixtures can live in a test file too. task is only used here, so it doesn’t need to be in conftest.py.
  • test_blank_title compares the whole detail list. It holds the same blank error you saw in Part 3. A change to the error’s type, location or message breaks the test, which is what you want once clients depend on it.
  • test_create_task also checks that the title was stripped and that the nested assignee and labels from Part 2 are in the response.
  • Priority.URGENT can go straight into json=. IntegerChoices members are integers, so it’s sent as 4.

Many cases, one test

The status workflow from Part 3 has a table of allowed moves. @pytest.mark.parametrize turns a table of cases into one test each. Add it to tests/test_tasks.py:

# tests/test_tasks.py
@pytest.mark.parametrize(
    "old, new, status_code",
    [
        (TaskStatus.TODO, TaskStatus.IN_PROGRESS, 200),
        (TaskStatus.IN_PROGRESS, TaskStatus.DONE, 200),
        (TaskStatus.TODO, TaskStatus.DONE, 409),
        (TaskStatus.DONE, TaskStatus.TODO, 409),
    ],
)
def test_status_transitions(alice, task, client_for, old, new, status_code):
    Task.objects.filter(id=task.id).update(status=old)
    response = client_for(alice).post(
        f"/projects/{task.project_id}/tasks/{task.id}/status", json={"status": new}
    )
    assert response.status_code == status_code

The test sets the starting status directly in the database. Going through the API would make a test for done depend on two earlier moves working.

Filters work the same way: one fixture with a few tasks, and a table of query strings with the titles each should return. Add both:

# tests/test_tasks.py
@pytest.fixture
def sample_tasks(project, bob):
    bug = project.labels.create(name="bug")
    project.tasks.create(
        title="Fix mobile menu", assignee=bob, priority=Priority.URGENT, due_date=date(2026, 10, 1)
    ).labels.add(bug)
    project.tasks.create(
        title="Set up analytics",
        description="Page views",
        status=TaskStatus.IN_PROGRESS,
        due_date=date(2026, 10, 20),
    )
    project.tasks.create(
        title="Compress images",
        assignee=bob,
        priority=Priority.HIGH,
        due_date=date(2026, 10, 3),
        status=TaskStatus.DONE,
    )


@pytest.mark.parametrize(
    "query, titles",
    [
        ("", ["Fix mobile menu", "Set up analytics", "Compress images"]),
        ("status=todo&status=in_progress", ["Fix mobile menu", "Set up analytics"]),
        ("assignee=bob&label=bug", ["Fix mobile menu"]),
        ("due_before=2026-10-05", ["Fix mobile menu", "Compress images"]),
        ("search=VIEWS", ["Set up analytics"]),
        ("ordering=-priority", ["Fix mobile menu", "Compress images", "Set up analytics"]),
    ],
)
def test_filter_tasks(alice, project, sample_tasks, client_for, query, titles):
    response = client_for(alice).get(f"/projects/{project.id}/tasks/?{query}")
    assert response.status_code == 200
    assert [t["title"] for t in response.json()["items"]] == titles
    assert response.json()["count"] == len(titles)
  • The query string goes straight into the path, including a repeated status parameter. TestClient also takes query_params={"status": ["todo", "in_progress"]} if you prefer a dict.
  • The titles are compared as a list, so each case also checks the order: created_at by default, or the ordering parameter from Part 5.
  • search=VIEWS finds a match in the description and checks that the search ignores case.
  • The paginated response wraps the tasks in items. Checking count as well makes sure the total that pagination reports agrees with the filtered list.

Go deeper: Testing

Comments and attachments

The last tests cover the object-level rules from Parts 4 and 6, and file uploads. Add them to the end of tests/test_tasks.py:

# tests/test_tasks.py
def test_only_author_can_edit_comment(alice, bob, task, client_for):
    comment = task.comments.create(author=bob, body="Looks good")
    url = f"/projects/{task.project_id}/tasks/{task.id}/comments/{comment.id}"

    response = client_for(alice).put(url, json={"body": "Changed"})
    assert response.status_code == 403

    response = client_for(alice).delete(url)  # the owner may delete it
    assert response.status_code == 204


def test_upload_attachment(alice, task, client_for, settings):
    file = SimpleUploadedFile("notes.txt", b"Remember the footer")
    response = client_for(alice).post(
        f"/projects/{task.project_id}/tasks/{task.id}/attachments",
        data={"name": "Meeting notes"},
        FILES={"file": file},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Meeting notes"
    assert data["url"].startswith("http://testlocation/media/attachments/")

    attachment = task.attachments.get()
    assert (settings.MEDIA_ROOT / attachment.file.name).read_bytes() == b"Remember the footer"


def test_upload_rejects_file_type(alice, task, client_for):
    file = SimpleUploadedFile("setup.exe", b"MZ")
    response = client_for(alice).post(
        f"/projects/{task.project_id}/tasks/{task.id}/attachments", FILES={"file": file}
    )
    assert response.status_code == 422
    assert response.json()["detail"][0]["type"] == "file_type"
    assert not task.attachments.exists()


def test_delete_attachment(alice, bob, task, client_for, settings):
    file = SimpleUploadedFile("plan.pdf", b"%PDF-1.4")
    attachment = task.attachments.create(file=file, name="plan.pdf", uploaded_by=alice)
    url = f"/projects/{task.project_id}/tasks/{task.id}/attachments/{attachment.id}"

    response = client_for(bob).delete(url)
    assert response.status_code == 403

    response = client_for(alice).delete(url)
    assert response.status_code == 204
    assert not (settings.MEDIA_ROOT / attachment.file.name).exists()
  • SimpleUploadedFile is Django’s in-memory upload. It has a name, a size and content, which is all that check_upload and the FileField need.
  • data= fills the form fields and FILES= the files, the two halves of the multipart/form-data request that curl -F sent in Part 6.
  • TestClient’s fake request answers build_absolute_uri() with the host testlocation, so the resolver from Part 6 returns http://testlocation/media/....
  • The settings fixture gives the test the MEDIA_ROOT that test_settings set, so it can check the stored file on disk, and that delete_attachment removed it.

Go deeper: Testing

Running the tests

Run pytest from the folder with manage.py. As with every other command, run it through uv:

$ uv run pytest
============================= test session starts ==============================
...
django: version: 6.1.1, settings: locallibrary_config.settings (from ini)
...
collected 31 items

tests/test_auth.py ....                                                  [ 12%]
tests/test_projects.py ......                                            [ 32%]
tests/test_tasks.py ....................                                 [ 96%]
tests/test_auth.py .                                                     [100%]

============================== 31 passed in 0.52s ==============================

tests/test_auth.py shows up twice because pytest-django runs the tests that don’t use the database last. That’s test_missing_token. Each parametrized case is a test of its own, so the 12 test functions in test_tasks.py make 20 tests.

-k picks tests by name, and -v lists them. Each parametrized case gets an id built from its parameters:

$ uv run pytest tests/test_tasks.py -k filter -v
...
collecting ... collected 20 items / 14 deselected / 6 selected

tests/test_tasks.py::test_filter_tasks[-titles0] PASSED                  [ 16%]
tests/test_tasks.py::test_filter_tasks[status=todo&status=in_progress-titles1] PASSED [ 33%]
tests/test_tasks.py::test_filter_tasks[assignee=bob&label=bug-titles2] PASSED [ 50%]
tests/test_tasks.py::test_filter_tasks[due_before=2026-10-05-titles3] PASSED [ 66%]
tests/test_tasks.py::test_filter_tasks[search=VIEWS-titles4] PASSED      [ 83%]
tests/test_tasks.py::test_filter_tasks[ordering=-priority-titles5] PASSED [100%]

======================= 6 passed, 14 deselected in 0.31s =======================

When a check goes missing

A test suite proves its worth when someone breaks a rule by accident. Delete the author check from update_comment in tasks/api.py:

# tasks/api.py
    if comment.author != request.auth:
        raise HttpError(403, "Only the author can edit a comment")

Every other test still passes, and the endpoint still works for its author. Only the permission test notices:

$ uv run pytest
...
tests/test_tasks.py ................F...                                 [ 96%]
tests/test_auth.py .                                                     [100%]

=================================== FAILURES ===================================
______________________ test_only_author_can_edit_comment _______________________

alice = <User: alice>, bob = <User: bob>, task = <Task: Draft the sitemap>
client_for = <function client_for.<locals>.make_client at 0x7f6059af7a60>

    def test_only_author_can_edit_comment(alice, bob, task, client_for):
        comment = task.comments.create(author=bob, body="Looks good")
        url = f"/projects/{task.project_id}/tasks/{task.id}/comments/{comment.id}"
    
        response = client_for(alice).put(url, json={"body": "Changed"})
>       assert response.status_code == 403
E       assert 200 == 403
E        +  where 200 = <ninja.testing.client.NinjaResponse object at 0x7f6059b9a990>.status_code

tests/test_tasks.py:127: AssertionError
=========================== short test summary info ============================
FAILED tests/test_tasks.py::test_only_author_can_edit_comment - assert 200 ==...
========================= 1 failed, 30 passed in 0.59s =========================

pytest shows the fixtures the test received, the failing line, and the actual value: alice’s edit of bob’s comment returned 200. Put the two lines back before you continue.

Go deeper: Testing

Recap

  • pytest-django needs DJANGO_SETTINGS_MODULE in pytest.ini. It creates a fresh test database and rolls back each test.
  • Fixtures in conftest.py build users, a project with memberships, and a client_for(user) factory that sends a JWT with every request.
  • TestClient(api) calls operations directly, with paths relative to the API. Auth, validation and error handling still run.
  • Send bodies with json=, form fields with data=, files with FILES= and SimpleUploadedFile, and query strings in the path.
  • parametrize turns a table of cases, such as status moves or filter queries, into separate tests.
  • Point MEDIA_ROOT at tmp_path and use a fast password hasher, so tests stay clean and take under a second.

Next

TaskFlow is complete and tested. In Part 8: OpenAPI polish & versioning, you’ll improve the generated docs with summaries, descriptions and examples, and run a second API version next to the first.

3.8 - Part 8: OpenAPI polish & versioning

Descriptions, examples and deprecations in the docs, and a version 2 next to version 1.

TaskFlow works and is tested. Other people will build against it, though, and for them the interactive docs are the API. In this last part you’ll make the generated schema say more: what the API is for, what each operation does, and what each field expects. Then you’ll ship a breaking change the safe way, as a version 2 that runs next to version 1.

Describing the API

Open taskflow/api.py. NinjaAPI takes a version and a description next to the title you set in Part 1:

# taskflow/api.py
from ninja import NinjaAPI

from accounts.auth import JWTAuth

DESCRIPTION = (
    "Track the tasks of your team's projects: members, labels, comments and attachments.\n\n"
    "Get a token from `POST /auth/token`, then click **Authorize** and paste it."
)

api = NinjaAPI(
    title="TaskFlow API",
    version="1.0",
    description=DESCRIPTION,
    urls_namespace="taskflow",
    auth=JWTAuth(),
)
...

Leave the add_router() calls below it as they are for now. You’ll see the complete file once version 2 is in place.

The three values end up in the info block of /taskflow/api/openapi.json:

{
    "title": "TaskFlow API",
    "version": "1.0",
    "description": "Track the tasks of your team's projects: members, labels, comments and attachments.\n\nGet a token from `POST /auth/token`, then click **Authorize** and paste it."
}
  • The docs page shows the title with the version next to it, and renders the description as Markdown, so **Authorize** comes out bold.
  • version is a plain string. Django Ninja doesn’t parse it, but it does use it for the default URL namespace, as you’ll see at the end of this page. TaskFlow sets urls_namespace= itself, as it has since Part 1.

Go deeper: The NinjaAPI Instance

Summaries and descriptions

Every operation gets a summary, the one-line name in the docs. By default it’s the function name in title case: list_tasks becomes “List Tasks”, and create_task becomes “Create Task”. Good function names already give good summaries. When they don’t, pass summary=. A docstring becomes the operation’s description:

# tasks/api.py
@router.get("/", response=list[TaskOut])
@paginate(PageNumberPagination, page_size=20)
def list_tasks(
    request,
    project_id: Path[int],
    filters: Query[TaskFilter],
    ordering: TaskOrdering = "created_at",
):
    """List the tasks of a project, 20 per page. Filters combine with AND."""
    project = get_project_or_404(request.auth, project_id)
    tasks = filters.filter(project.tasks.select_related("assignee").prefetch_related("labels"))
    return tasks.order_by(ordering, "id")


...


@router.post("/{task_id}/status", response=TaskOut, summary="Move a task to another status")
def change_status(request, project_id: Path[int], task_id: int, payload: StatusIn):
    """
    Allowed moves:

    - `todo` to `in_progress`
    - `in_progress` to `todo` or `done`
    - `done` to `in_progress`

    Any other move returns **409 Conflict**.
    """
    task = get_task_or_404(request.auth, project_id, task_id)
    if payload.status not in ALLOWED_TRANSITIONS[task.status]:
        raise HttpError(409, f"Can't move a task from {task.status} to {payload.status}")
    task.status = payload.status
    task.save()
    return task

“Change Status” would be a fine summary too, but it doesn’t say what it changes. Here is the operation in /taskflow/api/openapi.json, with the parameters and responses left out:

{
    "operationId": "tasks_api_change_status",
    "summary": "Move a task to another status",
    "description": "Allowed moves:\n\n- `todo` to `in_progress`\n- `in_progress` to `todo` or `done`\n- `done` to `in_progress`\n\nAny other move returns **409 Conflict**.",
    "tags": [
        "tasks"
    ]
}
  • The docstring’s indentation is removed, and the docs render the list and the bold text as Markdown.
  • The docstring is the place for rules a client can’t see in the schema, such as the transitions from Part 3. The schema already says that status is one of three values. It can’t say which moves are allowed.
  • operationId is built from the module and function name. Client generators typically use it to name the generated method. Pass operation_id="..." to an operation to choose your own. It must be unique across the schema.
  • tags comes from Router(tags=["tasks"]) in Part 1.

Go deeper: Operations: summary, Operations: operation_id

Describing fields

Field names like title and due_date explain themselves. assignee_id and label_ids have rules the schema can’t show on its own: they must point to a member and to labels of the same project. Pydantic’s Field() adds a description and example values:

# tasks/schemas.py
class TaskIn(TaskUpdate):
    assignee_id: int | None = Field(None, description="A member of the project", examples=[2])
    label_ids: list[int] = Field([], description="Labels of the same project", examples=[[1]])

...


class TaskFilter(FilterSchema):
    status: Annotated[list[TaskStatus] | None, FilterLookup("status__in")] = None
    assignee: Annotated[str | None, FilterLookup("assignee__username")] = None
    label: Annotated[str | None, FilterLookup("labels__name")] = None
    due_before: Annotated[date | None, FilterLookup("due_date__lt")] = None
    search: Annotated[
        str | None, FilterLookup(["title__icontains", "description__icontains"])
    ] = Field(None, description="Text to find in the title or the description")

The first argument of Field() is the default, the same value the fields had before. Field is already imported in tasks/schemas.py from Part 6. The two fields in the TaskIn schema:

{
    "assignee_id": {
        "anyOf": [
            {
                "type": "integer"
            },
            {
                "type": "null"
            }
        ],
        "description": "A member of the project",
        "examples": [
            2
        ],
        "title": "Assignee Id"
    },
    "label_ids": {
        "default": [],
        "description": "Labels of the same project",
        "examples": [
            [
                1
            ]
        ],
        "items": {
            "type": "integer"
        },
        "title": "Label Ids",
        "type": "array"
    }
}
  • examples is a list of complete values. For a list field, each example is itself a list, hence [[1]].
  • The same works in a FilterSchema. search is a query parameter, and its description appears next to it in the docs:
{
    "in": "query",
    "name": "search",
    "schema": {
        "anyOf": [
            {
                "type": "string"
            },
            {
                "type": "null"
            }
        ],
        "description": "Text to find in the title or the description",
        "title": "Search"
    },
    "required": false,
    "description": "Text to find in the title or the description"
}

Go deeper: Query Parameters, Request Body

Hiding an internal endpoint

A load balancer or an uptime monitor needs a URL that answers without a token. It’s not part of the API you offer to clients, so it doesn’t belong in the docs. Register it on api directly, with no auth and include_in_schema=False:

# taskflow/api.py
@api.get("/health", auth=None, include_in_schema=False)
def health(request):
    return {"status": "ok"}

GET /taskflow/api/health works without a token:

{
    "status": "ok"
}

/taskflow/api/health isn’t in the paths of /taskflow/api/openapi.json, so the docs don’t show it. Hiding an operation is not a security measure: anyone who knows the URL can still call it. Protect it with auth if it returns anything private.

Go deeper: Operations: include_in_schema

A second version

A mobile team wants to use TaskFlow with a standard OAuth2 client library. The library expects the token response to use OAuth2’s field names: access_token, refresh_token, token_type and expires_in. TaskFlow returns access and refresh. Renaming the fields would break every client that already logs in, so the new shape goes into a version 2 of the API. Version 1 stays as it is.

The new schemas go into accounts/schemas.py:

# accounts/schemas.py
from typing import Literal

from django.contrib.auth.models import User
from ninja import ModelSchema, Schema
from pydantic import Field

...


class OAuthTokenOut(Schema):
    access_token: str
    refresh_token: str
    token_type: Literal["bearer"]
    expires_in: int = Field(description="Seconds until the access token expires", examples=[900])


class OAuthRefreshIn(Schema):
    refresh_token: str

The version 2 endpoints go into a new module:

# accounts/api_v2.py
from ninja import Router

from . import api as v1
from .schemas import OAuthRefreshIn, OAuthTokenOut, RefreshIn, TokenIn
from .tokens import ACCESS_TOKEN_LIFETIME

tokens = Router(tags=["auth"], auth=None)


def oauth_response(pair):
    return {
        "access_token": pair["access"],
        "refresh_token": pair["refresh"],
        "token_type": "bearer",
        "expires_in": ACCESS_TOKEN_LIFETIME,
    }


@tokens.post("/token", response=OAuthTokenOut)
def obtain_token(request, payload: TokenIn):
    """Exchange a username and password for an access and a refresh token."""
    return oauth_response(v1.obtain_token(request, payload))


@tokens.post("/refresh", response=OAuthTokenOut)
def refresh_token(request, payload: OAuthRefreshIn):
    """Exchange a refresh token for a new pair of tokens."""
    return oauth_response(v1.refresh_token(request, RefreshIn(refresh=payload.refresh_token)))
  • Only the shape changes, so version 2 calls the version 1 views. An operation decorator registers the function and returns it unchanged, so v1.obtain_token is still a plain Python function.
  • The HttpError(401, ...) raised inside a v1 view propagates to the v2 operation and becomes the same 401 response.
  • The login request body, TokenIn, doesn’t change. The refresh body does, from refresh to refresh_token.

Splitting the auth router

A router is the unit you mount into an API. You can’t mount a router without some of its operations, so everything that differs between versions must live in a router of its own. In accounts/api.py, signup stays on router, and the two token endpoints move to a new tokens router that only version 1 mounts:

# accounts/api.py
from django.contrib.auth import authenticate
from django.contrib.auth.models import User
from django.contrib.auth.password_validation import validate_password
from django.core.exceptions import ValidationError
from ninja import Router, Status
from ninja.errors import HttpError

from .schemas import RefreshIn, SignupIn, TokenIn, TokenOut, UserOut
from .tokens import create_token_pair, decode_token

router = Router(tags=["auth"], auth=None)  # shared by v1 and v2
tokens = Router(tags=["auth"], auth=None)  # v1 only


@router.post("/signup", response={201: UserOut})
def signup(request, payload: SignupIn):
    if User.objects.filter(username=payload.username).exists():
        raise HttpError(409, "This username is taken")
    try:
        validate_password(payload.password)
    except ValidationError as e:
        raise HttpError(400, " ".join(e.messages))
    user = User.objects.create_user(payload.username, password=payload.password)
    return Status(201, user)


@tokens.post("/token", response=TokenOut, deprecated=True)
def obtain_token(request, payload: TokenIn):
    """Use `POST /taskflow/api/v2/auth/token` instead."""
    user = authenticate(username=payload.username, password=payload.password)
    if user is None:
        raise HttpError(401, "Invalid username or password")
    return create_token_pair(user.id)


@tokens.post("/refresh", response=TokenOut, deprecated=True)
def refresh_token(request, payload: RefreshIn):
    """Use `POST /taskflow/api/v2/auth/refresh` instead."""
    claims = decode_token(payload.refresh, "refresh")
    if claims is None or not User.objects.filter(id=claims["sub"], is_active=True).exists():
        raise HttpError(401, "Invalid or expired refresh token")
    return create_token_pair(int(claims["sub"]))

deprecated=True tells clients to move on without breaking them. The endpoints keep working, and the docs show them as deprecated. The docstring says where to go instead. In the v1 schema:

{
    "summary": "Obtain Token",
    "description": "Use `POST /taskflow/api/v2/auth/token` instead.",
    "deprecated": true
}

Go deeper: Operations: deprecated

Mounting both versions

Version 2 is a second NinjaAPI instance. Here is the complete taskflow/api.py:

# taskflow/api.py
from ninja import NinjaAPI

from accounts.auth import JWTAuth

DESCRIPTION = (
    "Track the tasks of your team's projects: members, labels, comments and attachments.\n\n"
    "Get a token from `POST /auth/token`, then click **Authorize** and paste it."
)

api = NinjaAPI(
    title="TaskFlow API",
    version="1.0",
    description=DESCRIPTION,
    urls_namespace="taskflow",
    auth=JWTAuth(),
)

api.add_router("/auth", "accounts.api.router")
api.add_router("/auth", "accounts.api.tokens")
api.add_router("/projects", "projects.api.router")
api.add_router("/projects/{project_id}/tasks", "tasks.api.router")


@api.get("/health", auth=None, include_in_schema=False)
def health(request):
    return {"status": "ok"}


api_v2 = NinjaAPI(
    title="TaskFlow API",
    version="2.0",
    description=DESCRIPTION + "\n\nNew in 2.0: token responses use OAuth2 field names.",
    urls_namespace="taskflow-v2",
    auth=JWTAuth(),
)

api_v2.add_router("/auth", "accounts.api.router")
api_v2.add_router("/auth", "accounts.api_v2.tokens")
api_v2.add_router("/projects", "projects.api.router")
api_v2.add_router("/projects/{project_id}/tasks", "tasks.api.router")
  • Both APIs mount the same signup, projects and tasks routers. Each mount gets its own copy of the router’s operations, bound to that API, so the two versions don’t interfere. A bug fix in tasks/api.py reaches both versions at once.
  • Only the tokens routers differ: accounts.api.tokens for version 1, and accounts.api_v2.tokens for version 2.
  • health is registered on api only. GET /taskflow/api/v2/health returns 404.
  • Mounting two routers at the same prefix, like the two /auth mounts, is fine as long as their paths don’t overlap.

Mounting the same router twice on one API is different. Every operation’s URL name must be unique within an API, so Django Ninja refuses the second mount:

ninja.errors.ConfigError: Router is already mounted to this API. When mounting the same router multiple times, you must provide unique url_name_prefix for each mount.

Pass url_name_prefix="..." to the second add_router() call if you really need both mounts.

Then give version 2 its own URL prefix. Replace the TaskFlow block you added to the end of locallibrary_config/urls.py in Part 1. Version 1 stays at /taskflow/api/, so existing clients don’t notice a thing:

# locallibrary_config/urls.py
# Mount the TaskFlow API from the Django Ninja Tutorial
from taskflow.api import api as taskflow_api, api_v2 as taskflow_api_v2

urlpatterns += [
    path('taskflow/api/v2/', taskflow_api_v2.urls),
    path('taskflow/api/', taskflow_api.urls),
]

Go deeper: Versioning, Routers

Try it

Log in through version 2 with POST /taskflow/api/v2/auth/token and {"username": "alice", "password": "wonderland-42"}:

{
    "access_token": "eyJhbGciOiAiSFMyNTYiLCAidHlwIjogIkpXVCJ9.eyJzdWIiOiAiMSIsICJ0eXBlIjogImFjY2VzcyIsICJleHAiOiAxNzkwNTk1OTIyfQ.uKbqv565JqGDjMxFrjzfwEF2zf5mmDv4fiFK2W0lI64",
    "refresh_token": "eyJhbGciOiAiSFMyNTYiLCAidHlwIjogIkpXVCJ9.eyJzdWIiOiAiMSIsICJ0eXBlIjogInJlZnJlc2giLCAiZXhwIjogMTc5MTE5OTgyMn0.sPAI0PFLkazt-NUORo9u7INLum8b9HrTKnqRT1-XEY8",
    "token_type": "bearer",
    "expires_in": 900
}

The tokens are the same JWTs as before, and both APIs check them with the same JWTAuth. So a token from either version works on both: with this access_token, GET /taskflow/api/projects/ and GET /taskflow/api/v2/projects/ both return 200.

The old field name no longer works on version 2. POST /taskflow/api/v2/auth/refresh with {"refresh": "..."} returns 422:

{
    "detail": [
        {
            "type": "missing",
            "loc": [
                "body",
                "payload",
                "refresh_token"
            ],
            "msg": "Field required"
        }
    ]
}

With {"refresh_token": "..."} it returns a new token pair in the version 2 shape.

Now compare the two docs pages. http://127.0.0.1:8000/taskflow/api/docs shows version 1.0, with the two token endpoints marked as deprecated. http://127.0.0.1:8000/taskflow/api/v2/docs shows version 2.0 and the extra line of description, and it loads its schema from /taskflow/api/v2/openapi.json. Both schemas have the same 14 paths. Only POST /auth/token and POST /auth/refresh differ, and the component schemas TokenOut and RefreshIn of version 1 become OAuthTokenOut and OAuthRefreshIn in version 2.

Go deeper: Versioning: multiple NinjaAPI instances

URL namespaces

Each NinjaAPI registers its URLs under its own namespace, api-<version> by default. Since Part 1, TaskFlow has set urls_namespace= instead, and version 2 does the same. The operation names come from the function names, so reverse() needs the namespace to tell the two versions apart:

$ uv run python manage.py shell
>>> from django.urls import reverse
>>> reverse('taskflow:obtain_token')
'/taskflow/api/auth/token'
>>> reverse('taskflow-v2:obtain_token')
'/taskflow/api/v2/auth/token'
>>> reverse('taskflow-v2:openapi-view')
'/taskflow/api/v2/docs'

In the original tutorial, TaskFlow is the only API in the project, so setting version on both APIs is enough to keep their namespaces apart: api-1.0 and api-2.0. Your locallibrary has a third API, the one from the Quickstart, and it still has the default version, "1.0.0", and the default namespace, api-1.0.0. In Parts 1 to 7, TaskFlow had the default version too. Two APIs with the same version share a namespace, and manage.py check warns about it. Without urls_namespace="taskflow", you would have seen this since Part 1:

$ uv run python manage.py check
System check identified some issues:

WARNINGS:
?: (urls.W005) URL namespace 'api-1.0.0' isn't unique. You may not be able to reverse all URLs in this namespace

System check identified 1 issue (0 silenced).

In that setup, reverse() can return a URL from the wrong API. That’s why both TaskFlow APIs set their own urls_namespace=, whatever their versions: if two APIs might share a version, give them distinct urls_namespace= values.

Go deeper: URLs & Reverse

Testing both versions

A version is a promise to its clients, so it deserves its own tests. TestClient takes any NinjaAPI, so each version gets a client:

# tests/test_versions.py
from ninja.testing import TestClient

from taskflow.api import api, api_v2

client_v1 = TestClient(api)
client_v2 = TestClient(api_v2)


def test_v2_token_response(alice):
    response = client_v2.post("/auth/token", json={"username": "alice", "password": "wonderland-42"})
    assert response.status_code == 200
    data = response.json()
    assert data["token_type"] == "bearer"
    assert data["expires_in"] == 900

    # Both versions accept the same access token
    headers = {"Authorization": f"Bearer {data['access_token']}"}
    assert client_v1.get("/projects/", headers=headers).status_code == 200
    assert client_v2.get("/projects/", headers=headers).status_code == 200


def test_v2_refresh(alice):
    tokens = client_v2.post(
        "/auth/token", json={"username": "alice", "password": "wonderland-42"}
    ).json()
    response = client_v2.post("/auth/refresh", json={"refresh_token": tokens["refresh_token"]})
    assert response.status_code == 200
    assert set(response.json()) == {"access_token", "refresh_token", "token_type", "expires_in"}

    response = client_v2.post("/auth/refresh", json={"refresh_token": tokens["access_token"]})
    assert response.status_code == 401

Paths are relative to the API, so /auth/token on client_v2 is /taskflow/api/v2/auth/token. The Part 7 tests run against api and still pass, because version 1 behaves exactly as before:

$ pytest
...
tests/test_auth.py ....                                                  [ 12%]
tests/test_projects.py ......                                            [ 30%]
tests/test_tasks.py ....................                                 [ 90%]
tests/test_versions.py ..                                                [ 96%]
tests/test_auth.py .                                                     [100%]

============================== 33 passed in 0.55s ==============================

pytest-django runs the tests that use the database first, which is why the last test of test_auth.py, test_missing_token, runs at the end.

Go deeper: Testing

Recap

  • NinjaAPI(title=..., version=..., description=...) fills the info block of the schema. The description is Markdown.
  • Summaries come from function names unless you pass summary=. Docstrings become descriptions, and operationId comes from the module and function name unless you pass operation_id=.
  • Field(description=..., examples=[...]) documents fields in body schemas and in filter schemas alike.
  • include_in_schema=False hides an endpoint from the docs, and deprecated=True marks one as on its way out. Both keep working.
  • A new version is a second NinjaAPI at its own URL prefix. Routers that didn’t change are mounted on both, and whatever changed goes into a router of its own.
  • Different versions get different URL namespaces, so reverse() and manage.py check stay happy.

Next

That’s the whole tutorial. TaskFlow has routers per app, nested schemas and computed fields, validation, JWT auth with roles, filtering and pagination, file uploads, a test suite and two documented API versions. From here, the Guide covers each feature in depth, including the ones this tutorial skipped, such as async views, throttling, custom renderers and parsers, and webhooks.