Skip to content

Declarative WebSocket ViewSets


`fapix.viewsets.websocket` brings RESTful structure to real-time WebSockets via `WebSocketApiView` (for custom non-DB actions) and `WebSocketModelViewSet` (for ORM-backed models).

---

## 1. Interactive Testing with Built-In Tester

Fapix provides an interactive WebSocket tester directly in your browser:

```text
http://localhost:8000/tester

You can test connection handshakes, JWT query parameter authentication, action routing, and inspect inbound/outbound real-time JSON frames.


2. Routing in urls.py

from fastapi import APIRouter, WebSocket
from .views import RealtimeChatWebSocketView, ProductWebSocketViewSet

router = APIRouter()

# Non-DB Custom Action Endpoint
@router.websocket("/ws/chat/realtime/")
async def realtime_chat_websocket_endpoint(websocket: WebSocket):
    await RealtimeChatWebSocketView.handle_connection(websocket)

# DB-backed Model ViewSet Endpoint
@router.websocket("/ws/products/")
async def products_websocket_endpoint(websocket: WebSocket):
    await ProductWebSocketViewSet.handle_connection(websocket)


3. Implementation Example

from fapix.viewsets.websocket import WebSocketApiView, WebSocketModelViewSet, BroadcastScope
from apps.auth.permissions import IsAuthenticated
from .models import Product
from .schemas import InboundChatMessageSchema, ProductSchema


# Non-DB Custom View
class RealtimeChatWebSocketView(WebSocketApiView):
    action_schemas = {"send_message": InboundChatMessageSchema}
    permission_classes = []
    action_permissions = {"send_message": [IsAuthenticated]}
    auto_broadcast = False
    broadcast_scope = BroadcastScope.GLOBAL

    async def action_send_message(self, payload: dict, params: dict = None):
        outbound_data = {
            "room_id": payload.get("room_id"),
            "text": payload.get("text", ""),
        }
        await self.manager.broadcast(
            message={"action": "new_message", "status": "success", "data": outbound_data},
            scope=self.broadcast_scope,
        )
        return outbound_data


# DB-backed Model ViewSet
class ProductWebSocketViewSet(WebSocketModelViewSet):
    model = Product
    schema = ProductSchema
    auto_broadcast = True
    broadcast_self = True
    broadcast_scope = BroadcastScope.GLOBAL


4. Method Overrides & Custom Hooks

class CustomWebSocketView(WebSocketApiView):
    async def authenticate_websocket(self, websocket: WebSocket):
        token = websocket.query_params.get("token")
        if not token:
            await websocket.close(code=4001, reason="Unauthorized")
            return None
        user = await verify_jwt_token(token)
        websocket.state.user = user
        return user

    async def on_connect(self, websocket: WebSocket):
        print(f"Client connected: {websocket.client}")

    async def on_disconnect(self, websocket: WebSocket, close_code: int):
        print(f"Client disconnected with code {close_code}")