django-bolt

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Django-Bolt

Critical rules

关键规则

Always use async handlers unless the user explicitly needs sync
Use async database access in async handlers (e.g., Django async ORM:
```
aget
```
,
```
acreate
```
,
```
afilter
```
)
Use
```
msgspec.Struct
```
for simple typed payloads; use
```
Serializer
```
for richer validation and reusable field sets
Keep route signatures typed -- let Django-Bolt validate inputs instead of hand-parsing
```
request.body
```
Use built-in HTTP exceptions (
```
HTTPException
```
) for expected API failures

Run with

python manage.py runbolt --dev

for development, not

uvicorn

gunicorn

除非用户明确需要同步处理，否则始终使用异步处理器
在异步处理器中使用异步数据库访问（例如 Django 异步 ORM：
```
aget
```
、
```
acreate
```
、
```
afilter
```
）
简单类型化负载使用
```
msgspec.Struct
```
；更丰富的校验逻辑和可复用字段集使用
```
Serializer
```
保持路由签名带类型标注 -- 让Django-Bolt自动校验输入，不要手动解析
```
request.body
```
预期内的API错误使用内置HTTP异常（
```
HTTPException
```
）返回

开发环境使用

python manage.py runbolt --dev

启动，不要用

uvicorn

或

gunicorn

Apply this skill when

适用该技能的场景

The user mentions

django-bolt

BoltAPI

runbolt

Depends

response_model

APIView

JWTAuthentication

OpenAPIConfig

TestClient

WebSocket

StreamingResponse

, or

SSE

The user wants to start a Django-Bolt app, add endpoints, wire auth, expose docs, add pagination, return files, streams, or SSE events, or test handlers
The user wants to migrate from FastAPI, DRF, or Django Ninja -- see
```
references/migration-playbook.md
```

用户提到

django-bolt

、

BoltAPI

、

runbolt

、

Depends

、

response_model

、

APIView

、

JWTAuthentication

、

OpenAPIConfig

、

TestClient

、

WebSocket

、

StreamingResponse

或

SSE

时

用户想要启动Django-Bolt应用、添加端点、配置认证、暴露文档、添加分页、返回文件、流或SSE事件，或者测试处理器时
用户想要从FastAPI、DRF或Django Ninja迁移时 -- 参考
```
references/migration-playbook.md
```

Do NOT apply when

请勿使用的场景

The user is working with plain Django views, templates, or Django admin
The user is building a standard DRF API without mentioning django-bolt
The task is about Django ORM models, migrations, or admin configuration only

用户在使用原生Django视图、模板或Django admin时
用户在构建标准DRF API且没有提到django-bolt时
任务仅涉及Django ORM模型、迁移或admin配置时

Code examples

代码示例

Minimal setup

最小配置

python

undefined

python

undefined

settings.py

INSTALLED_APPS = [ "django_bolt", ]


```python

INSTALLED_APPS = [ "django_bolt", ]


```python

myproject/api.py

from django_bolt import BoltAPI

api = BoltAPI()

@api.get("/health") async def health(): return {"status": "ok"}


Run: `python manage.py runbolt --dev`

from django_bolt import BoltAPI

api = BoltAPI()

@api.get("/health") async def health(): return {"status": "ok"}


启动命令：`python manage.py runbolt --dev`

BoltAPI constructor options

BoltAPI构造选项

python

from django_bolt import (
    BoltAPI,
    CompressionConfig,
    LoggingMiddleware,
    OpenAPIConfig,
    ScalarRenderPlugin,
    TimingMiddleware,
)

api = BoltAPI(
    prefix="/api/v1",                        # URL prefix for all routes
    trailing_slash="strip",                  # "strip", "append", or "keep"
    validate_response=True,                  # validate response_model output
    compression=CompressionConfig(),         # or False to disable
    enable_logging=True,
    middleware=[                              # Python middleware stack
        TimingMiddleware,
        LoggingMiddleware,
    ],
    openapi_config=OpenAPIConfig(
        title="My API",
        version="1.0.0",
        path="/docs",
        render_plugins=[ScalarRenderPlugin()],
        enabled=True,
    ),
)

python

from django_bolt import (
    BoltAPI,
    CompressionConfig,
    LoggingMiddleware,
    OpenAPIConfig,
    ScalarRenderPlugin,
    TimingMiddleware,
)

api = BoltAPI(
    prefix="/api/v1",                        # 所有路由的URL前缀
    trailing_slash="strip",                  # 可选值："strip"（去除）、"append"（追加）、"keep"（保留）
    validate_response=True,                  # 校验response_model输出
    compression=CompressionConfig(),         # 设为False可关闭压缩
    enable_logging=True,
    middleware=[                              # Python中间件栈
        TimingMiddleware,
        LoggingMiddleware,
    ],
    openapi_config=OpenAPIConfig(
        title="My API",
        version="1.0.0",
        path="/docs",
        render_plugins=[ScalarRenderPlugin()],
        enabled=True,
    ),
)

Route decorators and parameters

路由装饰器和参数

All HTTP methods:

@api.get

@api.post

@api.put

@api.patch

@api.delete

@api.head

@api.options

python

import msgspec
from django_bolt import BoltAPI

api = BoltAPI()


class ItemOut(msgspec.Struct):
    id: int
    name: str


@api.get(
    "/items/{item_id}",
    response_model=ItemOut,        # validates and documents the output shape
    status_code=200,               # default status code
    tags=["items"],                # OpenAPI tags
    summary="Get an item",         # OpenAPI summary
    description="Fetch item by ID",
    guards=[],                     # permission guards
    auth=[],                       # auth backends
    validate_response=True,        # validate output matches response_model
)
async def get_item(item_id: int):
    return {"id": item_id, "name": "Widget"}

支持所有HTTP方法：

@api.get

、

@api.post

、

@api.put

、

@api.patch

、

@api.delete

、

@api.head

、

@api.options

。

python

import msgspec
from django_bolt import BoltAPI

api = BoltAPI()


class ItemOut(msgspec.Struct):
    id: int
    name: str


@api.get(
    "/items/{item_id}",
    response_model=ItemOut,        # 校验并标注输出结构
    status_code=200,               # 默认状态码
    tags=["items"],                # OpenAPI标签
    summary="获取单个商品",         # OpenAPI摘要
    description="根据ID获取商品信息",
    guards=[],                     # 权限守卫
    auth=[],                       # 认证后端
    validate_response=True,        # 校验输出是否匹配response_model
)
async def get_item(item_id: int):
    return {"id": item_id, "name": "小部件"}

Typed request body (msgspec.Struct)

类型化请求体（msgspec.Struct）

python

import msgspec
from django_bolt import BoltAPI

api = BoltAPI()


class ItemIn(msgspec.Struct):
    name: str
    price: float
    tags: list[str] = []


class ItemOut(msgspec.Struct):
    id: int
    name: str
    price: float


@api.post("/items", response_model=ItemOut, status_code=201)
async def create_item(item: ItemIn):
    return {"id": 1, "name": item.name, "price": item.price}

python

import msgspec
from django_bolt import BoltAPI

api = BoltAPI()


class ItemIn(msgspec.Struct):
    name: str
    price: float
    tags: list[str] = []


class ItemOut(msgspec.Struct):
    id: int
    name: str
    price: float


@api.post("/items", response_model=ItemOut, status_code=201)
async def create_item(item: ItemIn):
    return {"id": 1, "name": item.name, "price": item.price}

Serializer with validation

带校验的序列化器

python

from __future__ import annotations

from typing import Annotated

import msgspec
from django_bolt import BoltAPI
from django_bolt.serializers import Serializer, field_validator, model_validator

api = BoltAPI()


class UserCreate(Serializer):
    username: str
    email: str
    password: Annotated[str, msgspec.Meta(min_length=8)]

    class Config:
        write_only = {"password"}

    @field_validator("email")
    def validate_email(cls, value):
        if "@" not in value:
            raise ValueError("Invalid email")
        return value.lower()

    @model_validator
    def validate_model(self):
        if self.username.lower() == "admin":
            raise ValueError("Username 'admin' is reserved")
        return self


class UserOut(Serializer):
    id: int
    username: str
    email: str
    is_active: bool

    class Config:
        read_only = {"id"}


@api.post("/users", response_model=UserOut, status_code=201)
async def create_user(data: UserCreate):
    return {"id": 1, "username": data.username, "email": data.email, "is_active": True}

python

from __future__ import annotations

from typing import Annotated

import msgspec
from django_bolt import BoltAPI
from django_bolt.serializers import Serializer, field_validator, model_validator

api = BoltAPI()


class UserCreate(Serializer):
    username: str
    email: str
    password: Annotated[str, msgspec.Meta(min_length=8)]

    class Config:
        write_only = {"password"}

    @field_validator("email")
    def validate_email(cls, value):
        if "@" not in value:
            raise ValueError("无效的邮箱地址")
        return value.lower()

    @model_validator
    def validate_model(self):
        if self.username.lower() == "admin":
            raise ValueError("用户名'admin'为保留字")
        return self


class UserOut(Serializer):
    id: int
    username: str
    email: str
    is_active: bool

    class Config:
        read_only = {"id"}


@api.post("/users", response_model=UserOut, status_code=201)
async def create_user(data: UserCreate):
    return {"id": 1, "username": data.username, "email": data.email, "is_active": True}

Parameter markers (Query, Header, Cookie, Form, File, Path, Depends)

参数标记（Query、Header、Cookie、Form、File、Path、Depends）

python

from __future__ import annotations

from typing import Annotated

import msgspec
from django_bolt import BoltAPI, Depends, UploadFile
from django_bolt.param_functions import Body, Cookie, File, Form, Header, Path, Query

api = BoltAPI()

python

from __future__ import annotations

from typing import Annotated

import msgspec
from django_bolt import BoltAPI, Depends, UploadFile
from django_bolt.param_functions import Body, Cookie, File, Form, Header, Path, Query

api = BoltAPI()

Query parameters with validation

带校验的查询参数

@api.get("/search") async def search( q: Annotated[str, Query(min_length=1, max_length=100)], page: Annotated[int, Query(ge=1)] = 1, limit: Annotated[int, Query(ge=1, le=100)] = 20, ): return {"query": q, "page": page, "limit": limit}

Grouped query params via msgspec.Struct

通过msgspec.Struct分组查询参数

class Filters(msgspec.Struct): status: str | None = None sort: str = "created" order: str = "desc"

@api.get("/items") async def list_items(filters: Annotated[Filters, Query()]): return {"status": filters.status, "sort": filters.sort}

class Filters(msgspec.Struct): status: str | None = None sort: str = "created" order: str = "desc"

@api.get("/items") async def list_items(filters: Annotated[Filters, Query()]): return {"status": filters.status, "sort": filters.sort}

Path parameter with alias

带别名的路径参数

@api.get("/users/{user_id}") async def get_user(user_id: Annotated[int, Path(ge=1)]): return {"user_id": user_id}

Header extraction

提取请求头

@api.get("/secure") async def secure(x_api_key: Annotated[str, Header(alias="x-api-key")]): return {"key": x_api_key}

Grouped headers via struct

通过struct分组请求头

class APIHeaders(msgspec.Struct): authorization: str x_request_id: str | None = None # snake_case auto-maps to X-Request-Id

@api.get("/with-headers") async def with_headers(headers: Annotated[APIHeaders, Header()]): return {"auth": headers.authorization}

class APIHeaders(msgspec.Struct): authorization: str x_request_id: str | None = None # snake_case会自动映射为X-Request-Id

@api.get("/with-headers") async def with_headers(headers: Annotated[APIHeaders, Header()]): return {"auth": headers.authorization}

Cookie extraction

提取Cookie

@api.get("/preferences") async def preferences( theme: Annotated[str, Cookie()] = "light", lang: Annotated[str, Cookie()] = "en", ): return {"theme": theme, "lang": lang}

Form data

表单数据

@api.post("/login") async def login( username: Annotated[str, Form()], password: Annotated[str, Form()], ): return {"username": username}

Grouped form data via struct

通过struct分组表单数据

class ContactForm(msgspec.Struct): name: str email: str message: str

@api.post("/contact") async def contact(data: Annotated[ContactForm, Form()]): return {"name": data.name, "email": data.email}

class ContactForm(msgspec.Struct): name: str email: str message: str

@api.post("/contact") async def contact(data: Annotated[ContactForm, Form()]): return {"name": data.name, "email": data.email}

File uploads with validation

带校验的文件上传

from django_bolt import FileSize

@api.post("/upload") async def upload( title: Annotated[str, Form()], file: Annotated[ UploadFile, File(max_size=FileSize.MB_30, allowed_types=["application/pdf", "image/*"]), ], ): content = await file.read() return {"title": title, "filename": file.filename, "size": len(content)}

from django_bolt import FileSize

Multiple file upload

多文件上传

@api.post("/upload-many") async def upload_many( files: Annotated[list[UploadFile], File(max_files=5)], ): return {"count": len(files)}

Dependency injection

依赖注入

async def get_db(): db = await connect_db() try: yield db finally: await db.close()

async def get_current_user(request): user_id = request.auth.get("user_id") return {"id": user_id, "username": "example"}

@api.get("/profile") async def profile( user: Annotated[dict, Depends(get_current_user)], ): return {"id": user["id"], "username": user["username"]}

async def get_db(): db = await connect_db() try: yield db finally: await db.close()

async def get_current_user(request): user_id = request.auth.get("user_id") return {"id": user_id, "username": "example"}

@api.get("/profile") async def profile( user: Annotated[dict, Depends(get_current_user)], ): return {"id": user["id"], "username": user["username"]}

Explicit JSON body marker

显式JSON体标记

class UpdatePayload(msgspec.Struct): name: str value: int

@api.put("/settings") async def update_settings(data: Annotated[UpdatePayload, Body()]): return {"name": data.name}

undefined

class UpdatePayload(msgspec.Struct): name: str value: int

@api.put("/settings") async def update_settings(data: Annotated[UpdatePayload, Body()]): return {"name": data.name}

undefined

Response types

响应类型

python

from django_bolt import BoltAPI, JSON, Response, StreamingResponse
from django_bolt.responses import FileResponse, HTML, PlainText, Redirect

api = BoltAPI()

python

from django_bolt import BoltAPI, JSON, Response, StreamingResponse
from django_bolt.responses import FileResponse, HTML, PlainText, Redirect

api = BoltAPI()

Dict/list returns auto-serialize to JSON

字典/列表返回值会自动序列化为JSON

@api.get("/auto-json") async def auto_json(): return {"message": "hello"}

Explicit JSON with custom status

带自定义状态码的显式JSON

@api.post("/created") async def created(): return JSON({"id": 1, "created": True}, status_code=201)

Response with custom headers and cookies

带自定义请求头和Cookie的响应

@api.post("/login") async def login(): return ( Response({"token": "abc"}, status_code=200) .set_cookie("session", "xyz", httponly=True, secure=True, samesite="Strict") .set_cookie("theme", "dark", max_age=86400) )

@api.post("/logout") async def logout(): return Response({"ok": True}).delete_cookie("session")

PlainText

纯文本响应

@api.get("/text") async def text(): return PlainText("Hello, World!")

HTML

HTML响应

@api.get("/page") async def page(): return HTML("<h1>Hello</h1><p>Welcome to the API</p>")

Redirect

重定向

@api.get("/old-path") async def old_path(): return Redirect("/new-path") # 307 temporary by default

@api.get("/moved") async def moved(): return Redirect("/new-location", status_code=301) # permanent

@api.get("/old-path") async def old_path(): return Redirect("/new-path") # 默认是307临时重定向

@api.get("/moved") async def moved(): return Redirect("/new-location", status_code=301) # 永久重定向

File download

文件下载

@api.get("/download") async def download(): return FileResponse( "/path/to/report.pdf", filename="report.pdf", media_type="application/pdf", )

Streaming response

流式响应

@api.get("/stream") async def stream(): async def generate(): for chunk in ["Hello ", "World ", "!"]: yield chunk

return StreamingResponse(generate(), media_type="text/plain")

undefined

@api.get("/stream") async def stream(): async def generate(): for chunk in ["Hello ", "World ", "!"]: yield chunk

return StreamingResponse(generate(), media_type="text/plain")

undefined

SSE (Server-Sent Events)

SSE（服务器发送事件）

python

import asyncio
import json

from django_bolt import BoltAPI, StreamingResponse, no_compress

api = BoltAPI()


@api.get("/events")
@no_compress  # disable compression for SSE
async def events():
    async def generate():
        for i in range(10):
            data = json.dumps({"count": i, "message": f"event-{i}"})
            yield f"data: {data}\n\n"
            await asyncio.sleep(1)
        yield "event: done\ndata: stream complete\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

python

import asyncio
import json

from django_bolt import BoltAPI, StreamingResponse, no_compress

api = BoltAPI()


@api.get("/events")
@no_compress  # SSE场景下关闭压缩
async def events():
    async def generate():
        for i in range(10):
            data = json.dumps({"count": i, "message": f"event-{i}"})
            yield f"data: {data}\n\n"
            await asyncio.sleep(1)
        yield "event: done\ndata: stream complete\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

WebSocket

python

from django_bolt import BoltAPI, WebSocket, WebSocketDisconnect

api = BoltAPI()

python

from django_bolt import BoltAPI, WebSocket, WebSocketDisconnect

api = BoltAPI()

Echo server

回声服务器

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

JSON messages

JSON消息处理

@api.websocket("/ws/chat") async def chat(websocket: WebSocket, room: str): await websocket.accept() try: while True: data = await websocket.receive_json() response = {"room": room, "user": data.get("user"), "text": data.get("text")} await websocket.send_json(response) except WebSocketDisconnect: pass

undefined

undefined

Authentication and guards

认证与守卫

python

from django_bolt import (
    AllowAny,
    APIKeyAuthentication,
    BoltAPI,
    HasAllPermissions,
    HasAnyPermission,
    HasPermission,
    IsAdminUser,
    IsAuthenticated,
    IsStaff,
    JWTAuthentication,
    Request,
    create_jwt_for_user,
)

api = BoltAPI()

python

from django_bolt import (
    AllowAny,
    APIKeyAuthentication,
    BoltAPI,
    HasAllPermissions,
    HasAnyPermission,
    HasPermission,
    IsAdminUser,
    IsAuthenticated,
    IsStaff,
    JWTAuthentication,
    Request,
    create_jwt_for_user,
)

api = BoltAPI()

Generate JWT for a user

为用户生成JWT

@api.post("/auth/login") async def login(username: str, password: str): user = await verify_credentials(username, password) # your auth logic if not user: from django_bolt.exceptions import HTTPException raise HTTPException(status_code=401, detail="Invalid credentials") token = create_jwt_for_user(user) return {"access_token": token}

@api.post("/auth/login") async def login(username: str, password: str): user = await verify_credentials(username, password) # 你的认证逻辑 if not user: from django_bolt.exceptions import HTTPException raise HTTPException(status_code=401, detail="无效的凭证") token = create_jwt_for_user(user) return {"access_token": token}

JWT-protected endpoint

JWT保护的端点

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

API key auth

API密钥认证

@api.get("/api/data", auth=[APIKeyAuthentication()], guards=[IsAuthenticated()]) async def api_data(request: Request): return {"data": "secret"}

Multiple guards

多重守卫

@api.delete( "/admin/users/{user_id}", auth=[JWTAuthentication()], guards=[IsAuthenticated(), IsAdminUser()], ) async def delete_user(user_id: int): await remove_user(user_id) # your deletion logic return {"deleted": user_id}

Permission-based guards

基于权限的守卫

@api.post( "/articles", auth=[JWTAuthentication()], guards=[IsAuthenticated(), HasPermission("blog.add_article")], ) async def create_article(title: str, body: str): return {"title": title}

@api.get( "/reports", auth=[JWTAuthentication()], guards=[IsAuthenticated(), HasAnyPermission("reports.view", "reports.export")], ) async def view_reports(): return {"reports": []}

@api.post( "/deploy", auth=[JWTAuthentication()], guards=[IsAuthenticated(), HasAllPermissions("deploy.create", "deploy.approve")], ) async def deploy(): return {"status": "deploying"}

@api.post( "/articles", auth=[JWTAuthentication()], guards=[IsAuthenticated(), HasPermission("blog.add_article")], ) async def create_article(title: str, body: str): return {"title": title}

@api.get( "/reports", auth=[JWTAuthentication()], guards=[IsAuthenticated(), HasAnyPermission("reports.view", "reports.export")], ) async def view_reports(): return {"reports": []}

@api.post( "/deploy", auth=[JWTAuthentication()], guards=[IsAuthenticated(), HasAllPermissions("deploy.create", "deploy.approve")], ) async def deploy(): return {"status": "deploying"}

Staff-only

仅员工可访问

@api.get("/staff/dashboard", auth=[JWTAuthentication()], guards=[IsStaff()]) async def staff_dashboard(): return {"stats": {}}

Public endpoint (no auth required)

公开端点（无需认证）

@api.get("/public", guards=[AllowAny()]) async def public(): return {"public": True}

undefined

@api.get("/public", guards=[AllowAny()]) async def public(): return {"public": True}

undefined

Middleware (CORS, rate limiting, custom)

中间件（CORS、限流、自定义）

python

from django_bolt import BoltAPI, BaseMiddleware, Request, Response, cors, middleware, no_compress, rate_limit, skip_middleware

api = BoltAPI()

python

from django_bolt import BoltAPI, BaseMiddleware, Request, Response, cors, middleware, no_compress, rate_limit, skip_middleware

api = BoltAPI()

Per-route CORS

单路由CORS配置

@api.get("/public/data") @cors( origins=["https://example.com", "https://app.example.com"], methods=["GET", "POST"], headers=["Authorization", "Content-Type"], credentials=True, max_age=3600, ) async def public_data(): return {"data": "accessible cross-origin"}

Rate limiting

限流

@api.post("/api/expensive") @rate_limit(rps=10, burst=20, key="ip") async def expensive_operation(): return {"result": "ok"}

Skip specific middleware on a route

路由级跳过指定中间件

@api.get("/internal") @skip_middleware("compression", "rate_limit") async def internal(): return {"internal": True}

Disable compression for streaming

流式响应关闭压缩

@api.get("/stream") @no_compress async def stream(): from django_bolt import StreamingResponse async def gen(): yield "chunk1" yield "chunk2" return StreamingResponse(gen(), media_type="text/plain")

Custom Python middleware

自定义Python中间件

class RequestTimingMiddleware(BaseMiddleware): async def process_request(self, request: Request) -> Response: import time start = time.perf_counter() response = await self.get_response(request) elapsed = time.perf_counter() - start response.headers["X-Response-Time"] = f"{elapsed:.4f}s" return response

Attach custom middleware to a route

为路由绑定自定义中间件

@api.post("/upload") @middleware(RequestTimingMiddleware) async def upload(request: Request): return {"uploaded": True}

undefined

@api.post("/upload") @middleware(RequestTimingMiddleware) async def upload(request: Request): return {"uploaded": True}

undefined

Pagination

分页

python

from django_bolt import (
    BoltAPI,
    CursorPagination,
    LimitOffsetPagination,
    PageNumberPagination,
    paginate,
)

api = BoltAPI()

python

from django_bolt import (
    BoltAPI,
    CursorPagination,
    LimitOffsetPagination,
    PageNumberPagination,
    paginate,
)

api = BoltAPI()

Page-number pagination (default)

页码分页（默认）

GET /users?page=2&page_size=25

请求地址：GET /users?page=2&page_size=25

@api.get("/users") @paginate(PageNumberPagination) async def list_users(request): return await get_all_users() # return a queryset or list

@api.get("/users") @paginate(PageNumberPagination) async def list_users(request): return await get_all_users() # 返回查询集或列表

Custom page size

自定义页大小

class LargePages(PageNumberPagination): page_size = 50 max_page_size = 200 page_size_query_param = "page_size"

@api.get("/products") @paginate(LargePages) async def list_products(request): return await get_active_products()

class LargePages(PageNumberPagination): page_size = 50 max_page_size = 200 page_size_query_param = "page_size"

@api.get("/products") @paginate(LargePages) async def list_products(request): return await get_active_products()

Limit-offset pagination

偏移量分页

GET /logs?limit=50&offset=100

请求地址：GET /logs?limit=50&offset=100

@api.get("/logs") @paginate(LimitOffsetPagination) async def list_logs(request): return await get_all_logs()

Cursor pagination (best for infinite scroll / real-time feeds)

游标分页（最适合无限滚动/实时信息流）

GET /feed?cursor=abc123

请求地址：GET /feed?cursor=abc123

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

@api.get("/feed") @paginate(FeedPagination) async def feed(request): return await get_all_posts()

undefined

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

@api.get("/feed") @paginate(FeedPagination) async def feed(request): return await get_all_posts()

undefined

OpenAPI documentation

OpenAPI文档

python

from django_bolt import (
    BoltAPI,
    OpenAPIConfig,
    RapidocRenderPlugin,
    RedocRenderPlugin,
    ScalarRenderPlugin,
    StoplightRenderPlugin,
    SwaggerRenderPlugin,
)

python

from django_bolt import (
    BoltAPI,
    OpenAPIConfig,
    RapidocRenderPlugin,
    RedocRenderPlugin,
    ScalarRenderPlugin,
    StoplightRenderPlugin,
    SwaggerRenderPlugin,
)

Minimal -- serves Swagger UI at /docs

最简配置 -- 在/docs路径提供Swagger UI

api = BoltAPI( openapi_config=OpenAPIConfig( title="My API", version="1.0.0", ), )

Full configuration

完整配置

api = BoltAPI( openapi_config=OpenAPIConfig( title="Acme API", version="2.0.0", description="The Acme Corp backend API", path="/docs", # docs URL enabled=True, render_plugins=[ SwaggerRenderPlugin(), # /docs (default) RedocRenderPlugin(), # /docs/redoc ScalarRenderPlugin(), # /docs/scalar RapidocRenderPlugin(), # /docs/rapidoc StoplightRenderPlugin(), # /docs/stoplight ], exclude_paths=["/admin", "/static"], include_error_responses=True, use_handler_docstrings=True, # use docstrings as descriptions ), )

api = BoltAPI( openapi_config=OpenAPIConfig( title="Acme API", version="2.0.0", description="Acme公司后端API", path="/docs", # 文档访问地址 enabled=True, render_plugins=[ SwaggerRenderPlugin(), # /docs（默认） RedocRenderPlugin(), # /docs/redoc ScalarRenderPlugin(), # /docs/scalar RapidocRenderPlugin(), # /docs/rapidoc StoplightRenderPlugin(), # /docs/stoplight ], exclude_paths=["/admin", "/static"], include_error_responses=True, use_handler_docstrings=True, # 使用函数文档字符串作为描述 ), )

Per-route OpenAPI metadata

单路由OpenAPI元数据

@api.get( "/items", tags=["items"], summary="List all items", description="Returns a paginated list of items with optional filters.", ) async def list_items(): """This docstring is also used as the description if use_handler_docstrings=True.""" return []

undefined

@api.get( "/items", tags=["items"], summary="列出所有商品", description="返回分页的商品列表，支持可选筛选条件", ) async def list_items(): """如果use_handler_docstrings=True，此处的文档字符串也会被用作描述""" return []

undefined

Error handling

错误处理

python

from django_bolt import BoltAPI
from django_bolt.exceptions import HTTPException

api = BoltAPI()


@api.get("/items/{item_id}")
async def get_item(item_id: int):
    item = await fetch_item(item_id)  # your data access logic
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item


@api.post("/items")
async def create_item(name: str, price: float):
    if price < 0:
        raise HTTPException(status_code=422, detail="Price must be positive")
    return {"id": 1, "name": name, "price": price}

python

from django_bolt import BoltAPI
from django_bolt.exceptions import HTTPException

api = BoltAPI()


@api.get("/items/{item_id}")
async def get_item(item_id: int):
    item = await fetch_item(item_id)  # 你的数据访问逻辑
    if not item:
        raise HTTPException(status_code=404, detail="商品未找到")
    return item


@api.post("/items")
async def create_item(name: str, price: float):
    if price < 0:
        raise HTTPException(status_code=422, detail="价格必须为正数")
    return {"id": 1, "name": name, "price": price}

Testing

测试

python

from django_bolt import BoltAPI
from django_bolt.testing import TestClient

api = BoltAPI()


@api.get("/hello")
async def hello():
    return {"message": "world"}


@api.post("/items")
async def create_item(name: str, price: float):
    return {"name": name, "price": price}

python

from django_bolt import BoltAPI
from django_bolt.testing import TestClient

api = BoltAPI()


@api.get("/hello")
async def hello():
    return {"message": "world"}


@api.post("/items")
async def create_item(name: str, price: float):
    return {"name": name, "price": price}

Basic usage

基础用法

with TestClient(api) as client: # GET request response = client.get("/hello") assert response.status_code == 200 assert response.json() == {"message": "world"}

# POST with JSON body
response = client.post("/items", json={"name": "Widget", "price": 9.99})
assert response.status_code == 200
assert response.json()["name"] == "Widget"

# Custom headers
response = client.get("/hello", headers={"X-Request-ID": "test-123"})
assert response.status_code == 200

# Test 404
response = client.get("/nonexistent")
assert response.status_code == 404

undefined

with TestClient(api) as client: # GET请求 response = client.get("/hello") assert response.status_code == 200 assert response.json() == {"message": "world"}

# 带JSON体的POST请求
response = client.post("/items", json={"name": "Widget", "price": 9.99})
assert response.status_code == 200
assert response.json()["name"] == "Widget"

# 自定义请求头
response = client.get("/hello", headers={"X-Request-ID": "test-123"})
assert response.status_code == 200

# 测试404
response = client.get("/nonexistent")
assert response.status_code == 404

undefined

Router composition (multiple api.py files)

路由组合（多个api.py文件）

python

undefined

python

undefined

users/api.py

from django_bolt import BoltAPI

api = BoltAPI(prefix="/users")

@api.get("/") async def list_users(): return []

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


```python

from django_bolt import BoltAPI

api = BoltAPI(prefix="/users")

@api.get("/") async def list_users(): return []

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


```python

orders/api.py

from django_bolt import BoltAPI

api = BoltAPI(prefix="/orders")

@api.get("/") async def list_orders(): return []


`runbolt` auto-discovers all `api.py` files in installed apps and the project root. Each `api = BoltAPI()` instance is merged into the unified router automatically.

from django_bolt import BoltAPI

api = BoltAPI(prefix="/orders")

@api.get("/") async def list_orders(): return []


`runbolt`会自动发现所有已安装应用和项目根目录下的`api.py`文件，每个`api = BoltAPI()`实例会自动合并到统一路由中。

Step-by-step workflow

分步工作流

Step 1: Classify the request

步骤1：分类需求

Place the task into one bucket before writing code:

First-time setup -- new project or first endpoint
New endpoint or CRUD surface -- adding routes
Auth or permission wiring -- JWT, API key, guards
Docs, responses, pagination, or testing -- polish features
Migration -- porting from FastAPI, DRF, or Django Ninja (see
```
references/migration-playbook.md
```
)

编写代码前先将任务归入以下类别：

首次搭建 -- 新项目或第一个端点
新增端点或CRUD接口 -- 添加路由
认证或权限配置 -- JWT、API密钥、守卫
文档、响应、分页或测试 -- 功能打磨
迁移 -- 从FastAPI、DRF或Django Ninja迁移（参考
```
references/migration-playbook.md
```
）

Step 2: Start from the minimal shape

步骤2：从最小结构开始

Add
```
"django_bolt"
```
to
```
INSTALLED_APPS
```
Create an
```
api.py
```
in the project or app directory
Instantiate
```
BoltAPI()
```
Add async route handlers with typed parameters
Run with
```
python manage.py runbolt --dev
```

If the user already has a Django project, preserve their existing models, admin, apps, and settings.

将
```
"django_bolt"
```
添加到
```
INSTALLED_APPS
```
在项目或应用目录下创建
```
api.py
```
实例化
```
BoltAPI()
```
添加带类型化参数的异步路由处理器
运行
```
python manage.py runbolt --dev
```
启动

如果用户已有Django项目，保留其现有模型、admin、应用和配置。

Step 3: Choose the right primitives

步骤3：选择合适的原语

Need	Use
Simple JSON body	`msgspec.Struct`
Rich validation / reusable fields	`Serializer` with `field_validator`
Output shape + docs	`response_model` on the route
Path/query/header/cookie/form/file params	Parameter markers from `django_bolt.param_functions`
Auth	`JWTAuthentication()` , `APIKeyAuthentication()` , or session auth
Permissions	Guards: `IsAuthenticated` , `IsAdminUser` , `IsStaff` , `HasPermission`
CRUD organization	`APIView` or function-based routes
Real-time	`@api.websocket()` with `WebSocket`
One-way live updates	`StreamingResponse(..., media_type="text/event-stream")`
API docs	`OpenAPIConfig` (served at `/docs` by default)
Tests	`django_bolt.testing.TestClient`

需求	实现方案
简单JSON体	`msgspec.Struct`
丰富校验/可复用字段	带 `field_validator` 的 `Serializer`
输出结构+文档	路由上配置 `response_model`
路径/查询/请求头/Cookie/表单/文件参数	`django_bolt.param_functions` 提供的参数标记
认证	`JWTAuthentication()` 、 `APIKeyAuthentication()` 或会话认证
权限	守卫： `IsAuthenticated` 、 `IsAdminUser` 、 `IsStaff` 、 `HasPermission`
CRUD组织	`APIView` 或函数式路由
实时通信	搭配 `WebSocket` 使用 `@api.websocket()`
单向实时更新	`StreamingResponse(..., media_type="text/event-stream")`
API文档	`OpenAPIConfig` （默认在 `/docs` 路径提供）
测试	`django_bolt.testing.TestClient`

Step 4: Write short, copyable examples

步骤4：编写简短可复制的示例

One endpoint per example
One concept per example
Minimal imports
Paste-and-run ready for
```
api.py
```

每个示例对应一个端点
每个示例对应一个概念
最小化导入
可直接粘贴到
```
api.py
```
中运行

Common tasks

常见任务

Add auth

添加认证

JWT:

auth=[JWTAuthentication()]

with guards like

IsAuthenticated()

IsStaff()

HasPermission("app.perm")

API key:
```
auth=[APIKeyAuthentication()]
```
Use
```
create_jwt_for_user(user)
```
to generate tokens
Access the authenticated user via
```
request.user
```
inside handlers

JWT：配置

auth=[JWTAuthentication()]

，搭配

IsAuthenticated()

、

IsStaff()

、

HasPermission("app.perm")

等守卫

API密钥：配置
```
auth=[APIKeyAuthentication()]
```
使用
```
create_jwt_for_user(user)
```
生成令牌
在处理器中通过
```
request.user
```
访问已认证用户

Add validation

添加校验

```
msgspec.Struct
```
for simple JSON bodies -- fields are validated automatically

Serializer

with

@field_validator

@model_validator

for richer checks

Annotated[type, Meta(...)]

for field constraints (

min_length

max_length

ge

le

pattern

)

Parameter markers (
```
Query
```
,
```
Header
```
,
```
Cookie
```
,
```
Form
```
,
```
File
```
,
```
Depends
```
) accept validation kwargs directly
Never manually parse
```
request.body
```
-- use typed parameters

简单JSON体使用
```
msgspec.Struct
```
-- 字段会自动校验

更复杂的校验使用带

@field_validator

@model_validator

的

Serializer

使用

Annotated[type, Meta(...)]

配置字段约束（

min_length

、

max_length

、

ge

、

le

、

pattern

）

参数标记（
```
Query
```
、
```
Header
```
、
```
Cookie
```
、
```
Form
```
、
```
File
```
、
```
Depends
```
）直接支持校验参数
永远不要手动解析
```
request.body
```
-- 使用类型化参数

Add CRUD

添加CRUD

Start with function-based
```
@api.get
```
,
```
@api.post
```
, etc. for small APIs
Use
```
@api.view("/path")
```
with
```
APIView
```
when several methods share one resource path

小型API优先使用函数式
```
@api.get
```
、
```
@api.post
```
等装饰器
多个方法共享同一资源路径时使用
```
@api.view("/path")
```
搭配
```
APIView
```

Add responses and streaming

添加响应和流式处理

Return dicts, lists, or
```
msgspec.Struct
```
for JSON (auto-serialized)
```
JSON(data, status_code=201)
```
for custom status codes
```
Response(data).set_cookie(...)
```
for headers and cookies
```
PlainText
```
,
```
HTML
```
,
```
Redirect
```
,
```
FileResponse
```
for other content types

StreamingResponse(generator(), media_type="text/event-stream")

for SSE with

data: ...\n\n

chunks

Use
```
@no_compress
```
on streaming endpoints

返回字典、列表或
```
msgspec.Struct
```
会自动序列化为JSON
自定义状态码使用
```
JSON(data, status_code=201)
```
配置请求头和Cookie使用
```
Response(data).set_cookie(...)
```
其他内容类型使用
```
PlainText
```
、
```
HTML
```
、
```
Redirect
```
、
```
FileResponse
```

SSE使用

StreamingResponse(generator(), media_type="text/event-stream")

，返回

data: ...\n\n

格式的块

流式端点添加
```
@no_compress
```
装饰器

Add pagination

添加分页

@paginate(PageNumberPagination)

for page-based (

?page=1&page_size=20

)

@paginate(LimitOffsetPagination)

for offset-based (

?limit=20&offset=40

)

```
@paginate(CursorPagination)
```
for cursor-based (
```
?cursor=abc
```
)

Subclass to customize

page_size

max_page_size

page_size_query_param

ordering

页码分页（

?page=1&page_size=20

）使用

@paginate(PageNumberPagination)

偏移量分页（

?limit=20&offset=40

）使用

@paginate(LimitOffsetPagination)

游标分页（
```
?cursor=abc
```
）使用
```
@paginate(CursorPagination)
```

子类化可自定义

page_size

、

max_page_size

、

page_size_query_param

、

ordering

Add OpenAPI docs

添加OpenAPI文档

Pass

openapi_config=OpenAPIConfig(title="...", version="...")

BoltAPI()

Use
```
tags
```
,
```
summary
```
,
```
description
```
,
```
response_model
```
on route decorators

Choose render plugin:

SwaggerRenderPlugin

RedocRenderPlugin

ScalarRenderPlugin

, etc.

Docs served at
```
/docs
```
by default (configurable via
```
path
```
)

给

BoltAPI()

传入

openapi_config=OpenAPIConfig(title="...", version="...")

在路由装饰器上使用
```
tags
```
、
```
summary
```
、
```
description
```
、
```
response_model
```

选择渲染插件：

SwaggerRenderPlugin

、

RedocRenderPlugin

、

ScalarRenderPlugin

等

默认在
```
/docs
```
路径提供文档（可通过
```
path
```
配置）

Add tests

添加测试

Use
```
TestClient(api)
```
as a context manager

Supports

.get()

.post()

.put()

.patch()

.delete()

with

json=

headers=

cookies=

Assert
```
response.status_code
```
,
```
.json()
```
,
```
.headers
```
Cover validation failures (422) and auth failures (401/403)

使用
```
TestClient(api)
```
作为上下文管理器

支持

.get()

、

.post()

、

.put()

、

.patch()

、

.delete()

方法，可传入

json=

、

headers=

、

cookies=

参数

校验
```
response.status_code
```
、
```
.json()
```
、
```
.headers
```
覆盖校验失败（422）和认证失败（401/403）场景

Troubleshooting

故障排查

"Module django_bolt not found"

CRITICAL: Ensure

"django_bolt"

is in

INSTALLED_APPS

settings.py

. The package is

django-bolt

(hyphen) but the Python module is

django_bolt

(underscore).

关键提示：确保

settings.py

的

INSTALLED_APPS

中包含

"django_bolt"

。包名为

django-bolt

（连字符），但Python模块名为

django_bolt

（下划线）。

Server won't start with runbolt

使用runbolt无法启动服务器

Verify

python manage.py runbolt --dev

is being used, not

uvicorn

gunicorn

Check that the Rust extension is built: run
```
just build
```
from the project root
Ensure
```
api.py
```
exists in the Django project root or in an installed app directory

确认使用的是

python manage.py runbolt --dev

启动，而非

uvicorn

或

gunicorn

检查Rust扩展是否已构建：在项目根目录运行
```
just build
```
确保
```
api.py
```
存在于Django项目根目录或已安装的应用目录中

Route not found (404)

路由未找到（404）

Confirm
```
api.py
```
is in a directory
```
runbolt
```
discovers: project root (same dir as
```
settings.py
```
) or an installed app
Verify the
```
api
```
variable is a
```
BoltAPI()
```
instance named exactly
```
api
```
Check route path matches (leading slash required:
```
/items
```
, not
```
items
```
)

确认
```
api.py
```
位于
```
runbolt
```
可发现的目录：项目根目录（与
```
settings.py
```
同目录）或已安装的应用目录
确认
```
api
```
变量是名为
```
api
```
的
```
BoltAPI()
```
实例
检查路由路径是否匹配（需要前导斜杠：
```
/items
```
，而非
```
items
```
）

Auth returns 401 unexpectedly

认证意外返回401

For JWT: verify token is passed in
```
Authorization: Bearer <token>
```
header
For JWT: check token expiration and secret key configuration
For session auth: ensure Django session and auth middleware are enabled
For guards: verify the user object has the required permissions/attributes

JWT场景：确认令牌通过
```
Authorization: Bearer <token>
```
请求头传递
JWT场景：检查令牌过期时间和密钥配置
会话认证场景：确认Django会话和认证中间件已启用
守卫场景：确认用户对象具备所需的权限/属性

References

参考资料

Read
```
references/migration-playbook.md
```
when porting from FastAPI, DRF, or Django Ninja

从FastAPI、DRF或Django Ninja迁移时请阅读
```
references/migration-playbook.md
```