Django Bolt

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 300k+ RPS. Uses Actix Web for HTTP, PyO3 for Python bridging, and msgspec for serialization.

Overview

State: django-bolt v0.11.0, Python 3.12–3.14, Django 4.2–6.0

Note: This skill documents django-bolt v0.11.0. Pin your dependency: pip install "django-bolt>=0.11.0,<0.12.0".

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?

• Native Rust server — no gunicorn/uvicorn needed
• 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}

@api.query QUERY Method (v0.9.0+)

from django_bolt import api_query

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

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 (v0.9.1+)

from django_bolt.auth import JWTAuthentication

# Cookie-based JWT (default, CSRF protection enabled)
api = BoltAPI(auth=[JWTAuthentication(cookie=True)])

# Bearer token (stateless, no CSRF)
api = BoltAPI(auth=[JWTAuthentication(cookie=False)])

Asymmetric JWT (v0.9.1–v0.10.0)

Supports PS256/PS384/PS512 and EdDSA algorithms with JWKS:

from django_bolt.auth import JWTAuthentication

# JWKS endpoint for public key retrieval
api = BoltAPI(
    auth=[JWTAuthentication(
        jwks_url="https://your-auth-server.com/.well-known/jwks.json",
        algorithm="RS256"  # or PS256, PS384, PS512, EdDSA
    )]
)

Works with Clerk, Auth0, Okta, and any OAuth 2.1 provider.

Access/Refresh Token Pairs (v0.10.0+)

from django_bolt.auth import JWTAuthentication

auth = JWTAuthentication(
    access_lifetime=900,      # 15 minutes
    refresh_lifetime=604800,  # 7 days
    rotate_refresh=True,      # Rotate on each use
    detect_reuse=True,        # Flag token reuse (security alert)
    revoke_all_on_reuse=True  # Bulk revoke on detected reuse
)

Access tokens include jti (JWT ID) for tracking and revocation.

CSRF Protection (v0.10.0, Breaking Change)

When cookie=True (default), CSRF check is ON. Non-browser clients receive 403 unless csrf=False:

# Browser clients (CSRF protected)
JWTAuthentication(cookie=True)  # default

# API clients (no CSRF)
JWTAuthentication(cookie=False)

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 "):
            user = await self.get_user(token)
            return AuthResult(user=user)
        return AuthResult()

    async def get_user(self, token: str):
        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:
        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,
    ]
)

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"

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

EventSourceResponse (v0.7.5+)

Server-Sent Events with automatic reconnection support:

from django_bolt.response import EventSourceResponse

async def news_feed():
    while True:
        yield {"event": "update", "data": {"news": "breaking"}}
        await asyncio.sleep(5)

@api.get("/news")
async def news_handler(request):
    return EventSourceResponse(news_feed())

Per-Chunk Compression (v0.8.1+)

Automatic compression per chunk (br/gzip/zstd):

@api.get("/large-stream")
async def large_stream(request):
    async def generate():
        for i in range(1000):
            yield f"data: {i}\n\n"
    return StreamingResponse(generate(), compress=True)

Union Return Types (v0.8.1+)

@api.get("/conditional")
async def conditional() -> dict[str, str] | list[int]:
    if some_condition:
        return {"status": "ok"}
    return [1, 2, 3]

HTTPException with Body (v0.11.0+)

from django_bolt.exceptions import HTTPException

raise HTTPException(
    status_code=400,
    body={"error": "validation_failed", "fields": {"email": "Invalid"}}
)

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 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):
    user = await User.objects.acreate(
        username=body.username,
        email=body.email,
        is_active=body.is_active
    )
    return {"id": user.id}

Loading Plans (v0.10.2+)

Auto select_related/prefetch_related/annotate from Serializer fields:

from django_bolt.serializers import ModelSerializer, loading_plan

class AuthorSerializer(ModelSerializer):
    class Meta:
        model = Author
        fields = ["id", "name"]

class ArticleSerializer(ModelSerializer):
    author: AuthorSchema  # Auto select_related("author")
    
    class Meta:
        model = Article
        fields = ["id", "title", "author"]
@api.get("/articles")
@loading_plan(ArticleSerializer)  # Auto-optimizes query
async def list_articles():
    return Article.objects.all()

from_models() / afrom_models() (v0.10.2+)

from django_bolt.serializers import from_models, afrom_models

class ArticleSchema(from_models(Article)):
    author_name: str  # Derived field

@api.get("/articles")
async def list_articles():
    articles = await Article.objects.all().alist()
    return [afrom_models(ArticleSchema, a) for a in articles]

articles = await Article.objects.all().alist()
return [afrom_models(ArticleSchema, a) for a in articles]

### Nested() Removed in v0.11.0

Use plain type hints instead:

```python
# Before v0.11.0 (removed)
# author: Nested(AuthorSerializer)

# v0.11.0+ (plain type hint)
class ArticleSerializer(ModelSerializer):
    author: AuthorSerializer  # Direct type hint

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 3.1 documentation with strict mode support.

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",
    )
)

Include/Exclude from Schema (v0.10.2+/0.11.0+)

# Exclude endpoint from OpenAPI
@api.get("/internal", include_in_schema=False)
async def internal_handler():
    return {"secret": True}

# Layered inclusion (v0.11.0+)
@api.get("/admin", include_in_schema={"admin": True})
async def admin_handler():
    return {"admin": True}

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": "test@example.com"}
        )
        self.assertEqual(response.status_code, 201)
        data = await response.json()
        self.assertEqual(data["username"], "testuser")

    async def test_get_user(self):
        user = await User.objects.acreate(username="testuser", email="test@example.com")
        
        response = await self.get(f"/api/users/{user.id}")
        self.assertEqual(response.status_code, 200)

Note: TestClient shares DB connection with handler threads for accurate concurrency testing.

Performance Benchmarks

Conditions: 8 processes, C=100, loopback, AMD Ryzen 5 5600G

Source: https://bolt.farhana.li/benchmarks/

Configuration

Settings

# settings.py

# 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,
}

Global Auth/Permission Classes (v0.6.0+)

# settings.py
BOLT_AUTHENTICATION_CLASSES = [
    "django_bolt.auth.JWTAuthentication",
]

BOLT_PERMISSION_CLASSES = [
    "django_bolt.auth.IsAuthenticated",
]

Production Server: runbolt

Worker Recycling (v0.11.0+)

# Recycle workers after memory threshold
python manage.py runbolt --max-rss 512000  # 512 MB

# Limit worker lifetime
python manage.py runbolt --workers-lifetime 3600  # 1 hour

# Auto-respawn failed workers
python manage.py runbolt --respawn-failed-workers

Graceful Shutdown

Django Bolt handles SIGTERM/SIGINT gracefully, draining active WebSocket connections with code 1012 (Service Restart).

Development Mode (v0.11.0+)

# Native Rust reloader (watches project root + import graph)
python manage.py runbolt --dev

# Custom reload directory
python manage.py runbolt --dev --reload-dir src/

# Skip startup checks
python manage.py runbolt --dev --skip-checks

Startup system checks run by default; warnings shown for unapplied migrations.

Deployment

systemd Service

# /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

supervisor

# /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

MCP Servers (v0.11.0+)

Django Bolt includes built-in MCP (Model Context Protocol) server support.

Installation

pip install "django-bolt[mcp]"

Mount MCP

from django_bolt.mcp import mount_mcp

api = BoltAPI()
mount_mcp(api)  # Mounts MCP endpoints at /mcp

• SSE-based transport for streaming
• Resource templates with variable substitution

Example: Custom MCP Tool

from django_bolt.mcp import mount_mcp, mcp_tool
from django_bolt import BoltAPI

api = BoltAPI()

@mcp_tool(description="Get user by ID")
async def get_user(user_id: int) -> dict:
    user = await User.objects.aget(id=user_id)
    return {"id": user.id, "username": user.username, "email": user.email}

mount_mcp(api, tools=[get_user])

Example: MCP Resource

from django_bolt.mcp import mcp_resource

@mcp_resource(uri="user://{user_id}", description="User profile")
async def user_profile(user_id: int) -> str:
    user = await User.objects.aget(id=user_id)
    return f"User: {user.username} ({user.email})"

mount_mcp(api, resources=[user_profile])

See https://bolt.farhana.li/topics/mcp/ for full documentation.

Rate Limiting

Per-Identity Keys (v0.10.3+)

from django_bolt import rate_limit

@api.get("/user-data", guards=[rate_limit(key="user")])
async def user_data(request):
    return {"data": "per-user rate limit"}

@api.get("/public", guards=[rate_limit(key="api_key")])
async def public_data(request):
    return {"data": "per-api-key rate limit"}

BOLT_TRUSTED_PROXIES (v0.10.3+, Breaking Change)

When behind a proxy, configure BOLT_TRUSTED_PROXIES or all callers share one bucket:

# settings.py
BOLT_TRUSTED_PROXIES = [
    "10.0.0.0/8",      # Internal network
    "172.16.0.0/12",
    "192.168.0.0/16",
]

Error Handling

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

HTTPException and Specialized Exceptions

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

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

from django_bolt.exceptions import RequestValidationError

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:

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

Error Handlers

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.

Exception Reference

Pagination

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

PageNumber 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"

@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

from django_bolt import LimitOffsetPagination, paginate

@api.get("/articles")
@paginate(LimitOffsetPagination)
async def list_articles(request):
    return Article.objects.all()

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

Cursor Pagination

from django_bolt import CursorPagination, paginate

class ArticlePagination(CursorPagination):
    page_size = 20
    ordering = "-created_at"

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

Query: /articles?cursor=eyJ2IjoxMDB9

Manual Pagination

from django_bolt import 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,
    }

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()

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

from django_bolt.auth import JWTAuthentication, IsAuthenticated

@api.websocket(
    "/ws/protected",
    auth=[JWTAuthentication()],
    guards=[IsAuthenticated()]
)
async def protected_ws(websocket: WebSocket):
    await websocket.accept()
    user = websocket.user
    async for message in websocket.iter_text():
        await websocket.send_json({"user": user.id, "message": message})

Graceful Shutdown

On shutdown, connections close with code 1012 (Service Restart).

Request Object

Access the full request using the request parameter.

Request Properties

@api.get("/info")
async def request_info(request):
    return {
        "method": request.get("method"),
        "path": request.get("path"),
        "query": request.get("query"),
        "params": request.get("params"),
        "headers": request.get("headers"),
        "body": request.get("body", b""),
        "context": request.get("context"),
    }

Type-Safe Request

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

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

Headers

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}

Cookies

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

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

Dependency Injection

Django-Bolt provides dependency injection using the Depends marker.

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

async def get_current_user(request):
    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

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)
):
    # 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

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

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:
        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

File Validation

from django_bolt import FileSize

@api.post("/upload")
async def upload(
    file: Annotated[UploadFile, File(
        max_size=FileSize.MB_10,
        min_size=1024,
        allowed_types=["image/*", "application/pdf"],
    )]
):
    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,
        max_size=FileSize.MB_5,
    )]
):
    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
    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
    await profile.asave()
    return {"avatar_url": profile.avatar.url}

Global Upload Settings

# settings.py
from django_bolt import FileSize

BOLT_MAX_UPLOAD_SIZE = FileSize.MB_10
BOLT_MEMORY_SPOOL_THRESHOLD = 5 * 1024 * 1024

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 {
        "

…(truncated)