django-bolt

name	django-bolt
description	Django Bolt - Rust-powered high-performance API framework, 60k+ RPS, decorator routing, built-in auth, async ORM
metadata	{"author":"OSS AI Skills","version":"1.0.0","tags":["python","django","bolt","api","rust","performance","async","high-performance"]}

Django Bolt

High-performance fully typed API framework for Django — Faster than FastAPI, but with Django ORM, Django Admin, and Django packages.

Django-Bolt is a Rust-powered API framework achieving 60k+ RPS. Uses Actix Web for HTTP, PyO3 for Python bridging, and msgspec for serialization.

Overview

Architecture:

• HTTP Server: Actix Web (Rust) — one of the fastest HTTP frameworks
• Python Bridge: PyO3 — seamless Rust-Python integration
• Serialization: msgspec — 5-10x faster than stdlib
• Routing: matchit — zero-copy path matching
• Async Runtime: Tokio

Why Django Bolt?

• No gunicorn or uvicorn needed — deploy directly
• Full Django ORM integration with async support
• Built-in auth (JWT, API Key) in Rust (no Python GIL)
• OpenAPI auto-generation
• Compatible with existing Django packages

---

Installation

pip install django-bolt

# settings.py
INSTALLED_APPS = [
    ...
    "django_bolt"
    ...
]

---

Quick Start

Basic API

# api.py
from django_bolt import BoltAPI
from django.contrib.auth import get_user_model
import msgspec

User = get_user_model()

api = BoltAPI()

class UserSchema(msgspec.Struct):
    id: int
    username: str

@api.get("/users/{user_id}")
async def get_user(user_id: int) -> UserSchema:
    # Response is type validated
    user = await User.objects.aget(id=user_id)
    # Django ORM works without any setup
    return {"id": user.id, "username": user.username}

Running the Server

# Development
python manage.py runbolt --dev

# Production (standalone)
python manage.py runbolt

---

Routing

HTTP Methods

from django_bolt import BoltAPI

api = BoltAPI()

@api.get("/endpoint")
async def get_handler(request):
    return {"message": "GET"}

@api.post("/endpoint")
async def post_handler(request):
    data = await request.json()
    return {"received": data}

@api.put("/endpoint")
async def put_handler(request):
    return {"message": "PUT"}

@api.delete("/endpoint/{id}")
async def delete_handler(id: int):
    return {"deleted": id}

@api.patch("/endpoint")
async def patch_handler(request):
    return {"message": "PATCH"}

Path Parameters

@api.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id}

@api.get("/posts/{post_id}/comments/{comment_id}")
async def get_comment(post_id: int, comment_id: int):
    return {"post_id": post_id, "comment_id": comment_id}

Query Parameters

from django_bolt import Query

@api.get("/search")
async def search_handler(
    query: str = Query(...),
    limit: int = Query(10),
    offset: int = Query(0)
):
    return {"query": query, "limit": limit, "offset": offset}

Request Body

import msgspec

class CreateUserRequest(msgspec.Struct):
    username: str
    email: str
    password: str

@api.post("/users")
async def create_user(request, body: CreateUserRequest):
    # body is automatically validated
    user = await User.objects.acreate(
        username=body.username,
        email=body.email
    )
    return {"id": user.id, "username": user.username}

---

Authentication

JWT Authentication

from django_bolt.auth import JWTBearer, jwt_required
from django_bolt.response import Json

auth = JWTBearer(secret_key="your-secret-key")

@api.get("/protected", guards=[jwt_required])
async def protected_handler(request):
    return {"user": request.user.id}

API Key Authentication

from django_bolt.auth import APIKeyBearer, api_key_required

auth = APIKeyBearer()

@api.get("/api-protected", guards=[api_key_required])
async def api_protected_handler(request):
    return {"message": "API key authenticated"}

Custom Authentication

from django_bolt.auth import BaseAuth, AuthResult
from django.contrib.auth import get_user_model

User = get_user_model()

class CustomAuth(BaseAuth):
    async def authenticate(self, request) -> AuthResult:
        token = request.headers.get("Authorization")
        if token and token.startswith("Bearer "):
            # Validate token
            user = await self.get_user(token)
            return AuthResult(user=user)
        return AuthResult()

async def get_user(self, token: str):
    # Your logic to get user from token
    try:
        return await User.objects.aget(id=int(token.split("_")[1]))
    except:
        return None

---

Permissions & Guards

Built-in Guards

from django_bolt.auth import IsAuthenticated, HasPermission, HasRole

# Require authentication
@api.get("/private", guards=[IsAuthenticated])
async def private_handler(request):
    return {"user_id": request.user.id}

# Require specific permission
@api.get("/edit-post", guards=[HasPermission("blog.change_post")])
async def edit_post_handler(request):
    return {"can_edit": True}

# Require role
@api.get("/admin-only", guards=[HasRole("admin")])
async def admin_handler(request):
    return {"access": "granted"}

Custom Guards

from django_bolt.auth import BaseGuard, AuthResult

class CustomGuard(BaseGuard):
    async def check(self, request) -> bool:
        # Your logic
        return request.headers.get("X-Custom-Header") == "secret"

---

Middleware

Built-in Middleware

from django_bolt.middleware import CORSMiddleware, RateLimitMiddleware, CompressionMiddleware

api = BoltAPI(
    middleware=[
        CORSMiddleware(
            allow_origins=["*"],
            allow_methods=["*"],
            allow_headers=["*"],
        ),
        RateLimitMiddleware(requests=100, window=60),  # 100 requests per minute
        CompressionMiddleware(),
    ]
)

Django Middleware Integration

from django.middleware.security import SecurityMiddleware

api = BoltAPI(
    django_middleware=[
        SecurityMiddleware,
        # Your Django middleware here
    ]
)

---

Responses

JSON Response

from django_bolt.response import Json

@api.get("/json")
async def json_handler(request):
    return Json({"key": "value"})

# Or shorthand (automatically JSON serialized)
@api.get("/auto-json")
async def auto_json_handler(request):
    return {"key": "value"}

HTML Response

from django_bolt.response import Html

@api.get("/html")
async def html_handler(request):
    return Html("<h1>Hello World</h1>")

Streaming Response (SSE)

from django_bolt.response import StreamingResponse

async def event_stream():
    for i in range(10):
        yield f"data: message {i}\n\n"
    # Or use Server-Sent Events format
    yield {"event": "message", "data": {"count": i}}

@api.get("/stream")
async def stream_handler(request):
    return StreamingResponse(event_stream())

File Response

from django_bolt.response import FileResponse

@api.get("/download")
async def download_handler(request):
    return FileResponse(
        path="/path/to/file.pdf",
        filename="document.pdf",
        content_type="application/pdf"
    )

Redirect Response

from django_bolt.response import Redirect

@api.get("/redirect")
async def redirect_handler(request):
    return Redirect(url="https://example.com", status=302)

---

Class-Based Views

ViewSet

from django_bolt.views import ViewSet, route

class UserViewSet(ViewSet):
    @route.get("/users")
    async def list(self, request):
        users = await User.objects.alist()
        return {"users": [{"id": u.id, "username": u.username} for u in users]}

    @route.get("/users/{pk}")
    async def retrieve(self, request, pk: int):
        user = await User.objects.aget(id=pk)
        return {"id": user.id, "username": user.username}

    @route.post("/users")
    async def create(self, request):
        data = await request.json()
        user = await User.objects.acreate(**data)
        return {"id": user.id}

    @route.put("/users/{pk}")
    async def update(self, request, pk: int):
        data = await request.json()
        user = await User.objects.aget(id=pk)
        for k, v in data.items():
            setattr(user, k, v)
        await user.asave()
        return {"id": user.id}

    @route.delete("/users/{pk}")
    async def destroy(self, request, pk: int):
        user = await User.objects.aget(id=pk)
        await user.adelete()
        return {"deleted": True}

api.register_viewset(UserViewSet, prefix="/api")

ModelViewSet

from django_bolt.views import ModelViewSet
from django.contrib.auth import get_user_model
from django_bolt.serializers import ModelSerializer

User = get_user_model()

class UserSerializer(ModelSerializer):
    class Meta:
        model = User
        fields = ["id", "username", "email"]

class UserModelViewSet(ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer

api.register_viewset(UserModelViewSet, prefix="/api")

---

Serializers (msgspec)

msgspec provides 5-10x faster JSON serialization than Python's stdlib.

Basic Struct

import msgspec

class UserSchema(msgspec.Struct):
    id: int
    username: str
    email: str
    is_active: bool = True

@api.post("/users")
async def create_user(request, body: UserSchema):
    # body is automatically validated
    user = await User.objects.acreate(
        username=body.username,
        email=body.email,
        is_active=body.is_active
    )
    return {"id": user.id}

With Validation

import msgspec

class UserCreateSchema(msgspec.Struct):
    username: str
    email: str
    password: str
    
    def __post_init__(self):
        if len(self.password) < 8:
            raise msgspec.ValidationError("Password must be at least 8 characters")
        if "@" not in self.email:
            raise msgspec.ValidationError("Invalid email format")

Nested Structures

class AddressSchema(msgspec.Struct):
    street: str
    city: str
    zip_code: str
    country: str

class UserSchema(msgspec.Struct):
    id: int
    username: str
    address: AddressSchema | None = None

@api.get("/users/{user_id}")
async def get_user_with_address(user_id: int):
    user = await User.objects.aget(id=user_id)
    return {
        "id": user.id,
        "username": user.username,
        "address": {
            "street": user.street,
            "city": user.city,
            "zip_code": user.zip_code,
            "country": user.country
        } if user.street else None
    }

---

OpenAPI / API Documentation

Django Bolt auto-generates OpenAPI documentation.

Access Docs

• Swagger: /docs
• ReDoc: /redoc
• Scalar: /scalar
• RapidDoc: /rapiddoc

Configure OpenAPI

from django_bolt import BoltAPI
from django_bolt.openapi import OpenAPIInfo

api = BoltAPI(
    info=OpenAPIInfo(
        title="My API",
        version="1.0.0",
        description="API description",
    )
)

---

Testing

Test Client

from django_bolt.test import AsyncAPITestClient

class UserAPITest(AsyncAPITestClient):
    async def test_create_user(self):
        response = await self.post(
            "/api/users",
            json={"username": "testuser", "email": "[e-mail]"}
        )
        self.assertEqual(response.status_code, 201)
        data = await response.json()
        self.assertEqual(data["username"], "testuser")

    async def test_get_user(self):
        # Create user first
        user = await User.objects.acreate(username="testuser", email="[e-mail]")
        
        response = await self.get(f"/api/users/{user.id}")
        self.assertEqual(response.status_code, 200)

---

Performance Benchmarks

Standard Endpoints

Endpoint Type	Requests/sec
Root endpoint	~100,000 RPS
JSON parsing/validation (10kb)	~83,700 RPS
Path + Query parameters	~85,300 RPS
HTML response	~100,600 RPS
Redirect response	~96,300 RPS
Form data handling	~76,800 RPS
ORM reads (SQLite, 10 records)	~13,000 RPS

Streaming (SSE)

10,000 concurrent clients:

• Total Throughput: 9,489 messages/sec
• Successful Connections: 100%
• CPU Usage: 11.9% average

---

Configuration

Settings

# settings.py

# Optional: Configure Bolt
BOLT = {
    "HOST": "0.0.0.0",
    "PORT": 8000,
    "DEBUG": False,
    "workers": 4,  # Number of worker processes
}

# Optional: JWT settings
JWT_SECRET_KEY = "your-secret-key"
JWT_ALGORITHM = "HS256"
JWT_EXPIRATION = 3600  # seconds

Production Deployment

# Run in production mode
python manage.py runbolt --workers 4

---

Error Handling

HTTP Exceptions

from django_bolt.exceptions import HTTPException

@api.get("/users/{user_id}")
async def get_user(user_id: int):
    try:
        user = await User.objects.aget(id=user_id)
    except User.DoesNotExist:
        raise HTTPException(status_code=404, detail="User not found")
    return {"id": user.id, "username": user.username}

# Custom error with extra data
raise HTTPException(
    status_code=400,
    detail={
        "error": "validation_failed",
        "fields": {"email": "Invalid email format"}
    }
)

Validation Error Formatting

import msgspec

class CreateUserRequest(msgspec.Struct):
    username: str
    email: str
    age: int

# Automatic validation errors from msgspec
@api.post("/users")
async def create_user(request, body: CreateUserRequest):
    # If body doesn't match schema, returns 422 with details:
    # {"detail": [{"loc": ["body", "age"], "msg": "expected int", "type": "type_error"}]}
    return {"username": body.username}

Global Exception Handler

from django_bolt.exceptions import exception_handler

def custom_exception_handler(exc, request):
    # Log the exception
    import logging
    logger = logging.getLogger(__name__)
    logger.error(f"Unhandled exception: {exc}")
    
    return {"error": "internal_server_error", "detail": str(exc)}, 500

api = BoltAPI(exception_handler=custom_exception_handler)

---

Pagination

Offset Pagination

from django_bolt import BoltAPI, Query

@api.get("/users")
async def list_users(
    request,
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
):
    offset = (page - 1) * page_size
    users = User.objects.all().order_by("id")
    
    total = await users.acount()
    items = await users[offset:offset + page_size].alist()
    
    return {
        "items": [{"id": u.id, "username": u.username} for u in items],
        "total": total,
        "page": page,
        "page_size": page_size,
        "pages": (total + page_size - 1) // page_size,
    }

Cursor-Based Pagination

from django_bolt import Query

@api.get("/posts")
async def list_posts(
    request,
    cursor: int = Query(None),
    limit: int = Query(20, le=100),
):
    posts = Post.objects.all().order_by("-id")
    
    if cursor:
        posts = posts.filter(id__lt=cursor)
    
    items = await posts[:limit + 1].alist()
    has_next = len(items) > limit
    items = items[:limit]
    
    return {
        "items": [{"id": p.id, "title": p.title} for p in items],
        "next_cursor": items[-1].id if has_next and items else None,
        "has_next": has_next,
    }

---

Django ORM Patterns

Async QuerySet Operations

from django_bolt import BoltAPI

api = BoltAPI()

# Async iteration
@api.get("/posts")
async def list_posts(request):
    posts = []
    async for post in Post.objects.all().order_by("-created_at")[:20]:
        posts.append({"id": post.id, "title": post.title})
    return {"posts": posts}

# select_related / prefetch_related
@api.get("/articles/{article_id}")
async def get_article(article_id: int):
    article = await Article.objects.select_related("author", "category").aget(id=article_id)
    return {
        "id": article.id,
        "title": article.title,
        "author": article.author.name,
        "category": article.category.name,
    }

# Bulk operations
@api.post("/articles/bulk")
async def bulk_create_articles(request):
    articles = await Article.objects.abulk_create([
        Article(title=f"Article {i}", content="...")
        for i in range(100)
    ])
    return {"created": len(articles)}

# Count with Exists (optimized)
from django.db.models import Exists, OuterRef

@api.get("/posts/with-comments")
async def posts_with_comments(request):
    posts = Post.objects.annotate(
        has_comments=Exists(Comment.objects.filter(post_id=OuterRef("pk")))
    )
    result = []
    async for post in posts:
        result.append({"id": post.id, "has_comments": post.has_comments})
    return {"posts": result}

Transactions

from django.db import transaction

@api.post("/orders")
async def create_order(request):
    data = await request.json()
    
    async with transaction.atomic():
        order = await Order.objects.acreate(
            user_id=data["user_id"],
            total=data["total"],
        )
        for item in data["items"]:
            await OrderItem.objects.acreate(
                order=order,
                product_id=item["product_id"],
                quantity=item["quantity"],
            )
    
    return {"order_id": order.id}

---

File Uploads

Single File Upload

from django_bolt import BoltAPI

@api.post("/upload")
async def upload_file(request):
    file = await request.file("document")
    if not file:
        return {"error": "No file provided"}, 400
    
    # file properties
    content = await file.read()
    filename = file.filename
    content_type = file.content_type
    
    # Save via Django's default storage
    from django.core.files.storage import default_storage
    path = default_storage.save(f"uploads/{filename}", file)
    
    return {"path": path, "size": len(content)}

Multiple Files

@api.post("/upload/multiple")
async def upload_multiple(request):
    files = await request.files("documents")
    paths = []
    for file in files:
        from django.core.files.storage import default_storage
        path = default_storage.save(f"uploads/{file.filename}", file)
        paths.append(path)
    return {"uploaded": len(paths), "paths": paths}

---

Background Tasks with Django 6.0

Django Bolt works seamlessly with Django 6.0's built-in Tasks Framework:

from django.tasks import task

@task
def process_video(video_path: str):
    # Heavy processing in background
    import subprocess
    subprocess.run(["ffmpeg", "-i", video_path, "-vcodec", "h264", f"{video_path}.mp4"])

@api.post("/videos")
async def upload_video(request):
    file = await request.file("video")
    from django.core.files.storage import default_storage
    path = default_storage.save(f"videos/{file.filename}", file)
    
    # Enqueue background task
    process_video.enqueue(path)
    
    return {"path": path, "status": "processing"}

---

Comparison with Alternatives

When to Choose Django Bolt

Feature	Django Bolt	Django REST Framework	Django Ninja
RPS	60,000+	~1,000	~3,000
Auth in Rust	✅	❌	❌
msgspec serialization	✅	❌	❌
OpenAPI auto-gen	✅	Needs drf-spectacular	✅
Django Admin	✅	✅	✅
Django ORM	✅ (async)	✅ (sync)	✅ (async)
Serializer	msgspec	DRF Serializers	Pydantic
Middleware in Rust	✅	❌	❌

Choose Django Bolt when:

• You need maximum API throughput (10x+ over DRF)
• You want built-in JWT/API Key auth without extra packages
• You're building API-only Django services
• You need SSE/WebSocket at scale

Choose DRF when:

• You need the mature DRF ecosystem (dry-rest-permissions, drf-writable-nested, etc.)
• Your team already knows DRF deeply
• You need browsable API for non-technical users

Choose Django Ninja when:

• You want Pydantic validation (familiar to FastAPI users)
• You need a middle ground between DRF and Bolt performance

---

Migration from Django REST Framework

Step 1: Install Django Bolt alongside DRF

pip install django-bolt

# settings.py - Keep DRF, add Bolt
INSTALLED_APPS = [
    ...
    "rest_framework",
    "django_bolt",
]

Step 2: Create Bolt API alongside DRF

# bolt_api.py - New file
from django_bolt import BoltAPI

api = BoltAPI()

Step 3: Migrate views incrementally

# Before: DRF ViewSet
class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer

# After: Django Bolt equivalent
from django_bolt.views import ModelViewSet
from django_bolt.serializers import ModelSerializer

class UserBoltSerializer(ModelSerializer):
    class Meta:
        model = User
        fields = ["id", "username", "email"]

class UserBoltViewSet(ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserBoltSerializer

api.register_viewset(UserBoltViewSet, prefix="/api/v2/users")

Step 4: Run both simultaneously

# DRF on standard Django server
python manage.py runserver 8000

# Bolt API on its own server
python manage.py runbolt 8001

Step 5: Remove DRF when fully migrated

# settings.py
INSTALLED_APPS = [
    ...
    "django_bolt",
    # "rest_framework",  # Remove when done
]

---

Production Deployment

systemd Service

# /etc/systemd/system/django-bolt.service
[Unit]
Description=Django Bolt API
After=network.target

[Service]
Type=simple
User=www-data
Group=www-data
WorkingDirectory=/opt/myapp
ExecStart=/opt/myapp/venv/bin/python manage.py runbolt --workers 4
Restart=always
RestartSec=5
Environment=DJANGO_SETTINGS_MODULE=myapp.settings

[Install]
WantedBy=multi-user.target

Docker

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

EXPOSE 8000
CMD ["python", "manage.py", "runbolt", "--workers", "4"]

Nginx Reverse Proxy

upstream bolt_api {
    server 127.0.0.1:8000;
}

server {
    listen 80;
    server_name api.example.com;

    location / {
        proxy_pass http://bolt_api;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

---

Why So Fast?

• Actix Web: Rust HTTP framework (one of the fastest)
• matchit: Zero-copy path matching
• msgspec: 5-10x faster JSON serialization
• PyO3: Direct Rust-Python interop
• No GIL for auth: JWT/API Key validation in Rust

---

Error Handling

Django-Bolt provides a structured exception hierarchy for HTTP errors and automatic error response formatting.

HTTPException and Specialized Exceptions

The base class for all HTTP errors:

from django_bolt.exceptions import HTTPException, NotFound, BadRequest, Unauthorized

# Basic usage
raise HTTPException(status_code=400, detail="Bad request")

# Specialized exceptions (pre-configured)
raise NotFound(detail="User not found")
raise BadRequest(detail="Invalid input")
raise Unauthorized(detail="Authentication required")
raise Forbidden(detail="Access denied")
raise Conflict(detail="Resource already exists")
raise TooManyRequests(detail="Rate limit exceeded")

Custom Error Responses

Add custom headers and extra data to error responses:

from django_bolt.exceptions import Unauthorized, BadRequest

# Custom headers
raise Unauthorized(
    detail="Authentication required",
    headers={"WWW-Authenticate": "Bearer", "X-Custom-Header": "value"}
)

# Extra data for debugging
raise BadRequest(
    detail="Invalid input",
    extra={
        "field": "email",
        "value": "invalid@",
        "reason": "Invalid email format"
    }
)

Validation Errors

RequestValidationError provides structured validation error responses:

from django_bolt.exceptions import RequestValidationError

# Manual validation errors
errors = [
    {"loc": ["body", "email"], "msg": "Invalid email format", "type": "value_error"},
    {"loc": ["body", "age"], "msg": "Must be positive", "type": "value_error"}
]
raise RequestValidationError(errors)

Response format for validation errors:

{
    "detail": [
        {"loc": ["body", "email"], "msg": "Invalid email format", "type": "value_error"},
        {"loc": ["body", "age"], "msg": "Must be positive", "type": "value_error"}
    ]
}

Error Handlers

Custom error handlers for different exception types:

from django_bolt.error_handlers import (
    http_exception_handler,
    request_validation_error_handler,
    generic_exception_handler,
    handle_exception
)

# Handle specific exception types
exc = NotFound(detail="User not found")
status, headers, body = http_exception_handler(exc)

# Handle validation errors
errors = [{"loc": ["body"], "msg": "Invalid", "type": "value_error"}]
exc = RequestValidationError(errors)
status, headers, body = request_validation_error_handler(exc)

# Handle unexpected exceptions (debug mode)
exc = ValueError("Something went wrong")
status, headers, body = generic_exception_handler(exc, debug=False)

# Universal handler
status, headers, body = handle_exception(some_exception)

Debug Mode

In debug mode (DEBUG=True), unhandled exceptions return Django's HTML error page with full traceback:

# debug=True: Full HTML traceback page
status, headers, body = handle_exception(exc, debug=True)

# debug=False: JSON error response
status, headers, body = handle_exception(exc, debug=False)

Exception Reference

Exception	Status Code	Default Message
BadRequest	400	Bad Request
Unauthorized	401	Unauthorized
Forbidden	403	Forbidden
NotFound	404	Not Found
MethodNotAllowed	405	Method Not Allowed
Conflict	409	Conflict
UnprocessableEntity	422	Unprocessable Entity
TooManyRequests	429	Too Many Requests
InternalServerError	500	Internal Server Error
ServiceUnavailable	503	Service Unavailable

---

Request Object

Access the full request using the request parameter for comprehensive request data.

Request Properties

The request dict contains all HTTP request information:

@api.get("/info")
async def request_info(request):
    return {
        "method": request.get("method"),      # GET, POST, etc.
        "path": request.get("path"),           # Request path
        "query": request.get("query"),         # Query parameters dict
        "params": request.get("params"),       # Path parameters dict
        "headers": request.get("headers"),     # Request headers dict
        "body": request.get("body", b""),      # Raw body bytes
        "context": request.get("context"),     # Authentication context
    }

Type-Safe Request

For better IDE support, use the Request type:

from django_bolt import Request
from django_bolt.auth import JWTAuthentication, IsAuthenticated

@api.get("/profile", auth=[JWTAuthentication()], guards=[IsAuthenticated()])
async def profile(request: Request):
    # IDE knows about .user, .session, .context, etc.
    user = await request.auser()
    return {"user_id": request.user.id, "username": request.user.username}

Headers

Extract specific headers using the Header parameter:

from typing import Annotated
from django_bolt.param_functions import Header

@api.get("/auth")
async def check_auth(
    authorization: Annotated[str, Header(alias="Authorization")]
):
    return {"auth": authorization}

# Optional headers
@api.get("/optional-header")
async def optional_header(
    custom: Annotated[str | None, Header(alias="X-Custom")] = None
):
    return {"custom": custom}

# All headers
@api.get("/headers")
async def all_headers(request):
    headers = request.get("headers", {})
    return {"headers": dict(headers)}

Cookies

Extract cookie values:

from typing import Annotated
from django_bolt.param_functions import Cookie

@api.get("/session")
async def get_session(
    session_id: Annotated[str, Cookie(alias="sessionid")]
):
    return {"session_id": session_id}

Sessions

Access Django sessions when using Django middleware:

from django_bolt import BoltAPI, Request
from django.contrib.auth import alogin, alogout
from datetime import datetime

api = BoltAPI(django_middleware=True)

@api.post("/login")
async def login(request: Request, username: str, password: str):
    user = await User.objects.filter(username=username).afirst()
    if user and user.check_password(password):
        await alogin(request, user)
        await request.session.aset("login_time", str(datetime.now()))
        return {"status": "ok"}
    return {"status": "error"}

@api.get("/profile")
async def profile(request: Request):
    user = await request.auser()
    if not user.is_authenticated:
        return {"error": "not logged in"}
    return {
        "username": user.username,
        "login_time": await request.session.aget("login_time"),
    }

@api.post("/logout")
async def logout(request: Request):
    await alogout(request)
    return {"status": "logged out"}

Session Async Methods

Method	Description
await session.aget(key, default)	Get a session value
await session.aset(key, value)	Set a session value
await session.apop(key, default)	Remove and return a value
await session.akeys()	Get all session keys
await session.aitems()	Get all key-value pairs
await session.aflush()	Delete session and create new

---

Dependency Injection

Django-Bolt provides dependency injection using the Depends marker for reusable parameter extractors.

Basic Usage

from django_bolt import BoltAPI, Depends

api = BoltAPI()

async def get_pagination(page: int = 1, limit: int = 20):
    return {"page": page, "limit": limit, "offset": (page - 1) * limit}

@api.get("/items")
async def list_items(pagination=Depends(get_pagination)):
    return {"pagination": pagination}

Request Access in Dependencies

Dependencies receive the request dict:

async def get_current_user(request):
    """Dependency that extracts the current user."""
    user_id = request.get("context", {}).get("user_id")
    if not user_id:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return await User.objects.aget(id=user_id)

@api.get("/profile")
async def get_profile(user=Depends(get_current_user)):
    return {"id": user.id, "username": user.username}

Authentication Dependency

from django_bolt.auth import get_current_user

@api.get("/me")
async def me(user=Depends(get_current_user)):
    return {
        "id": user.id,
        "username": user.username,
        "email": user.email
    }

Dependency Caching

By default, dependencies are cached per-request:

call_count = 0

async def expensive_operation(request):
    global call_count
    call_count += 1
    return {"count": call_count}

@api.get("/test")
async def test(
    dep1=Depends(expensive_operation),
    dep2=Depends(expensive_operation)  # Same dependency
):
    # expensive_operation is called ONCE, result is reused
    return {"dep1": dep1, "dep2": dep2}

# Disable caching
@api.get("/fresh")
async def fresh(dep=Depends(some_dependency, use_cache=False)):
    return dep

Nested Dependencies

Dependencies can depend on other dependencies:

async def get_settings(request):
    return await Settings.objects.afirst()

async def get_feature_flags(settings=Depends(get_settings)):
    return {
        "new_ui": settings.enable_new_ui,
        "beta": settings.beta_features,
    }

@api.get("/features")
async def features(flags=Depends(get_feature_flags)):
    return flags

Class-Based Dependencies

Classes can be used as dependencies:

class DatabaseSession:
    def __init__(self, request):
        self.request = request
        self.connection = None

    async def __aenter__(self):
        self.connection = await get_connection()
        return self.connection

    async def __aexit__(self, *args):
        if self.connection:
            await self.connection.close()

@api.get("/data")
async def get_data(db=Depends(DatabaseSession)):
    async with db:
        # Use database connection
        pass

---

File Uploads

Django-Bolt provides the UploadFile class for handling file uploads with Django integration.

Basic File Upload

from typing import Annotated
from django_bolt import UploadFile
from django_bolt.params import File

@api.post("/upload")
async def upload(file: Annotated[UploadFile, File()]):
    content = await file.read()
    return {
        "filename": file.filename,
        "size": file.size,
        "content_type": file.content_type,
    }

UploadFile Properties

Property	Type	Description
filename	str	Original filename
content_type	str	MIME type
size	int	Size in bytes
file	Django File	Django File object for FileField
headers	dict	Multipart headers

File Validation

from django_bolt import FileSize

@api.post("/upload")
async def upload(
    file: Annotated[UploadFile, File(
        max_size=FileSize.MB_10,           # Maximum 10MB
        min_size=1024,                    # Minimum 1KB
        allowed_types=["image/*", "application/pdf"],  # MIME types
    )]
):
    return {"filename": file.filename}

FileSize Enum

from django_bolt import FileSize

File(max_size=FileSize.KB_1)    # 1 KB
File(max_size=FileSize.MB_1)    # 1 MB
File(max_size=FileSize.MB_5)    # 5 MB
File(max_size=FileSize.MB_10)   # 10 MB
File(max_size=FileSize.MB_50)   # 50 MB

Multiple File Uploads

@api.post("/upload-multiple")
async def upload_multiple(
    files: Annotated[list[UploadFile], File(
        max_files=5,              # Maximum 5 files
        max_size=FileSize.MB_5,  # 5MB per file
    )]
):
    return {
        "count": len(files),
        "filenames": [f.filename for f in files],
    }

Saving to Django FileField/ImageField

from myapp.models import Document, UserProfile

# FileField
@api.post("/documents")
async def create_document(
    title: Annotated[str, Form()],
    upload: Annotated[UploadFile, File(max_size=FileSize.MB_10)],
):
    doc = Document(title=title)
    doc.file = upload.file  # Assign Django File to FileField
    await doc.asave()
    return {"id": doc.id, "url": doc.file.url}

# ImageField
@api.post("/avatar")
async def upload_avatar(
    avatar: Annotated[UploadFile, File(
        max_size=FileSize.MB_5,
        allowed_types=["image/*"],
    )],
    request,
):
    profile = await UserProfile.objects.aget(user=request.user)
    profile.avatar = avatar.file  # Assign Django File to ImageField
    await profile.asave()
    return {"avatar_url": profile.avatar.url}

Global Upload Settings

# settings.py
from django_bolt import FileSize

# Maximum upload size (requests exceeding this are rejected)
BOLT_MAX_UPLOAD_SIZE = FileSize.MB_10  # 10 MB global limit

# Memory threshold before spooling to disk (default: 1 MB)
BOLT_MEMORY_SPOOL_THRESHOLD = 5 * 1024 * 1024  # 5 MB

---

Pagination

Django-Bolt provides three pagination styles for handling large datasets efficiently.

PageNumber Pagination

Classic page-based pagination:

from django_bolt import BoltAPI, PageNumberPagination, paginate

api = BoltAPI()

class ArticlePagination(PageNumberPagination):
    page_size = 20
    max_page_size = 100
    page_size_query_param = "page_size"  # Allow client to customize

@api.get("/articles")
@paginate(ArticlePagination)
async def list_articles(request) -> list[ArticleSerializer]:
    return Article.objects.all()

Response:

{
    "count": 150,
    "page": 1,
    "page_size": 20,
    "total_pages": 8,
    "has_next": true,
    "has_previous": false,
    "next_page": 2,
    "previous_page": null,
    "items": [...]
}

LimitOffset Pagination

Flexible offset-based pagination:

from django_bolt import LimitOffsetPagination, paginate

@api.get("/articles", response_model=list[ArticleSerializer])
@paginate(LimitOffsetPagination)
async def list_articles(request):
    return Article.objects.all()

Query: /articles?limit=10&offset=20

Cursor Pagination

Efficient pagination for large datasets and real-time feeds:

from django_bolt import CursorPagination, paginate

class ArticlePagination(CursorPagination):
    page_size = 20
    ordering = "-created_at"  # Required: field to paginate by

@api.get("/articles")
@paginate(ArticlePagination)
async def list_articles(request) -> list[ArticleSerializer]:
    return Article.objects.all()

Query: /articles?cursor=eyJ2IjoxMDB9

ViewSet with Pagination

from django_bolt.views import ViewSet

@api.viewset("/articles")
class ArticleViewSet(ViewSet):
    queryset = Article.objects.all()

    @paginate(ArticlePagination)
    async def list(self, request) -> list[ArticleSerializer]:
        return await self.get_queryset()

ModelViewSet with Pagination

from django_bolt.views import ModelViewSet

@api.viewset("/articles")
class ArticleViewSet(ModelViewSet):
    queryset = Article.objects.all()
    serializer_class = ArticleDetailSerializer
    list_serializer_class = ArticleListSerializer
    pagination_class = ArticlePagination  # Automatically applied to list()

---

WebSockets

Django-Bolt provides WebSocket support for real-time bidirectional communication.

Basic WebSocket Endpoint

from django_bolt import BoltAPI, WebSocket

api = BoltAPI()

@api.websocket("/ws/echo")
async def echo(websocket: WebSocket):
    await websocket.accept()
    async for message in websocket.iter_text():
        await websocket.send_text(f"Echo: {message}")

Sending Messages

# Text messages
await websocket.send_text("Hello, World!")

# Binary messages
await websocket.send_bytes(b"\x00\x01\x02\x03")

# JSON messages
await websocket.send_json({"type": "message", "data": "Hello"})

Receiving Messages

# Text messages
message = await websocket.receive_text()
async for message in websocket.iter_text():
    print(f"Received: {message}")

# Binary messages
data = await websocket.receive_bytes()

# JSON messages
data = await websocket.receive_json()
async for data in websocket.iter_json():
    print(f"Received: {data}")

Path and Query Parameters

# Path parameters
@api.websocket("/ws/room/{room_id}")
async def room(websocket: WebSocket, room_id: str):
    await websocket.accept()
    async for message in websocket.iter_text():
        await websocket.send_text(f"[{room_id}] {message}")

# Query parameters (for authentication)
@api.websocket("/ws/connect")
async def connect(websocket: WebSocket, token: str | None = None):
    if token != "secret":
        await websocket.close(code=4001, reason="Invalid token")
        return
    await websocket.accept()

Closing Connections

from django_bolt import WebSocketDisconnect

@api.websocket("/ws")
async def handler(websocket: WebSocket):
    await websocket.accept()
    try:
        async for message in websocket.iter_text():
            await websocket.send_text(message)
    except WebSocketDisconnect:
        print("Client disconnected")

# Close from server
await websocket.close(code=1000, reason="Normal closure")

Authentication

Apply authentication to WebSocket endpoints:

from django_bolt.auth import JWTAuthentication, IsAuthenticated

@api.websocket(
    "/ws/protected",
    auth=[JWTAuthentication()],
    guards=[IsAuthenticated()]
)
async def protected(websocket: WebSocket):
    user_id = websocket.context.get("user_id")
    await websocket.accept()
    await websocket.send_text(f"Welcome, user {user_id}")

Real-Time Patterns

Broadcast to All Clients

connected_clients = set()

@api.websocket("/ws/broadcast")
async def broadcast(websocket: WebSocket):
    await websocket.accept()
    connected_clients.add(websocket)

    try:
        async for message in websocket.iter_text():
            for client in connected_clients:
                await client.send_text(message)
    finally:
        connected_clients.discard(websocket)

Room-Based Chat

rooms = {}  # room_id -> set of websockets

@api.websocket("/ws/room/{room_id}")
async def room(websocket: WebSocket, room_id: str):
    await websocket.accept()

    if room_id not in rooms:
        rooms[room_id] = set()
    rooms[room_id].add(websocket)

    try:
        async for message in websocket.iter_text():
            for client in rooms[room_id]:
                await client.send_text(f"[{room_id}] {message}")
    finally:
        rooms[room_id].discard(websocket)

---

Background Tasks

Django-Bolt integrates with Django's task framework for asynchronous background processing.

Integration with Celery

# tasks.py
from celery import shared_task

@shared_task
def process_data_async(data_id: int):
    """Process data in background."""
    import asyncio
    from myapp.models import DataRecord

    async def process():
        record = await DataRecord.objects.aget(id=data_id)
        # Expensive processing
        record.status = "processed"
        await record.asave()

    asyncio.run(process())

# API handler
from myapp.tasks import process_data_async

@api.post("/process")
async def start_processing(data_id: int):
    process_data_async.delay(data_id)
    return {"status": "processing_started", "data_id": data_id}

Django Q

# Alternative: Django Q
@api.post("/process")
async def start_processing(data_id: int):
    from django_q.tasks import async_task

    async_task("myapp.functions.process_data", data_id)
    return {"status": "queued", "data_id": data_id}

Using Channels for Real-Time Updates

# For WebSocket + background task integration
@api.post("/long-task")
async def start_long_task(request: Request):
    task_id = str(uuid.uuid4())

    # Start background task
    asyncio.create_task(background_processing(task_id))

    return {"task_id": task_id, "status": "started"}

async def background_processing(task_id: str):
    # Simulate long task
    await asyncio.sleep(10)

    # Notify via WebSocket (store connected clients globally)
    if task_id in connected_tasks:
        await connected_tasks[task_id].send_json({
            "type": "task_complete",
            "task_id": task_id
        })

---

Django ORM Patterns

Django-Bolt handlers use async, requiring Django's async ORM methods for maximum performance.

Basic Async ORM Methods

from myapp.models import Article

# Get a single object
article = await Article.objects.aget(id=1)

# Create an object
article = await Article.objects.acreate(
    title="My Article",
    content="Content here"
)

# Get or create
article, created = await Article.objects.aget_or_create(
    title="My Article",
    defaults={"content": "Default content"}
)

# Count
total = await Article.objects.acount()

# Check existence
exists = await Article.objects.filter(published=True).aexists()

# Delete
deleted_count, _ = await Article.objects.filter(draft=True).adelete()

# Update
updated_count = await Article.objects.filter(draft=True).aupdate(published=True)

Avoiding N+1 Queries

Use select_related for ForeignKey and OneToOne:

# Good: 1 query with JOIN
@api.get("/articles")
async def list_articles():
    articles = []
    async for article in Article.objects.select_related("author")[:20]:
        articles.append({
            "id": article.id,
            "author_name": article.author.username  # No extra query!
        })
    return {"articles": articles}

Use prefetch_related for ManyToMany and reverse ForeignKey:

# Good: 2 queries (articles + tags)
@api.get("/articles")
async def list_articles():
    queryset = Article.objects.select_related("author").prefetch_related("tags")
    async for article in queryset[:20]:
        tags = [tag.name for tag in article.tags.all()]  # Already prefetched!
    return {"articles": [...]}

Transactions

Use sync_to_async for database transactions:

from asgiref.sync import sync_to_async
from django.db import transaction

@api.post("/transfer")
async def transfer_funds(from_id: int, to_id: int, amount: float):
    @sync_to_async
    def do_transfer():
        with transaction.atomic():
            from_account = Account.objects.select_for_update().get(id=from_id)
            to_account = Account.objects.select_for_update().get(id=to_id)
            from_account.balance -= amount
            to_account.balance += amount
            from_account.save()
            to_account.save()
        return {"success": True}

    return await do_transfer()

Aggregations

from django.db.models import Count, Avg, Q

@api.get("/stats")
async def article_stats():
    stats = await Article.objects.aaggregate(
        total=Count("id"),
        published=Count("id", filter=Q(published=True)),
        avg_comments=Avg("comment_count")
    )
    return stats

Bulk Operations

# Bulk create
articles = [
    Article(title=f"Article {i}", content="...")
    for i in range(100)
]
created = await Article.objects.abulk_create(articles)

# Bulk update
await Article.objects.filter(draft=True).aupdate(published=True)

---

Settings Configuration

Full reference for Django-Bolt settings in settings.py:

# settings.py

# Required: Add to INSTALLED_APPS
INSTALLED_APPS = [
    ...
    "django_bolt"
    ...
]

# Django-Bolt Configuration
BOLT = {
    # Server settings
    "HOST": "0.0.0.0",           # Server host
    "PORT": 8000,                # Server port
    "PROCESSES": 4,              # Number of worker processes
    "BACKLOG": 2048,             # Socket backlog size
    "KEEP_ALIVE": 30,            # Keep-alive timeout in seconds

    # Debug mode
    "DEBUG": False,

    # Enable signals (may impact performance)
    "EMIT_SIGNALS": False,
}

# JWT Configuration
JWT_SECRET_KEY = "your-secret-key"
JWT_ALGORITHM = "HS256"
JWT_EXPIRATION = 3600  # seconds

# File Upload Settings
from django_bolt import FileSize
BOLT_MAX_UPLOAD_SIZE = FileSize.MB_50  # 50 MB max
BOLT_MEMORY_SPOOL_THRESHOLD = 5 * 1024 * 1024  # 5 MB

# Compression Configuration
from django_bolt.middleware import CompressionConfig

BOLT_COMPRESSION = CompressionConfig(
    backend="gzip",
    minimum_size=500,  # Only compress responses > 500 bytes
)

# CORS Configuration
BOLT_CORS = {
    "allow_origins": ["https://example.com"],
    "allow_methods": ["GET", "POST", "PUT", "DELETE"],
    "allow_headers": ["*"],
    "allow_credentials": True,
}

Command-Line Options

# Run development server
python manage.py runbolt --dev

# Run production server with multiple processes
python manage.py runbolt --host 0.0.0.0 --port 8000 --processes 4

# Increase socket backlog for high traffic
python manage.py runbolt --processes 4 --backlog 2048

# Adjust keep-alive timeout
python manage.py runbolt --processes 4 --keep-alive 30

---

Deployment

Production deployment guide for Django-Bolt.

Running as a Service

With systemd

Create /etc/systemd/system/django-bolt.service:

[Unit]
Description=Django-Bolt API Server
After=network.target

[Service]
User=www-data
Group=www-data
WorkingDirectory=/path/to/your/project
ExecStart=/path/to/venv/bin/python manage.py runbolt --host 127.0.0.1 --port 8000 --processes 4
Restart=always
RestartSec=3

[Install]
WantedBy=multi-user.target

Enable and start:

sudo systemctl daemon-reload
sudo systemctl enable django-bolt
sudo systemctl start django-bolt
sudo systemctl status django-bolt

With supervisor

Create /etc/supervisor/conf.d/django-bolt.conf:

[program:django-bolt]
command=/path/to/venv/bin/python manage.py runbolt --host 127.0.0.1 --port 8000 --processes 4
directory=/path/to/your/project
user=www-data
autostart=true
autorestart=true
redirect_stderr=true
stdout_logfile=/var/log/django-bolt.log

sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start django-bolt

Reverse Proxy with nginx

upstream django_bolt {
    server 127.0.0.1:8000;
}

server {
    listen 80;
    server_name api.example.com;

    location / {
        proxy_pass http://django_bolt;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

Database Connections

psycopg pool (recommended for Django 5.1+)

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": "mydb",
        "USER": "myuser",
        "PASSWORD": "mypassword",
        "HOST": "localhost",
        "CONN_MAX_AGE": 0,
        "OPTIONS": {
            "pool": {
                "min_size": 2,
                "max_size": 10,
            }
        },
    }
}

PgBouncer (external pooler)

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": "mydb",
        "HOST": "127.0.0.1",
        "PORT": "6432",  # PgBouncer port
        "CONN_MAX_AGE": 0,
        "DISABLE_SERVER_SIDE_CURSORS": True,
    }
}

Docker Deployment

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

EXPOSE 8000

CMD ["python", "manage.py", "runbolt", "--host", "0.0.0.0", "--port", "8000", "--processes", "4"]

# docker-compose.yml
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgres://user:pass@db:5432/mydb
    depends_on:
      - db

  db:
    image: postgres:15
    environment:
      POSTGRES_DB: mydb
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass

Workers vs Processes

Django-Bolt uses processes for parallelism, not workers. Each process has its own GIL:

# Use processes for parallelism (not workers)
python manage.py runbolt --processes 4

# Rule of thumb: set --processes to number of CPU cores

---

Comparison with DRF/Django Ninja

When to Choose Django Bolt

Feature	Django Bolt	Django REST Framework	Django Ninja
Performance	~188k RPS	~10-15k RPS	~50-70k RPS
Python GIL	Bypassed via processes	Blocked	Blocked
Django ORM	Full async	Sync only	Full async
Type Safety	Full (msgspec)	Partial (serializers)	Full (Pydantic)
Django Admin	Compatible	Compatible	Compatible
Django Packages	Most work	All work	Most work
Learning Curve	Low	Low	Medium
WebSocket	Built-in	Via channels	Via channels

Choose Django Bolt When:

• You need maximum performance (60k+ RPS)
• You want to keep using Django ORM without async wrappers
• You're building new APIs and performance matters
• You want to bypass Python's GIL limitations
• You prefer msgspec over Pydantic for validation
• You want to incrementally migrate from DRF

Choose Django REST Framework When:

• You have an existing DRF codebase
• You need extensive third-party packages
• You're new to async programming
• You need Django REST Framework's browserable API

Choose Django Ninja When:

• You prefer Pydantic for validation
• You're building new APIs from scratch
• You need a more Pythonic async experience
• You want better IDE support via Pydantic

---

Migration from DRF

Step-by-step guide to migrate Django REST Framework views to Django Bolt.

Step 1: Install Django Bolt

pip install django-bolt

Add to INSTALLED_APPS:

INSTALLED_APPS = [
    ...
    "django_bolt"
    ...
]

Step 2: Convert Function-Based Views

Before (DRF):

# views.py (DRF)
from rest_framework.decorators import api_view
from rest_framework.response import Response

@api_view(["GET", "POST"])
def user_list(request):
    if request.method == "GET":
        users = User.objects.all()
        serializer = UserSerializer(users, many=True)
        return Response(serializer.data)

    serializer = UserSerializer(data=request.data)
    if serializer.is_valid():
        serializer.save()
        return Response(serializer.data, status=201)
    return Response(serializer.errors, status=400)

After (Django Bolt):

# api.py (Django Bolt)
from django_bolt import BoltAPI
import msgspec

api = BoltAPI()

class UserSchema(msgspec.Struct):
    id: int
    username: str
    email: str

@api.get("/users")
async def list_users():
    users = await User.objects.all()
    return [{"id": u.id, "username": u.username, "email": u.email} for u in users]

@api.post("/users")
async def create_user(data: UserSchema):
    user = await User.objects.acreate(
        username=data.username,
        email=data.email
    )
    return {"id": user.id, "username": user.username}, 201

Step 3: Convert Class-Based Views

Before (DRF):

# views.py (DRF)
from rest_framework import viewsets
from rest_framework.permissions import IsAuthenticated
from .serializers import UserSerializer

class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer
    permission_classes = [IsAuthenticated]

After (Django Bolt):

# api.py (Django Bolt)
from django_bolt import BoltAPI, ViewSet
from django_bolt.auth import JWTAuthentication, IsAuthenticated
from django.contrib.auth import get_user_model

User = get_user_model()
api = BoltAPI()

@api.viewset("/users", auth=[JWTAuthentication()], guards=[IsAuthenticated()])
class UserViewSet(ViewSet):
    async def list(self, request):
        users = await User.objects.all()
        return [{"id": u.id, "username": u.username} for u in users]

    async def retrieve(self, request, pk: int):
        user = await User.objects.aget(id=pk)
        return {"id": user.id, "username": user.username}

    async def create(self, request):
        data = await request.json()
        user = await User.objects.acreate(**data)
        return {"id": user.id}, 201

    async def destroy(self, request, pk: int):
        await User.objects.filter(id=pk).adelete()
        return {"deleted": True}

Step 4: Convert Serializers

Before (DRF):

# serializers.py (DRF)
from rest_framework import serializers
from .models import Article

class ArticleSerializer(serializers.ModelSerializer):
    author_name = serializers.CharField(source="author.username", read_only=True)

    class Meta:
        model = Article
        fields = ["id", "title", "content", "author", "author_name", "created_at"]

After (Django Bolt):

# schemas.py (Django Bolt)
import msgspec

class AuthorSchema(msgspec.Struct):
    id: int
    username: str

class ArticleSchema(msgspec.Struct):
    id: int
    title: str
    content: str
    author: AuthorSchema
    created_at: str

Step 5: Convert URLs

Before (DRF):

# urls.py (DRF)
from rest_framework.routers import DefaultRouter
from .views import UserViewSet

router = DefaultRouter()
router.register(r"users", UserViewSet)
urlpatterns = router.urls

After (Django Bolt):

# myproject/urls.py
from django.urls import path
from myapp.api import api

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

Step 6: Running the Server

Before (DRF):

# Development
python manage.py runserver

# Production (with gunicorn)
gunicorn myproject.wsgi --workers 4

After (Django Bolt):

# Development
python manage.py runbolt --dev

# Production (standalone, no gunicorn needed)
python manage.py runbolt --processes 4

Migration Checklist

• Install django-bolt and add to INSTALLED_APPS
• Convert function-based views to Django Bolt handlers
• Convert class-based views to ViewSets
• Replace DRF serializers with msgspec structs
• Update URL configuration
• Test authentication (JWT, API key)
• Test file uploads if applicable
• Run load tests to verify performance
• Remove DRF dependencies when ready

---

Best Practices

API Design

# ✅ GOOD: Use decorators
from django_bolt.http import HttpRequest, JsonResponse
from django_bolt.utils.decorators import json_response

@json_response()
def get_users(request: HttpRequest) -> JsonResponse:
    return {"users": []}

# ✅ GOOD: Serializer validation
from django_bolt.serializers import Serializer

class UserSerializer(Serializer):
    name: str
    email: str
    
    def validate_email(self, value):
        if not value.endswith('@company.com'):
            raise ValidationError("Must use company email")
        return value

Error Handling

from django_bolt.errors import ApiError

# ✅ GOOD: Raise proper errors
def get_user(request, user_id):
    user = User.objects.get_or_404(user_id)
    if not user.is_active:
        raise ApiError(400, "User not active")
    return user

Performance

# ✅ GOOD: Use built-in caching
from django_bolt.cache import cache_page

@cache_page(60 * 15)
def get_data(request):
    ...

# ✅ GOOD: Query optimization
users = User.objects.select_related('profile').prefetch_related('posts')

Do:

• Use decorators for common patterns
• Leverage built-in serializers
• Enable caching for static data

Don't:

• Mix Django ORM with Bolt models
• Skip error handling

---

References

• GitHub: https://github.com/dj-bolt/django-bolt
• Documentation: https://bolt.farhana.li
• Performance Comparison: Faster than FastAPI, similar to Go/Node.js performance