# Django-Bolt LLM Quick Reference

Django-Bolt is a Rust-powered API framework for Django with 60k+ RPS performance.
For detailed docs, see: docs/README.md

## Installation & Setup

```bash
pip install django-bolt
python manage.py runbolt --host 0.0.0.0 --port 8000

# Development mode with auto-reload
python manage.py runbolt --dev

# Production multi-process
python manage.py runbolt --processes 4 --workers 1
```

## Basic API (Async and Sync)

```python
from django_bolt import BoltAPI

api = BoltAPI()

# Async handlers (preferred for I/O operations)
@api.get("/hello")
async def hello():
    return {"message": "Hello, World!"}

# Sync handlers (for CPU-bound operations)
@api.get("/compute")
def compute():
    return {"result": 42}
```


## Route Parameters

```python
# Path parameters - automatically typed
@api.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id}

# Query parameters - use defaults for optional
@api.get("/search")
async def search(q: str, page: int = 1):
    return {"query": q, "page": page}

# Headers, cookies, form data
from django_bolt.params import Header, Cookie, Form

@api.get("/protected")
async def protected(x_api_key: str = Header("X-API-Key")):
    return {"key": x_api_key}
```

## Serializers (Request/Response Validation)

Use `Serializer` for validation with field validators (5-10x faster than Pydantic):

```python
from django_bolt.serializers import Serializer, field_validator

class UserCreate(Serializer):
    name: str
    email: str

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

class UserResponse(Serializer):
    id: int
    name: str

@api.post("/users", response_model=UserResponse)
async def create_user(data: UserCreate):
    user = await User.objects.acreate(**data.to_dict())
    return UserResponse.from_model(user)
```

Key methods:
- `from_model(obj)` - Convert Django model to serializer
- `to_dict()` - Convert serializer to dict
- `@field_validator('field')` - Single field validation
- `@model_validator` - Cross-field validation

Invalid requests return 422 with validation errors. See docs/SERIALIZERS.md for details.

## Response Types

```python
from django_bolt.responses import PlainText, HTML, Redirect, FileResponse, StreamingResponse

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

@api.get("/download")
async def download():
    return FileResponse("/path/to/file.pdf")  # Streaming, for large files

@api.get("/events")
async def sse():
    async def generate():
        for i in range(10):
            yield f"data: Event {i}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")
```

## Django ORM (Async and Sync)

```python
# Async ORM (preferred for async handlers)
@api.get("/async/users/{user_id}")
async def get_user_async(user_id: int):
    user = await User.objects.aget(pk=user_id)
    return {"id": user.id, "name": user.username}

@api.get("/async/users")
async def list_users_async():
    users = []
    async for user in User.objects.all():
        users.append({"id": user.id, "name": user.username})
    return users

# Sync ORM (for sync handlers)
@api.get("/sync/users/{user_id}")
def get_user_sync(user_id: int):
    user = User.objects.get(pk=user_id)
    return {"id": user.id, "name": user.username}

@api.post("/sync/users")
def create_user_sync(user: UserCreate):
    new_user = User.objects.create(username=user.name, email=user.email)
    return {"id": new_user.id}
```

## Authentication & JWT Tokens

### Creating JWT Tokens (Login)

```python
from django_bolt.auth.jwt_utils import create_jwt_for_user
from django.contrib.auth import authenticate
from django_bolt.exceptions import Unauthorized

# Login endpoint that issues JWT tokens
@api.post("/auth/login")
async def login(credentials: LoginCredentials):
    # Authenticate user
    user = await authenticate(username=credentials.username, password=credentials.password)
    if not user:
        raise Unauthorized(detail="Invalid credentials")

    # Create JWT token
    token = create_jwt_for_user(
        user=user,
        secret="your-secret-key",
        expires_delta=timedelta(days=7)
    )

    return {"access_token": token, "token_type": "bearer"}

# Or create token manually with jwt library
import jwt
import time

@api.post("/auth/token")
async def create_token(user_id: int):
    payload = {
        "sub": str(user_id),  # Subject claim (user identifier)
        "user_id": str(user_id),
        "exp": int(time.time()) + 3600,  # Expires in 1 hour
        "iat": int(time.time()),  # Issued at
        # Optional claims:
        "is_staff": False,
        "is_superuser": False,
        "permissions": ["read", "write"],
    }
    token = jwt.encode(payload, "your-secret", algorithm="HS256")
    return {"token": token}
```

### Validating JWT Tokens (Protected Routes)

```python
from django_bolt.auth import JWTAuthentication, IsAuthenticated, APIKeyAuthentication

# JWT Authentication with automatic user loading
jwt_auth = JWTAuthentication(secret="your-secret")

@api.get("/me", auth=[jwt_auth], guards=[IsAuthenticated()])
async def get_current_user(request):
    # request.user is automatically loaded from database
    user = request.user
    return {"id": user.id, "username": user.username}

# Access auth context directly
@api.get("/profile", auth=[jwt_auth], guards=[IsAuthenticated()])
async def get_profile(request):
    auth = request.get("auth", {})
    user_id = auth.get("user_id")
    is_staff = auth.get("is_staff", False)
    return {"user_id": user_id, "is_staff": is_staff}

# API Key Authentication
api_key_auth = APIKeyAuthentication(api_keys={"secret-key-123"})

@api.get("/api/data", auth=[api_key_auth], guards=[IsAuthenticated()])
async def protected_data(request):
    return {"data": "sensitive"}

# Works in sync handlers too
@api.get("/sync/me", auth=[jwt_auth], guards=[IsAuthenticated()])
def get_current_user_sync(request):
    user = request.user
    return {"id": user.id}
```

### Token Revocation

```python
from django_bolt.auth.revocation import InMemoryRevocation, DjangoCacheRevocation

# Use in-memory revocation (simple, single-process)
revocation_store = InMemoryRevocation()

jwt_auth = JWTAuthentication(
    secret="your-secret",
    revocation_store=revocation_store
)

# Logout endpoint - revoke token
@api.post("/auth/logout", auth=[jwt_auth], guards=[IsAuthenticated()])
async def logout(request):
    auth = request.get("auth", {})
    jti = auth.get("jti")  # Token ID (include in JWT when creating)
    if jti:
        await revocation_store.revoke(jti)
    return {"message": "Logged out"}

# For multi-process: use DjangoCacheRevocation or DjangoORMRevocation
from django_bolt.auth.revocation import DjangoCacheRevocation
revocation_store = DjangoCacheRevocation()
```

Guards: `IsAuthenticated()`, `IsAdminUser()`, `IsStaff()`, `HasPermission("perm")`, `HasAnyPermission([...])`, `HasAllPermissions([...])`

## Rate Limiting

```python
from django_bolt.middleware import rate_limit

@api.get("/api/data")
@rate_limit(rps=100, burst=200, key="ip")
async def get_data():
    return {"data": "example"}
```

Keys: `"ip"` (default), `"user_id"`, `"api_key"`, or header name.

## CORS

Set in Django settings (recommended):

```python
CORS_ALLOWED_ORIGINS = ["http://localhost:3000", "https://example.com"]
CORS_ALLOW_CREDENTIALS = True
```

## Compression

Enabled by default (brotli with gzip fallback). Disable per-route:

```python
from django_bolt.compression import no_compress

@api.get("/stream")
@no_compress
async def stream():
    ...
```

For config details, see docs/COMPRESSION.md

## Dependency Injection

```python
from django_bolt import Depends

async def get_db_session():
    return await get_session()

@api.get("/items")
async def get_items(session = Depends(get_db_session)):
    return await session.execute(...)
```

## Class-Based Views

```python
from django_bolt.viewsets import ViewSet, ModelViewSet, action
from django_bolt.auth import IsAuthenticated

# Basic ViewSet
@api.viewset("/users")
class UserViewSet(ViewSet):
    async def list(self, request):
        """GET /users"""
        return [...]

    async def retrieve(self, request, pk: int):
        """GET /users/{pk}"""
        return {...}

    async def create(self, request, data: UserCreate):
        """POST /users"""
        return {...}

    async def update(self, request, pk: int, data: UserUpdate):
        """PUT /users/{pk}"""
        return {...}

    async def partial_update(self, request, pk: int, data: UserPartialUpdate):
        """PATCH /users/{pk}"""
        return {...}

    async def destroy(self, request, pk: int):
        """DELETE /users/{pk}"""
        return {"deleted": True}

    @action(methods=["POST"], detail=True)
    async def activate(self, request, pk: int):
        """POST /users/{pk}/activate"""
        return {...}

    @action(methods=["GET"], detail=False)
    async def active(self, request):
        """GET /users/active"""
        return [...]

# ModelViewSet (automatically implements CRUD)
@api.viewset("/articles")
class ArticleViewSet(ModelViewSet):
    queryset = Article.objects.all()
    serializer_class = ArticleSerializer

    # Optional: override default behavior
    async def list(self, request):
        articles = await self.get_queryset().all()
        return await self.serialize_many(articles)

    # Add custom actions
    @action(methods=["POST"], detail=True, guards=[IsAuthenticated()])
    async def publish(self, request, pk: int):
        article = await self.get_object(pk)
        article.published = True
        await article.asave()
        return await self.serialize(article)

# Sync ViewSet handlers
@api.viewset("/sync/items")
class SyncItemViewSet(ViewSet):
    def list(self, request):
        """Sync list handler"""
        items = Item.objects.all()
        return list(items.values())

    def create(self, request, data: ItemCreate):
        """Sync create handler"""
        item = Item.objects.create(**data.to_dict())
        return {"id": item.id}
```

## Pagination

```python
from django_bolt import paginate, PageNumberPagination

@api.get("/articles")
@paginate(PageNumberPagination)
async def list_articles(request):
    return Article.objects.all()

# Usage: GET /articles?page=2&page_size=20
```

Options: `PageNumberPagination`, `LimitOffsetPagination`, `CursorPagination`

## Error Handling

```python
from django_bolt.exceptions import NotFound, BadRequest, Unauthorized

@api.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await User.objects.filter(pk=user_id).afirst()
    if not user:
        raise NotFound(detail=f"User {user_id} not found")
    return user
```

## OpenAPI Documentation

Auto-generated at `/schema` by default:

```python
from django_bolt.openapi import OpenAPIConfig

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

## Testing

```python
from django_bolt.testing import TestClient, WebSocketTestClient
import pytest

# HTTP Testing
def test_hello():
    with TestClient(api) as client:
        response = client.get("/hello")
        assert response.status_code == 200
        assert response.json() == {"message": "Hello, World!"}

# HTTP Testing with stream=True for SSE
def test_sse():
    with TestClient(api, use_http_layer=True) as client:
        response = client.get("/events", stream=True)
        for chunk in response.iter_content(chunk_size=64, decode_unicode=True):
            if chunk:
                assert "data:" in chunk

# WebSocket Testing
@pytest.mark.asyncio
async def test_websocket():
    async with WebSocketTestClient(api, "/ws/echo") as ws:
        await ws.send_text("hello")
        response = await ws.receive_text()
        assert response == "Echo: hello"

# WebSocket with headers/query/cookies
@pytest.mark.asyncio
async def test_ws_with_params():
    headers = {"Authorization": "Bearer token"}
    async with WebSocketTestClient(
        api,
        "/ws/chat/room123",
        query_string="token=abc",
        headers=headers
    ) as ws:
        msg = await ws.receive_text()
        assert "room123" in msg
```

## Logging

```python
from django_bolt.logging import LoggingConfig

api = BoltAPI(
    logging_config=LoggingConfig(
        request_log_fields={"method", "path"},
        response_log_fields={"status_code", "duration"},
        skip_paths={"/health"},
    )
)
```

## Running the Server

```bash
# Development (auto-reload)
python manage.py runbolt --dev

# Production (multi-process)
python manage.py runbolt --host 0.0.0.0 --port 8000 --processes 4 --workers 1
```

## WebSockets

```python
from django_bolt import BoltAPI, WebSocket
from django_bolt.auth import JWTAuthentication, IsAuthenticated
from typing import Annotated
from django_bolt.param_functions import Query, Header, Cookie

api = BoltAPI()

# Basic WebSocket echo
@api.websocket("/ws/echo")
async def echo_handler(websocket: WebSocket):
    await websocket.accept()
    async for message in websocket.iter_text():
        await websocket.send_text(f"Echo: {message}")

# WebSocket with path parameters (with type coercion)
@api.websocket("/ws/chat/{room_id}")
async def chat_handler(websocket: WebSocket, room_id: int):
    await websocket.accept()
    await websocket.send_text(f"Joined room: {room_id}")

# WebSocket with query/header/cookie parameters
@api.websocket("/ws/advanced")
async def advanced_ws(
    websocket: WebSocket,
    token: Annotated[str, Query()],
    auth: Annotated[str, Header()],
    session: Annotated[str, Cookie()]
):
    await websocket.accept()
    await websocket.send_json({"token": token, "auth": auth})

# Protected WebSocket with guards
@api.websocket(
    "/ws/protected",
    auth=[JWTAuthentication(secret="secret")],
    guards=[IsAuthenticated()]
)
async def protected_ws(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_text("authenticated")

# JSON, binary, and close handling
async for data in websocket.iter_json():  # JSON messages
    await websocket.send_json({"received": data})

async for data in websocket.iter_bytes():  # Binary messages
    await websocket.send_bytes(data)

await websocket.close(code=1000, reason="done")  # Close connection
```

WebSocket features:
- Path parameters with type coercion (int, float, bool, str)
- Query/Header/Cookie parameter injection
- Guards and authentication (runs in Rust)
- Text, JSON, and binary message handling
- Graceful close with custom codes

## Server-Sent Events (SSE)

```python
from django_bolt import StreamingResponse
import asyncio

# Async SSE
@api.get("/events")
async def sse_events():
    async def generate():
        for i in range(10):
            yield f"data: Event {i}\n\n"
            await asyncio.sleep(1)
    return StreamingResponse(generate(), media_type="text/event-stream")


## Key Points

- Handlers can be `async def` (preferred) or `def` (for CPU-bound work)
- Use `Serializer` for request/response schemas with validation
- Use Django async ORM in async handlers: `aget()`, `acreate()`, `afilter()`, `async for`
- Use Django sync ORM in sync handlers: `get()`, `create()`, `filter()`
- Auth/guards run in Rust (no Python GIL overhead)
- Compression enabled by default
- OpenAPI docs at `/schema`
- WebSockets support path/query/header/cookie parameters
- SSE works with both async and sync generators
