How to add middleware issues in FastAPI

[weeptao]

I think this middleware will be called first when sending any data, similar to the middleware of FastAPI

from fastapi import FastAPI
import socketio


app = FastAPI()

redis_manager = socketio.AsyncRedisManager(
    'redis://localhost:6379/0',
    channel='socketio',
    write_only=False
)

sio = socketio.AsyncServer(
    client_manager=redis_manager,
    async_mode='asgi',
    cors_allowed_origins='*',
    cors_headers=["X-User-ID", "Authorization", "X-Custom-Header"],
    logger=True,
    engineio_logger=True,
)

sio_app = socketio.ASGIApp(
    sio,
    socketio_path='socket.io'
)

# class MainNamespace(socketio.AsyncNamespace):
#     async def on_connect(self, sid, environ, auth):
#         print(f"on_connect connected: {sid}")

#     async def on_disconnect(self, sid):
#         print(f"MainNamespace: Client disconnected: {sid}")

#     async def on_message(self, sid, data):
#         print(f"on_message {sid}: {data}")

# Not working
# sio.register_namespace(MainNamespace('/socket.io'))


@sio.event
async def connect(sid, environ, auth):
    pass
    return True


@sio.event
async def disconnect(sid):
    pass


app.mount("/socket.io", sio_app)
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        '__main__:app',
        host="0.0.0.0",
    )

[miguelgrinberg]

There is a complete FastAPI example in this repository that I think will be useful to you: https://github.com/miguelgrinberg/python-socketio/blob/main/examples/server/asgi/fastapi-fiddle.py.

If you have any questions after seeing that let me know.

[weeptao]

I tried to add middleware, but found that only HTTP is supported in the middleware of FastAPI。
So I still don't know how to add specific middleware to my socket

#!/usr/bin/env python
from fastapi import FastAPI
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
import socketio
import uvicorn

app = FastAPI()
app.mount('/static', StaticFiles(directory='static'), name='static')

sio = socketio.AsyncServer(async_mode='asgi')
combined_asgi_app = socketio.ASGIApp(sio, app)


async def test(request, call_next):
    print("middleware")
    response = await call_next(request)


app.middleware("http")(test)


@app.get('/')
async def index():
    return FileResponse('fiddle.html')


@app.get('/hello')
async def hello():
    return {'message': 'Hello, World!'}


@sio.event
async def connect(sid, environ, auth):
    print(f'connected auth={auth} sid={sid}')
    await sio.emit('hello', (1, 2, {'hello': 'you'}), to=sid)


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


if __name__ == '__main__':
    uvicorn.run(combined_asgi_app, host='127.0.0.1', port=5000)

[miguelgrinberg]

Socket.IO does not provide an add_middleware function like FastAPI does. If you need a middleware that covers both the FastAPI and Socket.IO apps then you have to do it as an ASGI wrapper. Basically you create your own ASGI application that receives requests, then you forward those requests to combined_asgi_app.

You can look at the implementation of the ASGIApp class to get some ideas on how to do this, since this class is also designed as an ASGI wrapper that calls the FastAPI and the Socket.IO sub-apps.