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アプリケーションに流してくれるようです。

コード

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.

API呼び出し(curl):

# sidを指定してcurlでAPI呼び出し
$ curl -w '\n' http://localhost:8000/ping/db0b541370c040bea135855bbd0384eb
{"result":"OK"}

偏った言語信者の垂れ流し

python-socketioとFastAPIを組み合わせて動かす

コード

実行結果