django-bolt · diff
v2.0.0 to v3.0.0
86 added, 1954 removed. Audit A to A.
---
name: django-bolt
- description: "Django Bolt - Rust-powered high-performance API framework, 300k+ RPS, decorator routing, built-in auth, async ORM"
+ description: Use when building high-performance APIs with django-bolt - routing, JWT auth, msgspec serializers, WebSockets, SSE streaming, file uploads, ORM patterns, runbolt deployment, or migrating from DRF
metadata:
- author: mte90
- version: 2.0.0
- tags:
- - python
- - django
- - bolt
- - api
- - rust
- - performance
- - async
- - high-performance
+ author: MT
+ version: 3.0.0
+ tags: python, django, bolt, api, rust, performance, async
---
# 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.
+ Django Bolt is a high-performance Django framework that bypasses the Python GIL using Rust, delivering **300k+ RPS** for simple endpoints.
## Overview
- **State:** django-bolt v0.11.0, Python 3.12–3.14, Django 4.2–6.0
+ Django Bolt combines Django's ORM and ecosystem with Rust's performance. Built on top of:
- **Note:** This skill documents django-bolt v0.11.0. Pin your dependency: `pip install "django-bolt>=0.11.0,<0.12.0"`.
+ - **Hyper** (Rust HTTP server)
+ - **Tokio** (async runtime)
+ - **msgspec** (5-10x faster than Pydantic)
- **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
+ Key features:
- **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
+ - **Async ORM** - Native async/await support for Django ORM
+ - **Built-in WebSocket** - Real-time communication without channels
+ - **SSE Streaming** - Server-Sent Events for live updates
+ - **JWT Authentication** - Built-in auth with refresh token rotation
+ - **msgspec Serializers** - Fast schema validation and serialization
+ - **OpenAPI Docs** - Auto-generated at `/docs`
---
## Installation
```bash
pip install django-bolt
```
+ Add to `INSTALLED_APPS`:
+
```python
- # settings.py
INSTALLED_APPS = [
...
- "django_bolt"
- ...
+ "django_bolt",
]
```
---
## Quick Start
- ### Basic API
-
```python
- # api.py
+ # myapp/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("/")
+ async def hello():
+ return {"message": "Hello, World!"}
- @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}
+ @api.get("/users")
+ async def list_users():
+ users = await User.objects.all()
+ return [{"id": u.id, "username": u.username} for u in users]
```
- ### Running the Server
+ Mount in `urls.py`:
+ ```python
+ # myproject/urls.py
+ from django.urls import path
+ from myapp.api import api
+
+ urlpatterns = [
+ path("api/", api.urls),
+ ]
+ ```
+
+ Run:
+
```bash
- # Development
python manage.py runbolt --dev
-
- # Production (standalone)
- python manage.py runbolt
```
---
## Routing
- ### HTTP Methods
+ ### Basic Routes
```python
- 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"}
+ @api.get("/users")
+ @api.post("/users")
+ @api.put("/users/{pk}")
+ @api.delete("/users/{pk}")
+ @api.patch("/users/{pk}")
```
### Path Parameters
```python
@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}
+ user = await User.objects.aget(id=user_id)
+ return {"id": user.id, "username": user.username}
```
### Query Parameters
```python
- 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+)
-
- ```python
- 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}
+ @api.get("/users")
+ async def list_users(page: int = 1, limit: int = 10):
+ users = await User.objects.all()[limit*(page-1):limit*page]
+ return [{"id": u.id} for u in users]
```
### Request Body
```python
import msgspec
- class CreateUserRequest(msgspec.Struct):
+ class UserCreate(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+)
-
- ```python
- 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:
-
- ```python
- 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+)
-
- ```python
- 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`:
-
- ```python
- # Browser clients (CSRF protected)
- JWTAuthentication(cookie=True) # default
-
- # API clients (no CSRF)
- JWTAuthentication(cookie=False)
- ```
-
- ### API Key Authentication
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- from django.middleware.security import SecurityMiddleware
-
- api = BoltAPI(
- django_middleware=[
- SecurityMiddleware,
- ]
- )
- ```
-
- ---
-
- ## Responses
-
- ### JSON Response
-
- ```python
- 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
-
- ```python
- from django_bolt.response import Html
-
- @api.get("/html")
- async def html_handler(request):
- return Html("<h1>Hello World</h1>")
- ```
-
- ### Streaming Response (SSE)
-
- ```python
- 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:
-
- ```python
- 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):
-
- ```python
- @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+)
-
- ```python
- @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+)
-
- ```python
- from django_bolt.exceptions import HTTPException
-
- raise HTTPException(
- status_code=400,
- body={"error": "validation_failed", "fields": {"email": "Invalid"}}
- )
- ```
-
- ### File Response
-
- ```python
- 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"
- )
+ async def create_user(data: UserCreate):
+ user = await User.objects.acreate(username=data.username, email=data.email)
+ return {"id": user.id}, 201
```
- ### Redirect Response
+ ### Middleware
```python
- from django_bolt.response import Redirect
+ from django_bolt.middleware import CompressionMiddleware
- @api.get("/redirect")
- async def redirect_handler(request):
- return Redirect(url="https://example.com", status=302)
+ api.add_middleware(CompressionMiddleware())
```
---
## Class-Based Views
### ViewSet
```python
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
```python
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
-
- ```python
- 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:
-
- ```python
- 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+)
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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.
+ Django Bolt auto-generates OpenAPI 3.1 documentation.
### Access Docs
- **Swagger**: `/docs`
- **ReDoc**: `/redoc`
- **Scalar**: `/scalar`
- **RapidDoc**: `/rapiddoc`
### Configure OpenAPI
```python
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+)
```python
- # 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
```python
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
| Endpoint Type | Requests/sec |
|--------------|-------------|
| Hello-world (10KB JSON) | **~311,000 RPS** |
| 10KB JSON response | **~187,000 RPS** |
| 10-row ORM query | **~21,000–27,000 RPS** |
Source: https://bolt.farhana.li/benchmarks/
---
## Configuration
### Settings
```python
# 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
+ "HOST": "0.0.0.0",
+ "PORT": 8000,
+ "PROCESSES": 4,
+ "BACKLOG": 2048,
+ "KEEP_ALIVE": 30,
"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
+ JWT_EXPIRATION = 3600
# 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
- )
+ BOLT_MAX_UPLOAD_SIZE = FileSize.MB_50
+ BOLT_MEMORY_SPOOL_THRESHOLD = 5 * 1024 * 1024
# 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+)
-
- ```python
- # settings.py
- BOLT_AUTHENTICATION_CLASSES = [
- "django_bolt.auth.JWTAuthentication",
- ]
-
- BOLT_PERMISSION_CLASSES = [
- "django_bolt.auth.IsAuthenticated",
- ]
- ```
-
---
## Production Server: runbolt
### Worker Recycling (v0.11.0+)
```bash
# Recycle workers after memory threshold
- python manage.py runbolt --max-rss 512000 # 512 MB
+ python manage.py runbolt --max-rss 512000
# Limit worker lifetime
- python manage.py runbolt --workers-lifetime 3600 # 1 hour
+ python manage.py runbolt --workers-lifetime 3600
# 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+)
```bash
- # Native Rust reloader (watches project root + import graph)
+ # Native Rust reloader
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
-
- ```ini
- # /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:
-
- ```bash
- sudo systemctl daemon-reload
- sudo systemctl enable django-bolt
- sudo systemctl start django-bolt
- sudo systemctl status django-bolt
- ```
-
- ### supervisor
-
- ```ini
- # /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
- ```
-
- ```bash
- sudo supervisorctl reread
- sudo supervisorctl update
- sudo supervisorctl start django-bolt
- ```
-
- ### Reverse Proxy with nginx
-
- ```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+)
-
- ```python
- 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)
-
- ```python
- 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
- # 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"]
- ```
-
- ```yaml
- # 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:
-
- ```bash
- # 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
-
- ```bash
- pip install "django-bolt[mcp]"
- ```
-
- ### Mount MCP
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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+)
-
- ```python
- 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:
-
- ```python
- # 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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:
-
- ```json
- {
- "detail": [
- {"loc": ["body", "email"], "msg": "Invalid email format", "type": "value_error"},
- {"loc": ["body", "age"], "msg": "Must be positive", "type": "value_error"}
- ]
- }
- ```
-
- ### Error Handlers
-
- ```python
- 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
-
- | 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 |
-
- ---
-
- ## Pagination
-
- Django-Bolt provides three pagination styles for handling large datasets efficiently.
-
- ### PageNumber Pagination
-
- ```python
- 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:
-
- ```json
- {
- "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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- # 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
-
- ```python
- # 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
-
- ```python
- # 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- @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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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.
-
- ### Basic Usage
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- 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
-
- ```python
- @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
-
- ```python
- 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
-
- ```python
- # 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
-
- ```python
- 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)}
- ```
-
- ### Basic Async ORM Methods
-
- ```python
- 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")
-
- # 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
-
- ```python
- # select_related for ForeignKey/OneToOne
- @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
- })
- return {"articles": articles}
-
- # prefetch_related for ManyToMany
- @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()]
- return {"articles": [...]}
- ```
-
- ### Transactions
-
- ```python
- 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}
- ```
-
- ### Aggregations
-
- ```python
- 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
- ```
-
- ### DJANGO_BOLT_ORM_THREADS (v0.11.0+)
-
- Bounded thread pool for sync ORM operations:
-
- ```python
- # settings.py
- DJANGO_BOLT_ORM_THREADS = 20 # Bounded pool size
- ```
-
- ---
-
- ## Background Tasks
-
- ### Django 6.0 Tasks Framework
-
- ```python
- from django.tasks import task
-
- @task
- def process_video(video_path: str):
- 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)
-
- process_video.enqueue(path)
-
- return {"path": path, "status": "processing"}
- ```
-
- ### Celery Integration
-
- ```python
- # tasks.py
- from celery import shared_task
-
- @shared_task
- def process_data_async(data_id: int):
- import asyncio
- from myapp.models import DataRecord
-
- async def process():
- record = await DataRecord.objects.aget(id=data_id)
- record.status = "processed"
- await record.asave()
-
- asyncio.run(process())
-
- @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
-
- ```python
- @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}
- ```
-
- ---
- ---
-
- ## 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 (300k+ 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
-
- ```bash
- pip install django-bolt
- ```
-
- Add to `INSTALLED_APPS`:
-
- ```python
- INSTALLED_APPS = [
- ...
- "django_bolt"
- ...
- ]
- ```
-
- ### Step 2: Convert Function-Based Views
-
- **Before (DRF):**
-
- ```python
- # 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):**
-
- ```python
- # 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):**
-
- ```python
- # 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):**
-
- ```python
- # 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):**
-
- ```python
- # 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):**
-
- ```python
- # 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):**
-
- ```python
- # 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):**
-
- ```python
- # 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):**
-
- ```bash
- # Development
- python manage.py runserver
-
- # Production (with gunicorn)
- gunicorn myproject.wsgi --workers 4
- ```
-
- **After (Django Bolt):**
-
- ```bash
- # 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
### Performance
1. **Use async ORM methods** - Always use `aget()`, `alist()`, `acreate()` in handlers
2. **Avoid N+1 queries** - Use `select_related()` for ForeignKey, `prefetch_related()` for ManyToMany
3. **Enable compression** - Use `CompressionMiddleware` for responses > 500 bytes
4. **Tune process count** - Set `--processes` to CPU core count for CPU-bound workloads
### Security
1. **Always validate input** - Use msgspec structs for request body validation
2. **Use HTTPS in production** - Terminate TLS at nginx/load balancer
3. **Configure BOLT_TRUSTED_PROXIES** - Required behind reverse proxies for accurate rate limiting
4. **Rotate JWT secrets** - Implement key rotation for long-lived deployments
### Maintainability
1. **Organize by feature** - Group related endpoints in feature modules
2. **Use dependency injection** - Extract reusable logic with `Depends()`
3. **Write tests** - Use `AsyncAPITestClient` for integration tests
4. **Document APIs** - Leverage auto-generated OpenAPI docs at `/docs`
---
## Command-Line Reference
```bash
# Development server with native Rust reloader
python manage.py runbolt --dev
# Production with custom settings
python manage.py runbolt --host 0.0.0.0 --port 8000 --processes 4
# Worker recycling
python manage.py runbolt --max-rss 512000 --workers-lifetime 3600
# Increase socket backlog
python manage.py runbolt --processes 4 --backlog 2048
# Adjust keep-alive timeout
python manage.py runbolt --processes 4 --keep-alive 30
# Skip startup checks
python manage.py runbolt --skip-checks
```
---
## Version Migration Gates
| Version | Breaking Change |
|---------|-----------------|
| v0.4.0 | Python 3.12+ required |
| v0.6.0 | SessionAuthentication removed |
| v0.10.0 | bolt-mcp 0.2 API + cookie-JWT CSRF check ON by default |
| v0.10.3 | BOLT_TRUSTED_PROXIES required behind proxies |
| v0.11.0 | Nested() removed — use plain type hints |
---
## References
- **Official Docs**: https://bolt.farhana.li/
- **LLMs.txt**: https://bolt.farhana.li/llms.txt
- **Releases**: https://github.com/dj-bolt/django-bolt/releases
- **CHANGELOG**: https://github.com/dj-bolt/django-bolt/blob/main/CHANGELOG.md
- **Benchmarks**: https://bolt.farhana.li/benchmarks/
- **MCP Guide**: https://bolt.farhana.li/topics/mcp/
+
+ ---
+
+ ## Deep Dives
+
+ The following reference files contain detailed coverage of advanced topics. Load them on demand when working on specific features:
+
+ 1. **auth-security.md** — Authentication, Permissions & Guards, Middleware, Rate Limiting, Error Handling
+ 2. **responses-serialization.md** — Responses, msgspec Serializers, Pagination, Request Object, Dependency Injection
+ 3. **websockets-uploads.md** — WebSockets, File Uploads
+ 4. **orm-background.md** — Django ORM Patterns, Background Tasks
+ 5. **deployment.md** — Deployment, MCP Servers
+ 6. **drf-migration.md** — Comparison with DRF/Django Ninja, Migration from DRF
+
+ ---