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 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": "test@example.com"}
        )
        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="test@example.com")
        
        response = await self.get(f"/api/users/{user.id}")
        self.assertEqual(response.status_code, 200)

Performance Benchmarks

Standard Endpoints

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

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

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

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

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

…(truncated)