Sending websocket message from http route

[MateuszBelczowski]

Description
Let's say that I want to notify client application about the changes performed by other users.
To do that, I would like to create a websocket endpoint (e.g. /notify) that user can connect to and listen for incoming updates from server.

However, I wasn't able to find any example of doing that in the documentation. What I'm looking for is an easy way to send websocket message to the user from HTTP route. Do you have any suggestions, how can I achieve that?

[CINOAdam]

Hi there, I was able to get something like this working using Redis with Socket IO with Fast API.

https://python-socketio.readthedocs.io/en/latest/server.html#emitting-from-external-processes

[MateuszBelczowski]

Yes, that's what I meant, thanks :) Any other packages that can be used to achieve this?

[fcarron]

Hello

I think that you could keep track of all websocket clients (like it is done in the websockets package (not the starlette one) https://websockets.readthedocs.io/en/stable/intro.html.

Doing so would look like this:

import asyncio
from fastapi import FastAPI, Depends
from starlette.websockets import WebSocket

ws_clients = []
app = FastAPI()


def get_ws_clients():
    return ws_clients


@app.get("/update")
async def notify(clients=Depends(get_ws_clients)):
    """
    Notify all websocket clients
    """
    await asyncio.wait([ws.send_text("notification!") for ws in clients])
    return {}


@app.websocket("/notify")
async def websocket_endpoint(websocket: WebSocket, clients=Depends(get_ws_clients)):
    await websocket.accept()
    clients.append(websocket)
    try:
        while True:
            _ = await websocket.receive_text()
    finally:
        ws_clients.remove(websocket)

Note that this will only work if the app is running in a single process . Perhaps you should also perform some error handling when sending the notification. (?)

[tiangolo]

Thanks everyone for the help here!

@MateuszBelczowski does that solve your problem? Do you want to close the issue?

[MateuszBelczowski]

@tiangolo Yes, I was used to Django Channels to handle that and was curious what are the alternatives in the FastAPI world. So thanks everyone for the feedback :)

[Gaston-Ortiz]

Hello,

I am using the same example that @fcarron recommended. Now I would like to create some unit testing for the API "/update" and I doing that with this code:

`
@app.get("/update")
async def notify(clients=Depends(get_ws_clients)):
"""
Notify all websocket clients
"""
await asyncio.wait([ws.send_text("notification!") for ws in clients])
return {'test': 'test1'}

from fastapi.testclient import TestClient

client = TestClient(app)

def test_update():
# create the websocket
with client.websocket_connect("/notify") as ws:
# Do the request to update
response = client.get("/update")
#get the response
result_dict = response.json()
print("RESPONSE JSON:", result_dict)
# validate the outputs
assert response.status_code == 200
assert result_dict == {'test': 'test1'}
# close the websocket conection
ws.close()
`

But when I run pytest to check the assert I get the error WebSocketDisconnect. This is the output in the terminal:

`test_main.py::test_update FAILED [100%]

====================================================== FAILURES =======================================================
_____________________________________________________ test_update _____________________________________________________

def test_update():
    # create the websocket
    with client.websocket_connect("/notify") as ws:
        # Do the request to update
        response = client.get("/update")
        #get the response
        result_dict = response.json()
        print("RESPONSE JSON:", result_dict)
        # validate the outputs
        assert response.status_code == 200
        assert result_dict == {'test': 'test1'}
        # close the websocket conection

      ws.close()

test_main.py:50:

---

C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\starlette\testclient.py:326: in exit
raise message
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\anyio\from_thread.py:187: in _call_func
retval = await retval
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\starlette\testclient.py:336: in _run
await self.app(scope, receive, send)
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\fastapi\applications.py:259: in call
await super().call(scope, receive, send)
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\starlette\applications.py:112: in call
await self.middleware_stack(scope, receive, send)
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\starlette\middleware\errors.py:146: in call
await self.app(scope, receive, send)
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\starlette\exceptions.py:58: in call
await self.app(scope, receive, send)
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\fastapi\middleware\asyncexitstack.py:21: in call
raise e
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\fastapi\middleware\asyncexitstack.py:18: in call
await self.app(scope, receive, send)
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\starlette\routing.py:656: in call
await route.handle(scope, receive, send)
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\starlette\routing.py:315: in handle
await self.app(scope, receive, send)
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\starlette\routing.py:77: in app
await func(session)
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\fastapi\routing.py:274: in app
await dependant.call(**values)
test_main.py:28: in websocket_endpoint
_ = await websocket.receive_text()
C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\starlette\websockets.py:85: in receive_text
self._raise_on_disconnect(message)

---

self = <starlette.websockets.WebSocket object at 0x0000021196B095E0>
message = {'code': 1000, 'type': 'websocket.disconnect'}

def _raise_on_disconnect(self, message: Message) -> None:
    if message["type"] == "websocket.disconnect":

      raise WebSocketDisconnect(message["code"])

E starlette.websockets.WebSocketDisconnect: 1000

C:\Users\gasto\miniconda3\envs\ai_ml\lib\site-packages\starlette\websockets.py:80: WebSocketDisconnect
------------------------------------------------ Captured stdout call -------------------------------------------------
RESPONSE JSON: {'test': 'test1'}
=============================================== short test summary info ===============================================
FAILED test_main.py::test_update - starlette.websockets.WebSocketDisconnect: 1000
================================================== 1 failed in 2.11s ==================================================`

I tried different approaches but always get the same error. Please can you help me with that? I will really appreciate it.

[xiki808]

This works for me when running on uvicorn, but not when using gunicorn. The list of clients keep resetting

Edit:
I figure it's due to multiple workers. Is there a way how to solve this?

[s-weigand]

If you need support for multiple workers you could use broadcaster as the docs recommend in the Tip

[JarroVGIT]

Broadcaster does not support multiple workers per se. It holds a central object in memory for all connections, but when you have multiple workers, the connections are scattered across workers. The principles are pretty much the same for the example code provided in the docs.

[s-weigand]

@JarroVGIT While I understand that the memory backend would have the same problem since each worker creates its own async.Queue which isn't shared with other workers. It should work (not tested) with other backends (e.g. redis) where each worker would subscribe to the same queue 🤔

[JarroVGIT]

Well, that is very true. My point wasn't very explicit I realise; it was that Broadcaster doesn't 'solve' the issue of multiple workers. Let's say you have an endpoint to which someone can post a message that need to be broadcasted to all WS clients. This cannot be done in multiple workers, as the receiving endpoint (the one that gets the post with the message to broadcast) only lives in 1 worker.

However, for external queues, Broadcast can definitely work well. But, that wouldn't really solve the issue of @xiki808 I would think. But then again, I have no idea how having multiple workers would lead to a suddenly empty list (other then unwanted restarts of the workers) :P

[xiki808]

Thanks @JarroVGIT, exactly what I would like to solve. My only way to fix this is to resort to one worker, for just one functionality that needs to send a message to all clients connected to the backend.

[JarroVGIT]

Well, there might be a solution. Just to confirm I have the problem right: you can only have one worker that listens to something and broadcasts to all connected clients, correct? The problem lies in that you can only have one active worker listening, not in that clients are connecting to different workers.

If that is the case, you could go for a multi-layered approach in which you have two sub applications:

1. Application A is running on its own uvicorn/gunicorn instance with only 1 worker. It is a simple "listen and broadcast" application such as you have now. This application is not public and only serves to broadcast to instances of Application B.
2. Application B is also running on its own uvicorn/gunicorn instance but with multiple workers. External clients connect only to application B. Application B connects to the websocket endpoint of Application A and for each incoming message, distributes that to all connected clients.

This way, you are much more scalable while still dealing with the limitation that only 1 worker can serve as a listener to the source.

Anyway, this is way beyond the scope of this issue. I hope it helps but if you run into any problems, please open up a new one.