Fastapi: [QUESTION] Use of socket.io

[siedi]

Description

How can I use socket.io instead of the plain websocket integration? I guess this is more a question belonging to starlette.

Currently migrating from a flask application using flask-socketio / python-socketio

Any hint is appreciated. Thx.

[tiangolo]

Socket-IO has an ASGI compatible app: https://python-socketio.readthedocs.io/en/latest/server.html#uvicorn-daphne-and-other-asgi-servers

You should be able to mount that app like if it was a FastAPI sub-application: https://fastapi.tiangolo.com/tutorial/sub-applications-proxy/#mount-the-sub-application

[marcus-oscarsson]

Hi,

Also have a question regarding python-socketio, Ive tried:

```
sio = socketio.AsyncServer(async_mode='asgi')
sio_asgi_app = socketio.ASGIApp(sio, app)
app.mount("/api/socket.io", sio_asgi_app)

Which does indeed seem to work, but only partially. Only GET requests seems to get through 
and POST's are somehow blocked, I get:

INFO:uvicorn:('127.0.0.1', 51888) - "GET /api/socket.io/ HTTP/1.1" 200
INFO:uvicorn:('127.0.0.1', 51890) - "POST /api/socket.io/ HTTP/1.1" 405
`` Is there a way to also allow POST on/api/socket.io/` ?

Thanks

[BlackHoleFox]

I am unable to get this to work at all, even with the snippet above.

[marcus-oscarsson]

I see, here is my complete example,

    app = FastAPI(debug=True)
    sio = socketio.AsyncServer(async_mode='asgi')
    sio_asgi_app = socketio.ASGIApp(sio, app, socketio_path="/api/socket.io")

    app.mount("/api/socket.io", sio_asgi_app)
    sio.register_namespace(ws.ConnectNS('/'))

Where the ws module contains:

class ConnectNS(socketio.AsyncNamespace):
    def on_connect(self, sid, environ):
        logging.debug("Websocket connected %s", sid)

    def on_disconnect(self, sid):
        logging.debug("Websocket disconnected %s" % sid)

I then run uvicorn with the app object, running it with sio_asgi_app works fine but the FastAPI app does not receive any traffic.

It seems that the routing mechanism in either FastAPI ot Starlette somehow only allows GET requests, but I might be wrong and I'm maybe doing something else wrong.

@BlackHoleFox: it did not work at all for me either until I noticed that second parameter of ASGIApp, which gets me to the currently partially working state ...

[marcus-oscarsson]

Finally,

app = FastAPI(debug=True) sio = socketio.AsyncServer(async_mode='asgi') sio_asgi_app = socketio.ASGIApp(sio, app, socketio_path="/api/socket.io") sio.register_namespace(ws.ConnectNS('/'))

Works just fine if you use the ASGIApp object, It might maybe have other consequences but it seems
to work fine for me so far. I had some other issues with the routing in my example application making it fail the first time i tried. I never got app.mount("/api/socket.io", sio_asgi_app) to work properly though.

[BlackHoleFox]

@marcus-oscarsson I ended up with the same results. I can directly use the ASGIApp with Uvicorn but the mounting method never passes through the requests properly and my client gets a Unexpected response from server whenever it attempts to connect. Seems the solution for me at the moment will be to run the socket backend and REST backend separately.

[marcus-oscarsson]

@BlackHoleFox I see, I simply removed the app.mount(...) and I let the ASGIApp handle the rest. It works fine for me all running in the same process. So finally using the ASGIApp is a working solution for me.

[kientt86]

So are we giving up on app.mount(...)? ASGIApp is kind of work but it ends up with maintaining 2 applications
Any idea to go further on this?

[marcus-oscarsson]

@kientt86 I would say that its up to the maintainer, @tiangolo, to decide exactly what to do with app.mount. I'm not sure if I used it correctly in my example so it could also simply be that I'm missing some option. It would be nice if it worked withapp.mount since it would be more consistent.

Using the ASGIApp otherwise seems to work fine and one would still need to create and somehow handle it so I don't really see an issue with it

[tiangolo]

Thanks for all the reports guys.

I want to check how to integrate Socket.IO and integrate it directly in the docs, I haven't had the time though, but I'll do it hopefully soon.

[rudmac]

I´m looking forward for this integrate. It would be great for the project.

cheers!

[databasedav]

This is more of a feature request but related to using python-socketio. flask-socketio comes with a test_client that makes testing very convenient (no need to manually spin up a server in a separate process, can emit events with callbacks, stores messages received from the server, etc.). Comparatively, starlette.testclient.TestClient is very basic.

Hoping this can be implemented on top of the general socket.io support. I wouldn't mind working on a PR for this, but would require some direction. Thanks!

[szelenka]

I'm struggling with getting this to work with FastAPI. I've tried this code:

fast_app = FastAPI(
    openapi_url='/api/notifications/openapi.json',
    docs_url='/api/notifications/docs',
    redoc_url='/api/notifications/redoc'
)
sio = socketio.AsyncServer(
    async_mode='asgi',
    cors_allowed_origins='*'
)
app = socketio.ASGIApp(
    socketio_server=sio,
    other_asgi_app=fast_app,
    socketio_path='/api/notifications/socket.io/'
)

Which allows me to connect to the socket.io endpoint, but when I attempt to connect to a FastAPI defined endpoint, it'll raise a traceback from python-engineio:

INFO: ('127.0.0.1', 64336) - "GET /api/notifications/openapi.json HTTP/1.1" 500
ERROR: Exception in ASGI application
Traceback (most recent call last):
  File "/Users/szelenka/anaconda3/envs/python37/lib/python3.7/site-packages/uvicorn/protocols/http/httptools_impl.py", line 368, in run_asgi
    result = await app(self.scope, self.receive, self.send)
  File "/Users/szelenka/anaconda3/envs/python37/lib/python3.7/site-packages/uvicorn/middleware/debug.py", line 96, in __call__
    raise exc from None
  File "/Users/szelenka/anaconda3/envs/python37/lib/python3.7/site-packages/uvicorn/middleware/debug.py", line 78, in __call__
    await self.app(scope, receive, inner_send)
  File "/Users/szelenka/anaconda3/envs/python37/lib/python3.7/site-packages/engineio/async_drivers/asgi.py", line 53, in __call__
    await self.other_asgi_app(scope, receive, send)
TypeError: __call__() takes 2 positional arguments but 4 were given

As best I can tell, engineio is attempting to __call__ FastAPI with 4 parameters ('self', 'scope', 'receive', and 'send') .. while 0.33 version of FastAPI is simply using the Starlette __call__ method that only accepts ('self' and 'scope').

I'm interested to learn what version other people (@marcus-oscarsson ) are using where this works inside FastAPI app.

I've managed to hack around it with this helper class:

from fastapi import FastAPI
from starlette.types import ASGIInstance, Scope, Receive, Send


class FastAPISocketIO(FastAPI):
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> ASGIInstance:
        fn = super(FastAPISocketIO, self).__call__(scope=scope)
        return await fn(receive=receive, send=send)

But it seems the problem is an incompatibility with Starlette and python-socketio?

[szelenka]

Turns out my problem was with Starlette.

Starlette version 0.11.1 needed the above hack, while version 0.12.0 doesn't have the problem and works without the above hack.

[mjmare]

A bit more complete example, based on @szelenka's work that might help other people.
I don't like that it uses two apps but it seems to work.

'''
Run with:
uvicorn app_uvicorn_fastapi:app --reload --port 5000

'''
from starlette.staticfiles import StaticFiles
from fastapi import FastAPI
import uvicorn
import socketio

# There is some complexity in setting up Fastapi combined with socketio
# https://github.com/tiangolo/fastapi/issues/129

fast_app = FastAPI(
    openapi_url='/docs/openapi.json',
    docs_url='/docs/docs',
    redoc_url='/docs/redoc'
)
sio = socketio.AsyncServer(
    async_mode='asgi',
    cors_allowed_origins='*'
)
app = socketio.ASGIApp(
    socketio_server=sio,
    other_asgi_app=fast_app,
    socketio_path='/socket.io/'
)

@sio.event
def connect(sid, environ):
    print("connect ", sid, environ)

@sio.event
async def chat_message(sid, data):
    print("message ", data)
    await sio.emit('reply', room=sid)

@sio.event
def disconnect(sid):
    print('disconnect ', sid)


#  File serving, api serving by Fastapi etc

fast_app.mount("/static", StaticFiles(directory="static"), name="static")

@fast_app.get("/hello")
async def root():
    return {"message": "Hello World"}

[GustavoCarvalho]

Any news? I look forward to it working.

[rudmac]

@mjmare Very good... that's works perfectly using uvicorn.

But with gunicorn WSGI got some errors:

[2019-11-19 14:50:51 +0000] [23] [ERROR] Exception in ASGI application
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/uvicorn/protocols/websockets/websockets_impl.py", line 146, in run_asgi
    result = await self.app(self.scope, self.asgi_receive, self.asgi_send)
  File "/usr/local/lib/python3.7/site-packages/uvicorn/middleware/message_logger.py", line 58, in __call__
    raise exc from None
  File "/usr/local/lib/python3.7/site-packages/uvicorn/middleware/message_logger.py", line 54, in __call__
    await self.app(scope, inner_receive, inner_send)
  File "/usr/local/lib/python3.7/site-packages/engineio/async_drivers/asgi.py", line 46, in __call__
    await self.engineio_server.handle_request(scope, receive, send)
  File "/usr/local/lib/python3.7/site-packages/socketio/asyncio_server.py", line 326, in handle_request
    return await self.eio.handle_request(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/engineio/asyncio_server.py", line 297, in handle_request
    r['response'], environ)
  File "/usr/local/lib/python3.7/site-packages/engineio/async_drivers/asgi.py", line 168, in make_response
    'headers': headers})
  File "/usr/local/lib/python3.7/site-packages/uvicorn/middleware/message_logger.py", line 49, in inner_send
    await send(message)
  File "/usr/local/lib/python3.7/site-packages/uvicorn/protocols/websockets/websockets_impl.py", line 213, in asgi_send
    raise RuntimeError(msg % message_type)
RuntimeError: Expected ASGI message 'websocket.send' or 'websocket.close', but got 'http.response.start'.

After some testing, I'm using this code:

app = FastAPI(
    title=config.PROJECT_NAME,
    description=config.PROJECT_NAME,
    version=config.PROJECT_VERSION,
    debug=True
)

sio = socketio.AsyncServer(
    async_mode='asgi',
    cors_allowed_origins=','.join(config.ALLOW_ORIGIN)
)

sio_asgi_app = socketio.ASGIApp(
    socketio_server=sio,
    other_asgi_app=app
)

app.add_route("/socket.io/", route=sio_asgi_app, methods=['GET', 'POST'])
app.add_websocket_route("/socket.io/", sio_asgi_app)

kind off working with gunicorn...

sometimes, got the error above...

I will keep trying and let you guys know...

Cheers

[acnebs]

Can anyone provide some insight into whether they've gotten this to work well with a synchronous database connection (I'm using SQLAlchemy ORM) and what it takes to ensure thread-safety?

[shawnd]

RuntimeError: Expected ASGI message 'websocket.send' or 'websocket.close', but got 'http.response.start'.

kind off working with gunicorn...
sometimes, got the error above...

I'm receiving the same error with running using Gunicorn. Running with Uvicorn doesn't get me the same response.

[antnieszka]

I'm struggling with the same problem :( If I have 4 workers (gunicorn -w 4 ...) this error almost always happens. For 1 worker it rarely (or maybe never) does.

[jackdh]

Has there been any update on this? Still struggling to find a complete working example. Thanks.

[dmontagu]

There are a lot of different pieces in play here (gunicorn, uvicorn, starlette, fastapi, socket.io, python, ...), each of which is potentially relevant to figuring out if there is a problem.

My strong suspicion is that this issue is not specific to FastAPI, but rather is a problem with uvicorn or starlette. It's probably worth trying to replicate the problem using just starlette and/or creating an issue in the starlette repo asking for guidance.

At any rate, if you are having an issue and want help, please post the versions of each of the various dependencies you are using. That may not be enough to help resolve the issue, but as things stand it's basically impossible to help without knowing more.

[hillairet]

I had the same problem using the given example so I started from the python-socketio asgi example and added fastapi and it works :tada:

I upgraded a few packages compared to the example and here is my requirements.txt:

Click==7.0
fastapi==0.54.1
h11==0.8.1
httptools==0.1.1
pydantic==1.5.1
python-engineio==3.12.1
python-socketio==4.5.1
six==1.11.0
starlette==0.13.2
uvicorn==0.11.5
uvloop==0.14.0
websockets==8.1

And then the code is the example app.py mounted in a fastapi app (partial copy):

#!/usr/bin/env python 
import socketio    
from fastapi import FastAPI    


app = FastAPI()    

sio = socketio.AsyncServer(async_mode='asgi')    
socket_app = socketio.ASGIApp(sio, static_files={'/': 'app.html'})    
background_task_started = False    


async def background_task():    
<... EXACTLY LIKE app.py IN python-socketio/examples/server/asgi ...>

@sio.on('disconnect')
def test_disconnect(sid):
    print('Client disconnected')


@app.get("/hello")
async def root():
    return {"message": "Hello World"}


app.mount('/', socket_app)

Then used uvicorn fast_app:app --reload --host 0.0.0.0 and all good. All features seem to work properly.

[vlori2k]

This example works perfectly! Thank you boss.

[Harsha-BR]

Hello I m trying to mount the socketio as a sub application on top of a Fast API app, and it doesn't seem to work. Although if I dont mount it and run as a separate application it is just fine. Can some one comment on it? Have you faced the same issue before? Here is my code.
Server:

from fastapi import FastAPI
import uvicorn
import socketio

app = FastAPI(title="Core Stack", version="1.0.0")


@app.get("/", include_in_schema=False)
async def hello():
    return {"message": "Hello, world's best developer!"}


sio = socketio.AsyncServer(async_mode="asgi")
subapi = socketio.ASGIApp(sio)

app.mount("/subapi", subapi)


@sio.event
def my_event(sid, data):
    pass


@sio.on("connect")
def connect(sid, environ):
    print("connected", sid)


@sio.on("disconnect")
def disconnect(sid):
    print("disconnect", sid)

Client:

import socketio

sio = socketio.Client()


@sio.event
def connect():
    print("connection established")


@sio.event
def my_message(data):
    print("message received with ", data)
    sio.emit("my response", {"response": "my response"})


@sio.event
def disconnect():
    print("disconnected from server")


sio.connect("http://127.0.0.1:8000/subapi/")
sio.wait()

Error I get on Server side:

127.0.0.1:34844 - "GET /socket.io/?transport=polling&EIO=3&t=1602760515.7113314 HTTP/1.1" 404 Not Found

Here is the error I get on client side:

Traceback (most recent call last):
  File "src/tests/client_test.py", line 22, in <module>
    sio.connect("http://127.0.0.1:8000/subapi/")
  File "/home/harry/.cache/pypoetry/virtualenvs/core-stack-g4U8Q-rv-py3.8/lib/python3.8/site-packages/socketio/client.py", line 282, in connect
    six.raise_from(exceptions.ConnectionError(exc.args[0]), None)
  File "<string>", line 3, in raise_from
socketio.exceptions.ConnectionError: Unexpected status code 404 in server response

[erny]

I produced a complete example with unit tests in this comment of issue miguelgrinberg/python-socketio#332, although it doesn't use TestClient.websocket for testing, but the standard python-socketio client. Currently, for this to work, it needs to run a local uvicorn server, but starting and stopping should be extremely fast. (Just make sure that the port is free.) The UvicornTestServer class is rather reusable and extremely simple to setup, especially with pytest. Make sure you have the pytest-asyncio package installed. An example could be:

```python
import pytest
from test.utils import UvicornTestServer

@pytest.fixture
async def start_stop_server():
"""Start server as test fixture and tear down after test"""
server = UvicornTestServer() # use 'port' kwarg to set alternative port, default is 8000
await server.up()
yield
await server.down()

@pytest.mark.asyncio
async def test_method1(start_stop_server):
"""A simple http test"""
# your test goes here
# the gotcha here is that you can't use request.get, because it would block
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get('http://localhost:8000/mytest') as response:
assert response.status == 200
content = await response.text()
assert 'my test' in content

[erny]

@Harsha-BR, try to sio.connect('http://localhost:8000', socketio_path='/subapi/socket.io/')

[includeamin]

simple example

FastApi-Socket.io Server

import socketio
from fastapi import FastAPI
from socketio.asyncio_namespace import AsyncNamespace
from pydantic import BaseModel, ValidationError
from typing import Any
from logging import info

app = FastAPI()


class PacketModel(BaseModel):
    content: Any
    content_type: str


class GreyRookNameSpace(AsyncNamespace):
    # functions that have on_ prefix recognized as event
    async def on_connect(self, sid, *args, **kwargs):  # on connect event
        info(f"{sid}: Welcome!:)")

    async def on_disconnect(self, sid):  # on disconnect event
        info(f"{sid}: Bye!:(")

    async def on_packet(self, environ, *args, **kwargs):  # on packet event
        try:  # Packet Validation
            packet = PacketModel(**args[0])
        except ValidationError as ex:
            return PacketModel(
                content=str(ex.args), content_type="application/txt"
            ).dict()  # Call-Back
        await self.process_packet(packet)
        # await self.emit(
        #     "message", packet.dict(), namespace=self.namespace
        # )  # Emit to name-space
        return PacketModel(
            content="Delivered", content_type="application/txt"
        ).dict()  # Call-Back

    async def process_packet(self, packet: PacketModel):
        # some processing on packet
        # store in db
        # stream processing
        # upload chunks
        # etc
        pass


mgr = socketio.AsyncRedisManager(
    "redis://localhost/0"
)  # Message Queue is for working with distributed applications
sio = socketio.AsyncServer(
    async_mode="asgi", cors_allowed_origins="*"
)
sio.register_namespace(GreyRookNameSpace("/GreyRook"))  # register the namespace
asgi = socketio.ASGIApp(sio)

app.mount("/ws", asgi)  # mount Socket.Io to FastApi with /ws path


class SampleResponseModel(BaseModel):
    message: str


class Sample400ResponseModel(BaseModel):
    detail: str


@app.get("/", description="some description",
         responses={200: {'model': SampleResponseModel},
                    400: {'model': Sample400ResponseModel}})
async def index():
    return {"message": "Welcome to GreyRook!"}


"""
gunicorn app:app -w 1 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:3001  
horizontal scaling is the solution for high scale architecture
issue:  https://github.com/miguelgrinberg/Flask-SocketIO/issues/267#issuecomment-220525641
"""

Python Client

import socketio
from app import PacketModel

sio = socketio.Client()
sio.connect(
    "http://localhost:3001/ws", namespaces="/GreyRook", socketio_path="/ws/socket.io"
)


@sio.on("message", namespace="/GreyRook")
def new_packet(packet):
    print("\nMessage: ", packet)


def call_back(data):
    print("\ncall-back", data)


while True:
    msg = input("Message: ")
    sio.emit(
        "packet",
        PacketModel(content={"message": msg}, content_type="application/json").dict(),
        namespace="/GreyRook",
        callback=call_back,
    )

js

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/2.3.1/socket.io.js"></script>
</head>
<body>
<script>
    socket = io('http://localhost:3001/ws', {
        path: '/ws/socket.io'
    });
</script>
</body>
</html>

[Harsha-BR]

Thanks @erny and @includeamin your suggestions helped. Its working. :)

[samypr100]

TL;DR paths are important, beware of mounting it in / as it may cause issues.

Mounting it outside of / made it work for me on the first try for both uvicorn and gunicorn and it also allowed me to see what my particular issue was. Thanks for @includeamin for helping me figure it out on my side.

sio = socketio.AsyncServer(async_mode='asgi', cors_allowed_origins="*")
app.mount("/ws", socketio.ASGIApp(sio))

Will yield a connection in /ws/socket.io and will play nicely with other routes in /
If you try to mount it on app.mount("/socket.io", socketio.ASGIApp(sio)), this will yield a path in the form of /socket.io/socket.io because of socketio_path='socket.io' by default so you'd have to change your client's path accordingly.

Problems may arise if you have any kind of rules/middleware/etc. that apply to /* such as the RuntimeError: Expected ASGI message 'websocket.send' or 'websocket.close', but got 'http.response.start'. error I saw here.

[mylot-python]

EVENT 1: (if other_asgi_app arg)

Server:
app = FastAPI()
sio = socketio.AsyncServer(async_mode='asgi', cors_allowed_origins='*')
sio_asgi_app = socketio.ASGIApp(sio, other_asgi_app=app, socketio_path='socket.io')
app.mount("/", sio_asgi_app)
uvicorn.run(app='main:app', host="0.0.0.0", port=7777, reload=True)

Client:
sio = socketio.Client()
sio.connect('http://127.0.0.1:7777/' , socketio_path='some nonexistent socketio_path')
sio.wait()



md5-bb092272d81e4f678da31d374bfd2827



Server:
app = FastAPI()
sio = socketio.AsyncServer(async_mode='asgi', cors_allowed_origins='*')
sio_asgi_app = socketio.ASGIApp(sio, socketio_path='socket.io')
app.mount("/", sio_asgi_app)
uvicorn.run(app='main:app', host="0.0.0.0", port=7777, reload=True)



md5-7bde2563efed65c4882560da9e90a0dc



Like EVENT1 Client

The Server none any exceptions, only raise <404 Not Found> console message

[MatthewScholefield]

@includeamin I tried the simple example you provided but I get [ERROR] ASGI callable returned without starting response. with a 500 internal server error when the js client tries to initiate the websocket. Has anyone else experienced this?

[includeamin]

@includeamin I tried the simple example you provided but I get [ERROR] ASGI callable returned without starting response. with a 500 internal server error when the js client tries to initiate the websocket. Has anyone else experienced this?

did you run exactly my sample code?
did you use this command ?gunicorn app:app -w 1 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:3001

[MatthewScholefield]

Oh, whoops, yeah you're right, there were a few problems I figured out:

• I was using the latest socket.io javascript library which recently bumped to major version 3 (incompatible)
• I needed to run pip3 install uvicorn[standard] (or just pip install websockets) to actually support websockets so it wouldn't resort to long polling
• I was proxying the frontend and backend together using nginx. I needed to add some config values:

        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";

[makusudi]

There are a lot of different pieces in play here (gunicorn, uvicorn, starlette, fastapi, socket.io, python, ...), each of which is potentially relevant to figuring out if there is a problem.

My strong suspicion is that this issue is not specific to FastAPI, but rather is a problem with uvicorn or starlette. It's probably worth trying to replicate the problem using _just_ starlette and/or creating an issue in the starlette repo asking for guidance.

At any rate, if you are having an issue and want help, please post the versions of each of the various dependencies you are using. That may not be _enough_ to help resolve the issue, but as things stand it's basically impossible to help without knowing more.

@dmontagu exactly. I had the same problems, but then I tried to get all the dependencies with versions provided by @hillairet. Finally, it's working now, huh...