Bazis WS

Extension package for Bazis, providing WebSocket connections with authentication support, Redis pub/sub, and user online status tracking.

Quick Start

uv add bazis-ws

# Add mixin to user model
from django.contrib.auth.models import AbstractUser
from bazis.contrib.ws.models_abstract import UserWsMixin
from bazis.core.models_abstract import JsonApiMixin

class User(UserWsMixin, JsonApiMixin, AbstractUser):
    """User with WebSocket support"""
    class Meta:
        verbose_name = 'User'
        verbose_name_plural = 'Users'

# Register the WebSocket route in the router module of the project (BS_BAZIS_ROUTER_MODULE)
from bazis.core.routing import BazisRouter

router = BazisRouter(prefix='/api/v1')
router.register('bazis.contrib.ws.router')  # the socket: /api/v1/ws

# Notify after the commit, in the language of each user (in the hooks of a write)
from bazis.contrib.ws.messages import notification, notify, publish_changed

notify(ticket.author, lambda user: notification(_('Your ticket was answered'), reply.body, ticket))
publish_changed(ticket)  # {"resource": "support.ticket"} to every user session

Table of Contents

• Description
• Requirements
• Installation
• Core Components
  • UserWsMixin
  • WsEndpoint
  • Architecture
• Usage
  • Project Setup
  • Connecting to WebSocket
  • Notifications of bazis-front
  • Sending Messages to Users
  • Checking Online Status
• WebSocket Protocol
• Examples
• License
• Links

Description

Bazis WS is an extension package for the Bazis framework that provides a fully-featured WebSocket communication system. The package includes:

• UserWsMixin — mixin for user model with WebSocket support
• WsEndpoint — ready-to-use WebSocket endpoint with JWT authentication
• Redis Pub/Sub — messaging system between servers and clients
• Online Status Tracking — automatic detection of online/offline users
• Personal Channels — each user has their own channel for receiving messages
• Common Channel — for broadcasting messages to all connected users (not to anonymous sessions unless BS_BAZIS_WS_ANONYMOUS_COMMON_CHANNEL=true)

This package requires installation of bazis and a running Redis server.

Requirements

• Python: 3.12+
• bazis: latest version
• PostgreSQL: 12+
• Redis: For pub/sub and caching
• Additional libraries:
  • PyJWT — for JWT handling
  • psycopg[binary] — for asynchronous PostgreSQL access
  • redis — for Redis operations

Installation

Using uv (recommended)

uv add bazis-ws

Using pip

pip install bazis-ws

Core Components

UserWsMixin

Mixin for user model that adds WebSocket support.

Location: bazis.contrib.ws.models_abstract.UserWsMixin

Properties:

• user_channel — user's personal channel in Redis (format: user_ws::{user_id})
• ws_session — WebSocket session key in Redis (format: user_ws::{user_id}:session)
• is_online — boolean property indicating whether the user is connected to WebSocket

Methods:

• ws_publish(data: dict) -> int — send message to user via their personal channel

Usage Example:

from django.contrib.auth.models import AbstractUser
from bazis.contrib.ws.models_abstract import UserWsMixin
from bazis.core.models_abstract import JsonApiMixin

class User(UserWsMixin, JsonApiMixin, AbstractUser):
    """User with WebSocket support"""
    class Meta:
        verbose_name = 'User'
        verbose_name_plural = 'Users'

WsEndpoint

WebSocket endpoint with authentication and session management support.

Location: bazis.contrib.ws.ws.WsEndpoint

Based on: starlette.endpoints.WebSocketEndpoint

Key Features:

1. JWT Token Authentication:

   • On connection: ws://api.example.com/api/v1/ws?token=<jwt_token>
   • During session: sending {"token": "<jwt_token>"}
   • The token must be a valid, unexpired session token (exp and sub are required) of an active user.

   Anonymous clients send a token they generate themselves instead of a JWT: 16–128 characters A-Z a-z 0-9 _ - (e.g. a random UUID). The token is the only secret protecting the channel: anybody who knows it receives its messages, so it must come from a cryptographic random source (crypto.randomUUID()) and never be shared. It subscribes to the channel user_ws:anon:<token>, which can never be the channel of a user or the common channel. Anybody can open an anonymous session, so by default it does not receive the common channel (see Common Channel).

2. Automatic Online Status Tracking:

   • Status update every 5 seconds
   • Redis entry TTL: 10 seconds

3. Channel Subscription:

   • User's personal channel (or the channel of the anonymous token)
   • Common channel: user sessions; anonymous sessions only with BS_BAZIS_WS_ANONYMOUS_COMMON_CHANNEL=true
   • {"type": "subscribed"} is sent once the subscription is active: the messages published from then on are delivered (pub/sub keeps nothing published before)

4. Ping/Pong:

   • Client sends {"type": "ping"}
   • Server responds {"type": "pong"}

Connection Lifecycle:

1. on_connect()      → accept connection
2. session_start()   → authenticate and start background tasks
   ├─ task_online_update()    → update online status
   └─ task_listen_queue()     → listen to Redis channels
3. on_receive()      → handle incoming messages
4. on_disconnect()   → cleanup resources
5. session_stop()    → stop background tasks

Architecture

┌─────────────┐                    ┌──────────────┐
│   Client    │◄──WebSocket───────►│  WsEndpoint  │
│ (Browser/   │                    │              │
│  Mobile)    │                    │  Starlette   │
└─────────────┘                    └───────┬──────┘
                                           │
                                           │ JWT Auth
                                           ▼
                                    ┌──────────────┐
                                    │  PostgreSQL  │
                                    │  (User DB)   │
                                    └──────────────┘
                                           │
                                           │
                                           ▼
┌─────────────┐                    ┌──────────────┐
│   Backend   │────publish────────►│    Redis     │
│   Service   │                    │   Pub/Sub    │
└─────────────┘                    └───────┬──────┘
                                           │
                                           │ subscribe
                                           ▼
                                    ┌──────────────┐
                                    │  WsEndpoint  │
                                    │              │
                                    └───────┬──────┘
                                           │
                                           │ send_json
                                           ▼
                                    ┌──────────────┐
                                    │   Client     │
                                    └──────────────┘

Redis Channels:

• user_ws::{user_id} — user's personal channel
• user_ws:common — common channel of the user sessions (and of the anonymous sessions with BS_BAZIS_WS_ANONYMOUS_COMMON_CHANNEL=true)
• user_ws::{user_id}:session — active session key (TTL: 10 seconds)
• user_ws:anon:{token} — channel of an anonymous client (and ...:session, its session key)

A session that cannot run closes the socket (see Close Codes): code 1008 after a refused token, code 1011 when the server fails (the session start, or Redis while the session is running) so that the client reconnects.

Usage

Project Setup

1. Add mixin to user model:

# models.py
from django.contrib.auth.models import AbstractUser
from bazis.contrib.ws.models_abstract import UserWsMixin
from bazis.core.models_abstract import JsonApiMixin

class User(UserWsMixin, JsonApiMixin, AbstractUser):
    class Meta:
        verbose_name = 'User'
        verbose_name_plural = 'Users'

2. Add is_online field to user routes:

# routes.py
from bazis.contrib.ws.routes_abstract import UserWsRouteSet
from django.apps import apps

class UserRouteSet(UserWsRouteSet):
    model = apps.get_model('myapp.User')

3. Register WebSocket route in the router module of the project, the module of BS_BAZIS_ROUTER_MODULE:

# router.py
from bazis.core.routing import BazisRouter

router = BazisRouter(prefix='/api/v1')
router.register('bazis.contrib.ws.router')  # the socket: /api/v1/ws

router.register('myapp.router')

The socket is routed under the prefix of the router, like the other routes: /api/v1/ws here. Tools import the router module, not the main module of the project: the contract export of bazis-front finds the socket (and its path) there. ws_route (bazis.contrib.ws.ws) remains for a project that routes the socket itself: appended as it is to the routes of the root router (router.routes.append(ws_route)), it keeps the path /ws. Appending it to app.router.routes in the main module also works at runtime, but the tools do not see it.

Connecting to WebSocket

JavaScript Client

class WebSocketClient {
  constructor(url, token) {
    this.url = url;
    this.token = token;
    this.ws = null;
    this.reconnectInterval = 5000;
    this.pingInterval = 30000;
    this.pingTimer = null;
  }

  connect() {
    this.ws = new WebSocket(`${this.url}?token=${this.token}`);

    this.ws.onopen = () => {
      console.log('WebSocket connected');
      this.startPing();
    };

    this.ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      this.handleMessage(data);
    };

    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
    };

    this.ws.onclose = (event) => {
      console.log('WebSocket disconnected');
      this.stopPing();
      // 1008: the token was refused, the same token would be refused again
      if (event.code === 1008) return;
      // Reconnect (use a growing delay in production)
      setTimeout(() => this.connect(), this.reconnectInterval);
    };
  }

  handleMessage(data) {
    switch (data.type) {
      case 'pong':
        console.log('Received pong');
        break;
      case 'subscribed':
        // the messages published from now on are delivered: refetch what may have changed
        console.log('Subscribed');
        break;
      case 'data':
        console.log('Received data:', data.data);
        // Process received data
        this.onData(data.data);
        break;
      case 'error':
        // the server closes the socket after it: 1008 (a refused token) or 1011
        console.error('Error:', data.code, data.detail);
        break;
      default:
        console.log('Unknown message type:', data);
    }
  }

  startPing() {
    this.pingTimer = setInterval(() => {
      if (this.ws.readyState === WebSocket.OPEN) {
        this.ws.send(JSON.stringify({ type: 'ping' }));
      }
    }, this.pingInterval);
  }

  stopPing() {
    if (this.pingTimer) {
      clearInterval(this.pingTimer);
      this.pingTimer = null;
    }
  }

  onData(data) {
    // Override this method to handle data
    console.log('Data received:', data);
  }

  disconnect() {
    this.stopPing();
    if (this.ws) {
      this.ws.close();
    }
  }
}

// Usage
const ws = new WebSocketClient('ws://api.example.com/api/v1/ws', jwtToken);
ws.onData = (data) => {
  console.log('Processing data:', data);
  // Your processing logic
};
ws.connect();

Python Client

import asyncio
import json
import websockets

async def websocket_client(url, token):
    uri = f"{url}?token={token}"
    
    async with websockets.connect(uri) as websocket:
        print("WebSocket connected")
        
        # Background task for ping
        async def send_ping():
            while True:
                await asyncio.sleep(30)
                await websocket.send(json.dumps({"type": "ping"}))
        
        ping_task = asyncio.create_task(send_ping())
        
        try:
            async for message in websocket:
                data = json.loads(message)
                
                if data['type'] == 'pong':
                    print("Received pong")
                elif data['type'] == 'subscribed':
                    print("Subscribed")
                elif data['type'] == 'data':
                    print(f"Received data: {data['data']}")
                elif data['type'] == 'error':
                    print(f"Error: {data['code']} - {data['detail']}")
        finally:
            ping_task.cancel()

# Usage
asyncio.run(websocket_client('ws://api.example.com/api/v1/ws', jwt_token))

Notifications of bazis-front

bazis.contrib.ws.messages publishes the messages that the frontend of bazis-front reads (@/bazis/react/ws) from the code of a write, after the commit of the current transaction (at once outside a transaction) and robustly: a client refetching reads the committed data, a rolled back write publishes nothing, and Redis down does not fail a committed write.

• notify(users, message) — publishes to the channel of each user (a user or an iterable; None and repeated users are skipped) the message built for him by message(user). The message is built and serialized in the language of the user: user.language (UserLanguageMixin of bazis-users) or LANGUAGE_CODE, so lazy strings are translated per recipient. The message is not stored: a user who is not connected misses it. For notifications kept for the user (an inbox with unread ones, read/unread, e-mail), use bazis-notify, which publishes through this socket.
• notification(title, text=None, item=None) — the notification {"action": "notification", "title", "text", "resource", "id"}, about an item (the frontend shows it and refetches the item).
• publish_changed(item) — {"resource": "<JSON:API type>"} on the common channel: every user session refetches the resource with its own permissions. No id: the common channel reaches every user.

from django.utils.translation import gettext_lazy as _

from bazis.contrib.ws.messages import notification, notify, publish_changed


class ReplyRouteSet(...):
    def hook_after_create(self, item):
        super().hook_after_create(item)
        ticket = item.ticket
        notify(
            ticket.author,
            lambda user: notification(
                _('New reply to ticket #%(number)s') % {'number': ticket.uniq_number},
                item.body,
                ticket,
            ),
        )
        publish_changed(ticket)

Send notifications only to the users who may see the item. Publish from the hooks of the write (hook_after_create, hook_after_update, an action of a transit), not from validate_item, which may run more than once for one write.

Sending Messages to Users

user.ws_publish(data) is the low-level publishing: it publishes at once, whatever the transaction, to the sessions connected now. Use it outside a write (a task, code that runs after the commit). From the code of a write use notify and publish_changed (above), which publish after the commit: ws_publish inside a write notifies before the commit, also of a write that is rolled back. A message the user must see later, when he is not connected, is a stored notification of bazis-notify, not a check of is_online.

From Django View or API Endpoint

from django.contrib.auth import get_user_model

User = get_user_model()

def send_notification_to_user(user_id, message):
    """Send notification to specific user"""
    user = User.objects.get(id=user_id)
    
    if user.is_online:
        user.ws_publish({
            'type': 'notification',
            'title': 'New Notification',
            'message': message,
            'timestamp': datetime.now().isoformat()
        })
        return True
    return False

From Celery Task

from celery import shared_task
from django.contrib.auth import get_user_model

User = get_user_model()

@shared_task
def notify_user_async(user_id, notification_data):
    """Asynchronously send notification to user"""
    try:
        user = User.objects.get(id=user_id)
        user.ws_publish({
            'type': 'task_completed',
            'data': notification_data
        })
    except User.DoesNotExist:
        pass

Common Channel

The common channel user_ws:common (bazis.contrib.ws.COMMON_CHANNEL) reaches every user session, whatever the user may see. Publish there only what every user may know: that a resource changed, {"resource": "<JSON:API type>"}, without the id of the item (the format of bazis-front, whose pages then refetch the resource with their own permissions). Send the ids, the notifications and any data of an item only to the users who may see it, with user.ws_publish(...).

from bazis.contrib.ws.messages import publish_changed

publish_changed(item)  # after the commit: {"resource": "<JSON:API type of the item>"}

Anonymous sessions do not receive the common channel: an anonymous token is generated by the client, so anybody can open such a session. Set BS_BAZIS_WS_ANONYMOUS_COMMON_CHANNEL=true only if what the project publishes on the common channel is public (for example, pages that anonymous visitors see refresh on the {"resource"} messages).

Checking Online Status

In Django Template

from django.contrib.auth import get_user_model

User = get_user_model()

def user_list_view(request):
    users = User.objects.all()
    
    online_users = [user for user in users if user.is_online]
    offline_users = [user for user in users if not user.is_online]
    
    return render(request, 'users.html', {
        'online_users': online_users,
        'offline_users': offline_users
    })

Via API (using UserWsRouteSet)

GET /api/v1/<app>/<resource>/
Authorization: Bearer <token>

Response:

{
  "data": [
    {
      "type": "app.user",
      "id": "123",
      "attributes": {
        "username": "john_doe",
        "email": "john@example.com",
        "is_online": true
      }
    }
  ]
}

WebSocket Protocol

Messages from Client

Ping

{
  "type": "ping"
}

Authentication During Session

{
  "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
}

Messages from Server

Subscribed

Sent once the subscription of the session is active (after each accepted token): the messages published from then on are delivered.

{
  "type": "subscribed"
}

Pong

{
  "type": "pong"
}

Data

data is the published message as a JSON string, decoded by the client:

{
  "type": "data",
  "data": "{\"type\": \"notification\", \"title\": \"New Message\", \"message\": \"You have a new message from admin\"}"
}

Error

{
  "type": "error",
  "code": "expired_token",
  "detail": "Token expired"
}

Error Codes:

• expired_token — JWT token has expired
• invalid_token — the JWT is invalid, or the anonymous token does not match the format
• user_not_found — user not found in database or inactive
• internal_error — the session could not be started

Close Codes

The server closes the socket of a session that cannot run, after the error message if there is one:

• 1008 (policy violation) — the token was refused (expired_token, invalid_token, user_not_found): connect again only with another token (a refreshed session token).
• 1011 (internal error) — the session could not be started (internal_error), or Redis failed while it was running: reconnect with a growing delay.

The messages that the client sends after the server closed the socket are ignored.

Examples

Real-time Chat Example

Backend (sending message):

from django.contrib.auth import get_user_model
from django.http import JsonResponse
from django.views.decorators.http import require_POST
import json

User = get_user_model()

@require_POST
def send_message(request):
    data = json.loads(request.body)
    recipient_id = data.get('recipient_id')
    message = data.get('message')
    
    try:
        recipient = User.objects.get(id=recipient_id)
        
        # Send message via WebSocket (the view writes nothing: a view that saves the
        # message publishes it after the commit, and stores what an offline recipient
        # must see later with bazis-notify)
        recipient.ws_publish({
            'type': 'chat_message',
            'sender': {
                'id': str(request.user.id),
                'username': request.user.username
            },
            'message': message,
            'timestamp': datetime.now().isoformat()
        })
        return JsonResponse({'status': 'sent'})
            
    except User.DoesNotExist:
        return JsonResponse({'error': 'User not found'}, status=404)

Frontend (receiving message):

class ChatClient extends WebSocketClient {
  onData(data) {
    if (data.type === 'chat_message') {
      this.displayMessage(data.sender, data.message, data.timestamp);
    }
  }

  displayMessage(sender, message, timestamp) {
    const messageElement = document.createElement('div');
    messageElement.className = 'chat-message';
    messageElement.innerHTML = `
      <div class="sender">${sender.username}</div>
      <div class="message">${message}</div>
      <div class="timestamp">${new Date(timestamp).toLocaleString()}</div>
    `;
    document.getElementById('chat-messages').appendChild(messageElement);
  }
}

const chat = new ChatClient('ws://api.example.com/api/v1/ws', jwtToken);
chat.connect();

Task Notification Example

Celery Task:

from celery import shared_task
from django.contrib.auth import get_user_model

User = get_user_model()

@shared_task
def process_long_running_task(user_id, task_data):
    """Long-running task with user notification"""
    user = User.objects.get(id=user_id)
    
    # Notify about start
    user.ws_publish({
        'type': 'task_started',
        'task_id': process_long_running_task.request.id,
        'message': 'Processing started...'
    })
    
    try:
        # Execute task
        result = perform_processing(task_data)
        
        # Notify about success
        user.ws_publish({
            'type': 'task_completed',
            'task_id': process_long_running_task.request.id,
            'result': result,
            'message': 'Processing completed successfully'
        })
        
    except Exception as e:
        # Notify about error
        user.ws_publish({
            'type': 'task_failed',
            'task_id': process_long_running_task.request.id,
            'error': str(e),
            'message': 'An error occurred during processing'
        })

Online Indicator Example

JavaScript Component:

class OnlineIndicator {
  constructor(userId) {
    this.userId = userId;
    this.indicator = document.getElementById(`user-${userId}-status`);
  }

  async checkStatus() {
    const response = await fetch(`/api/v1/users/user/${this.userId}/`, {
      headers: {
        'Authorization': `Bearer ${token}`
      }
    });
    
    const data = await response.json();
    const isOnline = data.data.attributes.is_online;
    
    this.updateIndicator(isOnline);
  }

  updateIndicator(isOnline) {
    if (isOnline) {
      this.indicator.classList.add('online');
      this.indicator.classList.remove('offline');
      this.indicator.textContent = 'Online';
    } else {
      this.indicator.classList.add('offline');
      this.indicator.classList.remove('online');
      this.indicator.textContent = 'Offline';
    }
  }
}

// Periodic status check
const indicator = new OnlineIndicator('user-123');
setInterval(() => indicator.checkStatus(), 10000);

License

Apache License 2.0

See LICENSE file for details.

Links

• Bazis Documentation — main repository
• Bazis WS Repository — package repository
• Issue Tracker — report bugs or request features
• Starlette WebSockets — Starlette WebSocket documentation
• Redis Pub/Sub — Redis Pub/Sub documentation

Support

If you have questions or issues:

• Check the Bazis documentation
• Search through existing issues
• Create a new issue with detailed information

---

Made with ❤️ by Bazis team