django-bolt

Django-Bolt

关键规则

• 除非用户明确需要同步处理，否则始终使用异步处理器
• 在异步处理器中使用异步数据库访问（例如 Django 异步 ORM：aget、acreate、afilter）
• 对于简单的类型化负载，使用 msgspec.Struct；对于更丰富的验证和可复用字段集，使用 Serializer
• 为路由签名添加类型标注——让 Django-Bolt 验证输入，而不是手动解析 request.body
• 对于预期的 API 错误，使用内置 HTTP 异常（HTTPException）
• 开发时使用 python manage.py runbolt --dev 运行，而不是使用 uvicorn 或 gunicorn

在以下情况下应用此技能

• 用户提及 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

在以下情况下不要应用

• 用户使用的是普通 Django 视图、模板或 Django 管理后台
• 用户在构建标准 DRF API，且未提及 django-bolt
• 任务仅涉及 Django ORM 模型、迁移或管理后台配置

代码示例

最简配置

# settings.py
INSTALLED_APPS = [
    "django_bolt",
]

# myproject/api.py
from django_bolt import BoltAPI

api = BoltAPI()


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

运行：python manage.py runbolt --dev

BoltAPI 构造函数选项

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

路由装饰器和参数

所有 HTTP 方法：@api.get、@api.post、@api.put、@api.patch、@api.delete、@api.head、@api.options。

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

类型化请求正文（msgspec.Struct）

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 进行验证

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}

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

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


# Cookie extraction
@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
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)}


# 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"]}


# Explicit JSON body marker
class UpdatePayload(msgspec.Struct):
    name: str
    value: int


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

响应类型

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
@api.get("/auto-json")
async def auto_json():
    return {"message": "hello"}


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


# Response with custom headers and cookies
@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
@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


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

SSE（服务器发送事件）

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

WebSocket

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

身份验证和防护机制

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
@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}


# JWT-protected endpoint
@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.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"}


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

中间件（CORS、速率限制、自定义）

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

api = BoltAPI()


# Per-route 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
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}

分页

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

api = BoltAPI()


# Page-number pagination (default)
# 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


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


# Limit-offset pagination
# 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
class FeedPagination(CursorPagination):
    page_size = 20
    ordering = "-created_at"


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

OpenAPI 文档

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

# Minimal -- serves Swagger UI at /docs
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
    ),
)


# Per-route OpenAPI metadata
@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 []

错误处理

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}

测试

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

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

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

# orders/api.py
from django_bolt import BoltAPI

api = BoltAPI(prefix="/orders")

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

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

分步工作流

第 1 步：对请求分类

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

1. 首次配置——新项目或第一个端点
2. 新端点或 CRUD 接口——添加路由
3. 身份验证或权限配置——JWT、API 密钥、防护机制
4. 文档、响应、分页或测试——完善功能
5. 迁移——从 FastAPI、DRF 或 Django Ninja 移植（参见 references/migration-playbook.md）

第 2 步：从最简结构开始

1. 将 "django_bolt" 添加到 INSTALLED_APPS
2. 在项目或应用目录中创建 api.py
3. 实例化 BoltAPI()
4. 添加带有类型化参数的异步路由处理器
5. 使用 python manage.py runbolt --dev 运行

如果用户已有 Django 项目，请保留其现有模型、管理后台、应用和设置。

第 3 步：选择合适的基本组件

需求	使用
简单 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

第 4 步：编写简短、易于复制的示例

• 每个示例只包含一个端点
• 每个示例只介绍一个概念
• 尽量减少导入
• 可直接粘贴到 api.py 中运行

常见任务

添加身份验证

• JWT：使用 auth=[JWTAuthentication()]，并配合 IsAuthenticated()、IsStaff()、HasPermission("app.perm") 等守卫
• API 密钥：使用 auth=[APIKeyAuthentication()]
• 使用 create_jwt_for_user(user) 生成令牌
• 在处理函数中通过 request.user 访问已通过身份验证的用户

添加验证

• 简单的 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——请使用带类型的参数

添加 CRUD

• 对于小型 API，从基于函数的 @api.get、@api.post 等开始
• 当多个方法共用一个资源路径时，使用 @api.view("/path") 和 APIView

添加响应和流式传输

• 返回字典、列表或 msgspec.Struct 以生成 JSON（会自动序列化）
• 使用 JSON(data, status_code=201) 设置自定义状态码
• 使用 Response(data).set_cookie(...) 设置标头和 Cookie
• 对于其他内容类型，使用 PlainText、HTML、Redirect、FileResponse
• 使用 StreamingResponse(generator(), media_type="text/event-stream") 通过 data: ...\n\n 数据块实现 SSE
• 在流式传输端点上使用 @no_compress

添加分页

• 使用 @paginate(PageNumberPagination) 实现基于页码的分页（?page=1&page_size=20）
• 使用 @paginate(LimitOffsetPagination) 实现基于偏移量的分页（?limit=20&offset=40）
• 使用 @paginate(CursorPagination) 实现基于游标的分页（?cursor=abc）
• 创建子类以自定义 page_size、max_page_size、page_size_query_param、ordering

添加 OpenAPI 文档

• 将 openapi_config=OpenAPIConfig(title="...", version="...") 传递给 BoltAPI()
• 在路由装饰器上使用 tags、summary、description、response_model
• 选择渲染插件：SwaggerRenderPlugin、RedocRenderPlugin、ScalarRenderPlugin 等
• 默认在 /docs 提供文档（可通过 path 配置）

添加测试

• 将 TestClient(api) 用作上下文管理器
• 支持使用 json=、headers=、cookies= 调用 .get()、.post()、.put()、.patch()、.delete()
• 断言 response.status_code、.json()、.headers
• 覆盖验证失败（422）和身份验证失败（401/403）

故障排除

“找不到模块 django_bolt”

重要：确保 settings.py 中的 INSTALLED_APPS 包含 "django_bolt"。包名是 django-bolt（连字符），但 Python 模块名是 django_bolt（下划线）。

使用 runbolt 时服务器无法启动

1. 确认使用的是 python manage.py runbolt --dev，而非 uvicorn 或 gunicorn
2. 检查 Rust 扩展是否已构建：在项目根目录运行 just build
3. 确保 api.py 位于 Django 项目根目录或已安装的应用目录中

找不到路由（404）

1. 确认 api.py 位于 runbolt 能发现的目录中：项目根目录（与 settings.py 同一目录）或已安装的应用目录
2. 确认 api 变量是名为 api 的 BoltAPI() 实例
3. 检查路由路径是否匹配（必须以斜杠开头：/items，而不是 items）

身份验证意外返回 401

1. 对于 JWT：确认令牌通过 Authorization: Bearer <token> 标头传递
2. 对于 JWT：检查令牌是否过期，以及密钥配置是否正确
3. 对于会话身份验证：确保已启用 Django 会话和身份验证中间件
4. 对于守卫：确认用户对象具备所需的权限／属性

参考资料

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