القائمة
Django Bolt - Rust-powered high-performance API framework, 60k+ RPS, decorator routing, built-in auth, async ORM
يُحدَّد Prompt الذي يراجع المصدر أولًا بشكل افتراضي. يمكنك التبديل إلى أمر مباشر أو تنزيل نسخة محلية.
اقرأ SKILL.md وأي ملفات مرافقة يعرضها SkillsMP قبل أن تقرر التثبيت.
التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
يتجاوز الأمر المباشر Prompt المخصّص للمراجعة. افحص المصدر قبل تشغيله.
npx skills add https://github.com/Mte90/dotfiles --skill django-boltيبقى الأمر في سطر واحد. مرّر أفقيًا لمراجعته كاملًا قبل النسخ.
تفضّل نسخة محلية؟ نزّل الملفات المتاحة حاليًا لدى SkillsMP.
عرض SKILL.md
| name | django-bolt |
| description | Django Bolt - Rust-powered high-performance API framework, 60k+ RPS, decorator routing, built-in auth, async ORM |
| metadata | {"author":"mte90","version":"1.0.0","tags":["python","django","bolt","api","rust","performance","async","high-performance"]} |
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.
Architecture:
Why Django Bolt?
pip install django-bolt
# settings.py
INSTALLED_APPS = [
...
"django_bolt"
...
]
# 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}
# Development
python manage.py runbolt --dev
# Production (standalone)
python manage.py runbolt
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/{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}
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}
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}
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}
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"}
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
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"}
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"
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(),
]
)
from django.middleware.security import SecurityMiddleware
api = BoltAPI(
django_middleware=[
SecurityMiddleware,
# Your Django middleware here
]
)
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"}
from django_bolt.response import Html
@api.get("/html")
async def html_handler(request):
return Html("<h1>Hello World</h1>")
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())
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"
)
from django_bolt.response import Redirect
@api.get("/redirect")
async def redirect_handler(request):
return Redirect(url="https://example.com", status=302)
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")
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")
msgspec provides 5-10x faster JSON serialization than Python's stdlib.
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}
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")
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
}
Django Bolt auto-generates OpenAPI documentation.
/docs/redoc/scalar/rapiddocfrom django_bolt import BoltAPI
from django_bolt.openapi import OpenAPIInfo
api = BoltAPI(
info=OpenAPIInfo(
title="My API",
version="1.0.0",
description="API description",
)
)
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)
| Endpoint Type | Requests/sec |
|---|---|
| Root endpoint | ~100,000 RPS |
| JSON parsing/validation (10kb) | ~83,700 RPS |
| Path + Query parameters | ~85,300 RPS |
| HTML response | ~100,600 RPS |
| Redirect response | ~96,300 RPS |
| Form data handling | ~76,800 RPS |
| ORM reads (SQLite, 10 records) | ~13,000 RPS |
10,000 concurrent clients:
# 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
# Run in production mode
python manage.py runbolt --workers 4
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"}
}
)
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}
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)
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,
}
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,
}
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}
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}
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)}
@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}
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"}
| Feature | Django Bolt | Django REST Framework | Django Ninja |
|---|---|---|---|
| RPS | 60,000+ | ~1,000 | ~3,000 |
| Auth in Rust | ✅ | ❌ | ❌ |
| msgspec serialization | ✅ | ❌ | ❌ |
| OpenAPI auto-gen | ✅ | Needs drf-spectacular | ✅ |
| Django Admin | ✅ | ✅ | ✅ |
| Django ORM | ✅ (async) | ✅ (sync) | ✅ (async) |
| Serializer | msgspec | DRF Serializers | Pydantic |
| Middleware in Rust | ✅ | ❌ | ❌ |
pip install django-bolt
# settings.py - Keep DRF, add Bolt
INSTALLED_APPS = [
...
"rest_framework",
"django_bolt",
]
# bolt_api.py - New file
from django_bolt import BoltAPI
api = BoltAPI()
# 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")
# DRF on standard Django server
python manage.py runserver 8000
# Bolt API on its own server
python manage.py runbolt 8001
# settings.py
INSTALLED_APPS = [
...
"django_bolt",
# "rest_framework", # Remove when done
]
# /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
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"]
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;
}
}
Django-Bolt provides a structured exception hierarchy for HTTP errors and automatic error response formatting.
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")
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"
}
)
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"}
]
}
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)
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 | 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 |
Access the full request using the request parameter for comprehensive request data.
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
}
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}
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)}
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}
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"}
| 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 |
Django-Bolt provides dependency injection using the Depends marker for reusable parameter extractors.
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}
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}
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
}
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
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
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
Django-Bolt provides the UploadFile class for handling file uploads with Django integration.
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,
}
| 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 |
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}
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
@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],
}
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}
# 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
Django-Bolt provides three pagination styles for handling large datasets efficiently.
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": [...]
}
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
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
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()
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()
Django-Bolt provides WebSocket support for real-time bidirectional communication.
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}")
# 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"})
# 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 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()
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")
Apply authentication to WebSocket endpoints:
from django_bolt.auth import JWTAuthentication, IsAuthenticated
@api.websocket(
"/ws/protected",
auth=[JWTAuthentication()],
guards=[IsAuthenticated()]
)
async def protected(websocket: WebSocket):
user_id = websocket.context.get("user_id")
await websocket.accept()
await websocket.send_text(f"Welcome, user {user_id}")
connected_clients = set()
@api.websocket("/ws/broadcast")
async def broadcast(websocket: WebSocket):
await websocket.accept()
connected_clients.add(websocket)
try:
async for message in websocket.iter_text():
for client in connected_clients:
await client.send_text(message)
finally:
connected_clients.discard(websocket)
rooms = {} # room_id -> set of websockets
@api.websocket("/ws/room/{room_id}")
async def room(websocket: WebSocket, room_id: str):
await websocket.accept()
if room_id not in rooms:
rooms[room_id] = set()
rooms[room_id].add(websocket)
try:
async for message in websocket.iter_text():
for client in rooms[room_id]:
await client.send_text(f"[{room_id}] {message}")
finally:
rooms[room_id].discard(websocket)
Django-Bolt integrates with Django's task framework for asynchronous background processing.
# tasks.py
from celery import shared_task
@shared_task
def process_data_async(data_id: int):
"""Process data in background."""
import asyncio
from myapp.models import DataRecord
async def process():
record = await DataRecord.objects.aget(id=data_id)
# Expensive processing
record.status = "processed"
await record.asave()
asyncio.run(process())
# API handler
from myapp.tasks import process_data_async
@api.post("/process")
async def start_processing(data_id: int):
process_data_async.delay(data_id)
return {"status": "processing_started", "data_id": data_id}
# Alternative: Django Q
@api.post("/process")
async def start_processing(data_id: int):
from django_q.tasks import async_task
async_task("myapp.functions.process_data", data_id)
return {"status": "queued", "data_id": data_id}
# For WebSocket + background task integration
@api.post("/long-task")
async def start_long_task(request: Request):
task_id = str(uuid.uuid4())
# Start background task
asyncio.create_task(background_processing(task_id))
return {"task_id": task_id, "status": "started"}
async def background_processing(task_id: str):
# Simulate long task
await asyncio.sleep(10)
# Notify via WebSocket (store connected clients globally)
if task_id in connected_tasks:
await connected_tasks[task_id].send_json({
"type": "task_complete",
"task_id": task_id
})
Django-Bolt handlers use async, requiring Django's async ORM methods for maximum performance.
from myapp.models import Article
# Get a single object
article = await Article.objects.aget(id=1)
# Create an object
article = await Article.objects.acreate(
title="My Article",
content="Content here"
)
# Get or create
article, created = await Article.objects.aget_or_create(
title="My Article",
defaults={"content": "Default content"}
)
# Count
total = await Article.objects.acount()
# Check existence
exists = await Article.objects.filter(published=True).aexists()
# Delete
deleted_count, _ = await Article.objects.filter(draft=True).adelete()
# Update
updated_count = await Article.objects.filter(draft=True).aupdate(published=True)
Use select_related for ForeignKey and OneToOne:
# Good: 1 query with JOIN
@api.get("/articles")
async def list_articles():
articles = []
async for article in Article.objects.select_related("author")[:20]:
articles.append({
"id": article.id,
"author_name": article.author.username # No extra query!
})
return {"articles": articles}
Use prefetch_related for ManyToMany and reverse ForeignKey:
# Good: 2 queries (articles + tags)
@api.get("/articles")
async def list_articles():
queryset = Article.objects.select_related("author").prefetch_related("tags")
async for article in queryset[:20]:
tags = [tag.name for tag in article.tags.all()] # Already prefetched!
return {"articles": [...]}
Use sync_to_async for database transactions:
from asgiref.sync import sync_to_async
from django.db import transaction
@api.post("/transfer")
async def transfer_funds(from_id: int, to_id: int, amount: float):
@sync_to_async
def do_transfer():
with transaction.atomic():
from_account = Account.objects.select_for_update().get(id=from_id)
to_account = Account.objects.select_for_update().get(id=to_id)
from_account.balance -= amount
to_account.balance += amount
from_account.save()
to_account.save()
return {"success": True}
return await do_transfer()
from django.db.models import Count, Avg, Q
@api.get("/stats")
async def article_stats():
stats = await Article.objects.aaggregate(
total=Count("id"),
published=Count("id", filter=Q(published=True)),
avg_comments=Avg("comment_count")
)
return stats
# Bulk create
articles = [
Article(title=f"Article {i}", content="...")
for i in range(100)
]
created = await Article.objects.abulk_create(articles)
# Bulk update
await Article.objects.filter(draft=True).aupdate(published=True)
Full reference for Django-Bolt settings in settings.py:
# settings.py
# Required: Add to INSTALLED_APPS
INSTALLED_APPS = [
...
"django_bolt"
...
]
# Django-Bolt Configuration
BOLT = {
# Server settings
"HOST": "0.0.0.0", # Server host
"PORT": 8000, # Server port
"PROCESSES": 4, # Number of worker processes
"BACKLOG": 2048, # Socket backlog size
"KEEP_ALIVE": 30, # Keep-alive timeout in seconds
# Debug mode
"DEBUG": False,
# Enable signals (may impact performance)
"EMIT_SIGNALS": False,
}
# JWT Configuration
JWT_SECRET_KEY = "your-secret-key"
JWT_ALGORITHM = "HS256"
JWT_EXPIRATION = 3600 # seconds
# File Upload Settings
from django_bolt import FileSize
BOLT_MAX_UPLOAD_SIZE = FileSize.MB_50 # 50 MB max
BOLT_MEMORY_SPOOL_THRESHOLD = 5 * 1024 * 1024 # 5 MB
# Compression Configuration
from django_bolt.middleware import CompressionConfig
BOLT_COMPRESSION = CompressionConfig(
backend="gzip",
minimum_size=500, # Only compress responses > 500 bytes
)
# CORS Configuration
BOLT_CORS = {
"allow_origins": ["https://example.com"],
"allow_methods": ["GET", "POST", "PUT", "DELETE"],
"allow_headers": ["*"],
"allow_credentials": True,
}
# Run development server
python manage.py runbolt --dev
# Run production server with multiple processes
python manage.py runbolt --host 0.0.0.0 --port 8000 --processes 4
# Increase socket backlog for high traffic
python manage.py runbolt --processes 4 --backlog 2048
# Adjust keep-alive timeout
python manage.py runbolt --processes 4 --keep-alive 30
Production deployment guide for Django-Bolt.
Create /etc/systemd/system/django-bolt.service:
[Unit]
Description=Django-Bolt API Server
After=network.target
[Service]
User=www-data
Group=www-data
WorkingDirectory=/path/to/your/project
ExecStart=/path/to/venv/bin/python manage.py runbolt --host 127.0.0.1 --port 8000 --processes 4
Restart=always
RestartSec=3
[Install]
WantedBy=multi-user.target
Enable and start:
sudo systemctl daemon-reload
sudo systemctl enable django-bolt
sudo systemctl start django-bolt
sudo systemctl status django-bolt
Create /etc/supervisor/conf.d/django-bolt.conf:
[program:django-bolt]
command=/path/to/venv/bin/python manage.py runbolt --host 127.0.0.1 --port 8000 --processes 4
directory=/path/to/your/project
user=www-data
autostart=true
autorestart=true
redirect_stderr=true
stdout_logfile=/var/log/django-bolt.log
sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start django-bolt
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;
}
}
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,
}
},
}
}
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,
}
}
# Dockerfile
FROM python:3.12-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application
COPY . .
EXPOSE 8000
CMD ["python", "manage.py", "runbolt", "--host", "0.0.0.0", "--port", "8000", "--processes", "4"]
# docker-compose.yml
services:
api:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgres://user:pass@db:5432/mydb
depends_on:
- db
db:
image: postgres:15
environment:
POSTGRES_DB: mydb
POSTGRES_USER: user
POSTGRES_PASSWORD: pass
Django-Bolt uses processes for parallelism, not workers. Each process has its own GIL:
# Use processes for parallelism (not workers)
python manage.py runbolt --processes 4
# Rule of thumb: set --processes to number of CPU cores
| 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 |
Step-by-step guide to migrate Django REST Framework views to Django Bolt.
pip install django-bolt
Add to INSTALLED_APPS:
INSTALLED_APPS = [
...
"django_bolt"
...
]
Before (DRF):
# views.py (DRF)
from rest_framework.decorators import api_view
from rest_framework.response import Response
@api_view(["GET", "POST"])
def user_list(request):
if request.method == "GET":
users = User.objects.all()
serializer = UserSerializer(users, many=True)
return Response(serializer.data)
serializer = UserSerializer(data=request.data)
if serializer.is_valid():
serializer.save()
return Response(serializer.data, status=201)
return Response(serializer.errors, status=400)
After (Django Bolt):
# api.py (Django Bolt)
from django_bolt import BoltAPI
import msgspec
api = BoltAPI()
class UserSchema(msgspec.Struct):
id: int
username: str
email: str
@api.get("/users")
async def list_users():
users = await User.objects.all()
return [{"id": u.id, "username": u.username, "email": u.email} for u in users]
@api.post("/users")
async def create_user(data: UserSchema):
user = await User.objects.acreate(
username=data.username,
email=data.email
)
return {"id": user.id, "username": user.username}, 201
Before (DRF):
# views.py (DRF)
from rest_framework import viewsets
from rest_framework.permissions import IsAuthenticated
from .serializers import UserSerializer
class UserViewSet(viewsets.ModelViewSet):
queryset = User.objects.all()
serializer_class = UserSerializer
permission_classes = [IsAuthenticated]
After (Django Bolt):
# api.py (Django Bolt)
from django_bolt import BoltAPI, ViewSet
from django_bolt.auth import JWTAuthentication, IsAuthenticated
from django.contrib.auth import get_user_model
User = get_user_model()
api = BoltAPI()
@api.viewset("/users", auth=[JWTAuthentication()], guards=[IsAuthenticated()])
class UserViewSet(ViewSet):
async def list(self, request):
users = await User.objects.all()
return [{"id": u.id, "username": u.username} for u in users]
async def retrieve(self, request, pk: int):
user = await User.objects.aget(id=pk)
return {"id": user.id, "username": user.username}
async def create(self, request):
data = await request.json()
user = await User.objects.acreate(**data)
return {"id": user.id}, 201
async def destroy(self, request, pk: int):
await User.objects.filter(id=pk).adelete()
return {"deleted": True}
Before (DRF):
# serializers.py (DRF)
from rest_framework import serializers
from .models import Article
class ArticleSerializer(serializers.ModelSerializer):
author_name = serializers.CharField(source="author.username", read_only=True)
class Meta:
model = Article
fields = ["id", "title", "content", "author", "author_name", "created_at"]
After (Django Bolt):
# schemas.py (Django Bolt)
import msgspec
class AuthorSchema(msgspec.Struct):
id: int
username: str
class ArticleSchema(msgspec.Struct):
id: int
title: str
content: str
author: AuthorSchema
created_at: str
Before (DRF):
# urls.py (DRF)
from rest_framework.routers import DefaultRouter
from .views import UserViewSet
router = DefaultRouter()
router.register(r"users", UserViewSet)
urlpatterns = router.urls
After (Django Bolt):
# myproject/urls.py
from django.urls import path
from myapp.api import api
urlpatterns = [
path("api/", api.urls),
]
Before (DRF):
# Development
python manage.py runserver
# Production (with gunicorn)
gunicorn myproject.wsgi --workers 4
After (Django Bolt):
# Development
python manage.py runbolt --dev
# Production (standalone, no gunicorn needed)
python manage.py runbolt --processes 4
# ✅ GOOD: Use decorators
from django_bolt.http import HttpRequest, JsonResponse
from django_bolt.utils.decorators import json_response
@json_response()
def get_users(request: HttpRequest) -> JsonResponse:
return {"users": []}
# ✅ GOOD: Serializer validation
from django_bolt.serializers import Serializer
class UserSerializer(Serializer):
name: str
email: str
def validate_email(self, value):
if not value.endswith('@company.com'):
raise ValidationError("Must use company email")
return value
from django_bolt.errors import ApiError
# ✅ GOOD: Raise proper errors
def get_user(request, user_id):
user = User.objects.get_or_404(user_id)
if not user.is_active:
raise ApiError(400, "User not active")
return user
# ✅ GOOD: Use built-in caching
from django_bolt.cache import cache_page
@cache_page(60 * 15)
def get_data(request):
...
# ✅ GOOD: Query optimization
users = User.objects.select_related('profile').prefetch_related('posts')