python-socketioのsioインスタンスを操作するWebAPIを作りたくて調べていました。
試した環境はUbuntu 18.04、Python 3.8、python-socketio 4.4.0、python-engineio 3.11.2、FastAPI 0.52.0です。
python-socketioのASGIAppクラスに other_asgi_app
という引数があり、ここにASGIアプリケーションを指定すれば、socketio以外のトラフィックを指定したASGIアプリケーションに流してくれるようです。
python-socketio.readthedocs.io
コード
server.py:
import socketio from fastapi import FastAPI # setup fastapi app_fastapi = FastAPI() # setup socketio sio = socketio.AsyncServer(async_mode='asgi') app_socketio = socketio.ASGIApp(sio, other_asgi_app=app_fastapi) @app_fastapi.get("/") async def index(): """fastapiのAPI実装(socketioに関係ない) """ return {"result": "Index"} @app_fastapi.get("/ping/{sid}") async def ping(sid: str): """指定されたsidにemitするエンドポイント """ sio.start_background_task( sio.emit, "ping", {"message": "ping from server"}, room=sid) return {"result": "OK"} @sio.event async def connect(sid, environ): """socketioのconnectイベント """ print('connect ', sid) @sio.event async def disconnect(sid): """socketioのdisconnectイベント """ print('disconnect ', sid)
client.py:
import asyncio import socketio sio = socketio.AsyncClient(reconnection=False) @sio.on('ping') async def on_ping(data): print('on_ping: ', data) @sio.event async def connect(): print('connected.') @sio.event async def disconnect(): print('disconnected.') async def main(): await sio.connect('http://localhost:8000') await sio.wait() asyncio.run(main())
requirements.txt:
fastapi uvicorn python-socketio aiohttp
実行結果
server:
# uvicornでサーバーアプリケーションを実行 $ uvicorn server:app_socketio INFO: Server initialized for asgi. INFO: Started server process [14975] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit) INFO: db0b541370c040bea135855bbd0384eb: Sending packet OPEN data {'sid': 'db0b541370c040bea135855bbd0384eb', 'upgrades': ['websocket'], 'pingTimeout': 60000, 'pingInterval': 25000} connect db0b541370c040bea135855bbd0384eb INFO: db0b541370c040bea135855bbd0384eb: Sending packet MESSAGE data 0 INFO: 127.0.0.1:50052 - "GET /socket.io/?transport=polling&EIO=3&t=1583778107.0980704 HTTP/1.1" 200 OK INFO: ('127.0.0.1', 50052) - "WebSocket /socket.io/" [accepted] INFO: db0b541370c040bea135855bbd0384eb: Received request to upgrade to websocket INFO: db0b541370c040bea135855bbd0384eb: Upgrade to websocket successful INFO: db0b541370c040bea135855bbd0384eb: Received packet PING data None INFO: db0b541370c040bea135855bbd0384eb: Sending packet PONG data None INFO: 127.0.0.1:50054 - "GET /ping/db0b541370c040bea135855bbd0384eb HTTP/1.1" 200 OK INFO: emitting event "ping" to db0b541370c040bea135855bbd0384eb [/] INFO: db0b541370c040bea135855bbd0384eb: Sending packet MESSAGE data 2["ping",{"message":"ping from server"}]
client:
# サーバー起動後にクライアントを起動 $ python client.py connected. on_ping: {'message': 'ping from server'} disconnected.
# sidを指定してcurlでAPI呼び出し $ curl -w '\n' http://localhost:8000/ping/db0b541370c040bea135855bbd0384eb {"result":"OK"}