Nexios

Appearance

🔌 WebSockets ​

python
from nexios import NexiosApp
from nexios.routing import WebsocketRoute
app = NexiosApp()
@app.ws_route("/ws")
async def ws_handler(ws):
    await ws.accept()
    ...

Websocket routing follows the same pattern as other http routes making it easy to use.

WebSocket Lifecycle

  1. Connection: Client initiates WebSocket connection
  2. Acceptance: Server accepts the connection
  3. Communication: Bidirectional message exchange
  4. Disconnection: Connection closes (graceful or abrupt)
  5. Cleanup: Resources are cleaned up

Websocket also provides a WebsocketRoute class for more complex routing needs

You can use it like this

python
from nexios import NexiosApp
from nexios.routing import WebsocketRoute
app = NexiosApp()
async def ws_handler(ws):
    await ws.accept()
    ...
app.add_ws_route(WebsocketRoute("/ws", ws_handler))

WebSocket Router ​

Use a Router to group WebSocket routes:

python
from nexios.routing import Router

router = Router(prefix="/ws")
router.add_ws_route(path="/chat", handler=ws_handler)

app.mount_router(router)

TIP

app.mount_router does not take a prefix. Use Router(prefix="/...") instead.

WARNING

Router.add_ws_route expects either a WebsocketRoute instance or keyword args (path=..., handler=...).

📤 Sending Messages ​

the WebSocket class has some methods that can be used to send messages to a connected client.

python
from nexios.websockets.base import WebSocket

async def ws_handler(ws):
    await ws.accept()
    await ws.send_text("Hello World")
    # await ws.send_json({"message": "Hello World"})
    # await ws.send_bytes(b"Hello World")

📥 Receiving Messages ​

The WebSocket class has some methods that can be used to receive messages from a connected client.

python
from nexios.websockets.base import WebSocket

async def ws_handler(ws):
    await ws.accept()
    message = await ws.receive_text()
    # message = await ws.receive_json()
    # message = await ws.receive_bytes()
    print(message)

Nexios supports three primary message formats:

1. Text Messages

python
text_data = await ws.receive_text()
await ws.send_text(f"Received: {text_data}")

2. Binary Messages

python
binary_data = await ws.receive_bytes()
await ws.send_bytes(binary_data)  # Echo binary data

3. JSON Messages

python
json_data = await ws.receive_json()
await ws.send_json({"status": "success", "data": json_data})

🔄 Connection Lifecycle ​

A WebSocket connection follows a clear lifecycle:

Accept the Connection

python
await ws.accept()

Receive Messages (Loop) ​

python
while True:
    data = await ws.receive()
    # Process data

🚪 Handle Disconnections ​

python
from nexios.websockets.base import WebSocketDisconnect
except WebSocketDisconnect:
    print("Client disconnected")

Close the Connection ​

python
finally:
    await ws.close()