This is the multi-page printable view of this section.
Click here to print.
Return to the regular view of this page.
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
- Project setup & layout: one router per app, schemas from models, and task URLs nested under projects.
- Relations & computed fields: assignees and labels as nested output, computed and annotated fields, and fewer queries.
- Validation & workflows: status and priority choices, field and model validators,
PATCH updates and status transitions.
- Auth & permissions: JWT tokens from scratch, project memberships with roles, and object-level rules for comments.
- Filtering, search, ordering & pagination: a
FilterSchema for tasks, safe sort orders, and page-number and cursor pagination.
- File uploads & forms: multipart uploads with form fields, file validation, media files and absolute URLs.
- Testing: a pytest suite with fixtures and
TestClient, covering validation, permissions, filters and uploads.
- 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.
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.
What you'll learn
- How to lay out a multi-app project: one
Router per app, one NinjaAPI for the project
- How to mount routers with
add_router() and group them in the docs with tags
- How to generate input and output schemas with
ModelSchema
- How to nest one app’s endpoints under another app’s URL, with
Path[...] for prefix parameters
Time: about 15 minutes
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.
Note
The Quick Start returned (status, body) tuples. They still work, but they’re deprecated in favor of Status(...), which is what TaskFlow uses throughout.
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.
Why urls_namespace?
The original tutorial writes NinjaAPI(title="TaskFlow API"). Your locallibrary already has a NinjaAPI, the one in locallibrary_config/api.py from the Quickstart. Each NinjaAPI registers its URLs under a Django URL namespace, which is api-<version> by default, and both APIs have the default version, 1.0.0. urls_namespace="taskflow" gives TaskFlow its own namespace, so the two APIs don’t get in each other’s way. Part 8 explains namespaces in more detail.
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.
Don't forget Path[...] on prefix parameters
Django Ninja only treats an argument as a path parameter automatically if it appears in the operation’s own path. It doesn’t look at the router’s prefix. If you write project_id: int in list_tasks, it becomes a required query parameter. The docs then ask for ?project_id=, and the {project_id} in the URL is ignored. GET /taskflow/api/projects/1/tasks/ fails:
{
"detail": [
{
"type": "missing",
"loc": ["query", "project_id"],
"msg": "Field required"
}
]
}
With Path[int], the value is read from the URL where it belongs.
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.
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.
What you'll learn
- How to return related objects as nested schemas, while clients send plain ids
- How to set a many-to-many relation after creating an object
- How to compute a field with a
resolve_<field> static method
- How to read ORM annotations and reverse relations into a schema
- How to spot N+1 queries and fix them with
select_related and prefetch_related
Time: about 15 minutes
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)
🚨 Don't delete db.sqlite3
The original tutorial suggests deleting db.sqlite3 so that your ids match the ones on the pages. Don’t. That file also holds your superuser, the users you made in Prep 06, every book, author, and genre in your catalog, and your chat rooms and messages.
You don’t need to delete it, either. Project, Label, Task, and the models from later parts get brand-new tables, so their ids start at 1 and match the pages as long as you follow along exactly. Only the user ids are different. In the original, alice is user 1 and bob is user 2 (and Carol, in Part 4, is user 3). Your locallibrary already has users, so yours come after them. The last line above prints them; yours may not be 4 and 5.
Write down alice’s and bob’s ids. Wherever this tutorial sends a user id in a request, use yours. Wherever it shows a user id in a response, expect yours.
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>.
Note
Fields you declare on a ModelSchema come first in the output, before the fields from Meta.fields. That’s why assignee and labels appear before id in the responses below.
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.
Tip
A single get_task response has no N+1 problem, so it uses a plain get_object_or_404. Optimize the lists first. They’re where the query count grows with the data.
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 - 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.
What you'll learn
- How Django
choices become enums in your schemas and in OpenAPI
- How to write field validators and model validators, with your own error messages
- What a
422 validation error looks like, and how to read it
- How to accept partial updates with
PatchDict, and what it changes about validation
- How to reject an invalid state change with
HttpError
Time: about 15 minutes
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"
}
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.
Why check_fields=False?
Pydantic checks that a validator’s field names exist when the class is created. ModelSchema adds the Meta.fields afterwards, so without check_fields=False a validator for title fails with PydanticUserError: Decorators defined with incorrect fields. Fields you declare yourself, such as priority, don’t need it.
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.
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.
What you'll learn
- How to issue and verify JWT access and refresh tokens with nothing but the standard library
- How to protect a whole API with an
HttpBearer authenticator, and open up a few endpoints with auth=None
- How to read the current user from
request.auth
- How to write one permission helper that returns
404 to non-members and 403 to members without the right role
- How to scope every query to the user’s projects, and add object-level rules for comments
Time: about 15 minutes
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 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.
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.
What you'll learn
- How to turn query parameters into ORM lookups with
FilterSchema and FilterLookup
- How to search several fields at once, and accept a list of values for one parameter
- How to restrict an
ordering parameter to a fixed set of values with Literal
- How to paginate a list with
@paginate and PageNumberPagination
- How cursor pagination works, and when to prefer it
Time: about 15 minutes
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
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.
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.
What you'll learn
- How to accept a file together with form fields using
File[...] and Form[...]
- How to validate an upload with a reusable annotated type, so errors come back as
422
- How to configure
MEDIA_ROOT and serve uploaded files during development
- How to build absolute URLs in a response with a resolver that reads the request
- How to delete a file together with its database row, behind a permission check
Time: about 15 minutes
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:
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.
Note
Django only fills request.FILES for POST requests. Django Ninja raises a ConfigError if you declare a File parameter on a PUT or PATCH operation, unless you add its compatibility middleware. The File Uploads guide explains how.
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.
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.
What you'll learn
- How to set up pytest and pytest-django for a Django Ninja project
- How to write fixtures for users, projects with memberships, and logged-in clients
- How
TestClient calls an API, and how to send tokens, query strings and files with it
- How to test validation errors, permissions and filters, and cover many cases with
parametrize
- How to keep uploads and password hashing from slowing down or littering your tests
Time: about 15 minutes
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
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.
Tip
TestClient skips urls.py and middleware to stay fast. To test the full stack, such as a custom middleware or the URL prefix, use Django’s own django.test.Client with the /taskflow/api/... URLs. Django Ninja views are normal Django views and work with it unchanged.
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.
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.
What you'll learn
- How to set the API’s title, version and description
- How summaries, docstrings and
Field(description=..., examples=...) show up in the schema
- How to hide an internal endpoint and mark old ones as deprecated
- How to run two
NinjaAPI versions side by side, sharing the routers that didn’t change
- How URL namespaces keep
reverse() working with more than one API
Time: about 15 minutes
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:
/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.