Django-Bolt

Django-Bolt uses two validation layers and collects all errors into a single RequestValidationError (similar to Pydantic) when using model_validate() or model_validate_json():

1. Meta constraints: Declarative constraints defined using Annotated and msgspec.Meta (e.g., min_length, pattern, ge). These are high-performance and handled by msgspec.
2. Custom validators: Logic defined via @field_validator and @model_validator.

Validation Order:

1. msgspec parses and validates (Type checking + Meta constraints).
2. @field_validator runs.
3. @model_validator runs.
4. All errors from all layers are raised together.

from typing import Annotated
from msgspec import Meta
from django_bolt.serializers import Serializer, field_validator

class UserSerializer(Serializer):
    # Layer 1: Meta constraints
    name: Annotated[str, Meta(min_length=2)]
    email: str

    # Layer 2: Custom field validator
    @field_validator("email")
    def validate_email(cls, value):
        if "@" not in value:
            raise ValueError("Invalid email")
        return value

Real-time communication often requires sending messages to multiple clients.

Broadcast to all clients

Maintain a global set of connected WebSocket objects. When a message is received, iterate through the set and call send_text on each client. Use a finally block to ensure clients are removed from the set when they disconnect.

Room-based chat

Maintain a mapping of room_id to a set of WebSocket objects. When a client connects to a specific room path, add them to that room's set. Broadcast messages only to the clients within that specific room.

# Broadcast pattern
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)

# Room-based pattern
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 uses a dependency injection system via the Depends marker. A dependency is any callable (a function or a class) that returns a value. When a handler parameter is annotated with Depends(), Django-Bolt performs the following lifecycle:

1. Calls the dependency function: Executes the callable.
2. Passes the result to the handler: Injects the return value into the endpoint function.
3. Caches the result: By default, the result is cached for the duration of the same request. If the same dependency is requested multiple times within one request, the cached result is reused instead of re-executing the logic.

Dependencies can be synchronous or asynchronous, and they can be nested (one dependency depending on another).

from django_bolt import BoltAPI, Depends

api = BoltAPI()

async def get_current_user(request):
    # ... logic ...
    return user

@api.get("/profile")
async def get_profile(user=Depends(get_current_user)):
    return {"id": user.id, "username": user.username}

When request data fails validation, Django-Bolt raises a RequestValidationError which returns a 422 Unprocessable Entity status.

Django-Bolt collects all errors from both Meta constraints and custom @field_validator/@model_validator functions before raising the exception, allowing for multi-error reporting.

Accessing errors programmatically:

try:
    serializer = MySerializer(email="bad", password="short")
except RequestValidationError as e:
    all_errors = e.errors()  # Returns list of error dicts
    print(str(e))  # "body.email: Invalid email; body.password: Too short"

Response Format:

{
    "detail": [
        {
            "loc": ["body", "email"],
            "msg": "Invalid email",
            "type": "value_error"
        },
        {
            "loc": ["body", "password"],
            "msg": "Password too short",
            "type": "value_error"
        }
    ]
}

Recycling is spawn-first and graceful, making it safe for long-lived connections like WebSockets:

1. Replacement First: The supervisor forks a replacement worker before stopping the old one. SO_REUSEPORT allows both to share the port.
2. Draining: The old worker receives SIGTERM and stops accepting new connections. The kernel routes new traffic to healthy workers.
3. WebSocket Handling: Active WebSocket connections receive a proper close frame with code 1012 (Service Restart). Clients must be built with reconnection logic to handle this.
4. HTTP Completion: In-flight HTTP requests are allowed to finish up to the --workers-kill-timeout (or the DJANGO_BOLT_SHUTDOWN_TIMEOUT env var, default 30s).
5. Force Kill: If the worker hasn't exited after the timeout, it is SIGKILLed.

Note on WebSockets: No server can migrate a live TCP connection. Use Redis or a database to persist per-connection session state so reconnections are transparent to users.

Django-Bolt's Serializer uses two layers of validation. The behavior regarding error collection depends on how you trigger validation:

1. Meta constraints (e.g., min_length, pattern, ge, le in Meta class).
2. Custom validators (@field_validator, @model_validator).

Error Collection Behavior

Note: Direct instantiation (Serializer(field=value)) runs custom validators but bypasses Meta constraints for performance reasons.

Example of multi-error collection:

try:
    UserSerializer.model_validate({
        "name": "X",        # Meta: min_length=2 violated
        "email": "bad",     # Meta: pattern violated
        "age": 200          # Meta: le=150 violated
    })
except RequestValidationError as e:
    print(len(e.errors()))  # 3 errors, not just 1