Builtin Services: WebSockets

https://github.com/xellu/nautica-api/tree/main/nautica/services/builtins/websocket

The WebSocket API is provided by napi.ws. It is built on top of HTTP API. Like HTTP API, it also runs on Starlette + Uvicorn, and similarly uses file-based routing, derived from file's location under src/ws/.

from napi.ws import WS, Context, Error

Note that Require and Reply from HTTP API aren't available.

Routing

File-based Routes

Routes are defined in .py files under src/ws/. The URL path is derived from the file path:

File	Path
src/ws/users.py	/users
src/ws/api/v1/auth.py	/api/v1/auth
src/ws/api/v1/auth/+root.py	/api/v1/auth

+root.py acts as the index file for a directory

Defining an Endpoint

Decorators such as: @WS.OnConnect(), @WS.OnDisconnect() and @WS.OnPacket(...), are used to handle websocket connections, define at least one to create an endpoint.

@WS.OnPacket("hello")
async def on_hello(ctx: Context, data: any):
    return "Hello World!" # This can be anything JSON-Serializable

This returns a structured packet, wrapping your response:

{
    "id": "hello",
    "data": "Hello World!" // From the return statement
}

Packet Structure

A simple packet looks like this:

{
    "id": "hello", // This is the identifier, which determines which handler gets called
    "data": "Hello World!" // This is returned or passed onto your handler as "data" argument
}

WebSocketContext

Every handler receives a Context object as its first argument. This context is unique for each connection, and persists between handlers.

@WS.OnConnect()
async def on_connection(ctx: Context):
    ctx.hello = "Hello World!" # Set a value
    
@WS.OnPacket("getHello")
async def get_hello(ctx: Context, data: any):
    return ctx.hello # This works, because Context persists across handlers

Context Fields

Field	Type	Description
ctx.ws	WebSocket	Starlette's WebSocket Object
ctx.clientIp	Address	The client's host and port
ctx.send(data: dict)	Method	sends raw JSON, not wrapped in a packet
ctx.close(code: int)	Method	Closes the connection

You can define your own persistent fields using __setattr__ and read with __getattr__, as shown above.

Errors

To return an error, use the Error class from napi.ws

from napi.ws import WS, Context, Error

@WS.OnPacket("evilHello")
async def evil_hello(ctx: Context, data):
    raise Error("This is an error!")

This returns a structured error packet:

{
    "id": "error",
    "data": {
        "details": "This is an error!"
    }
}