~12 min · library, python-socketio, fastapi
Python 의 표준 Socket.IO 서버는 python-socketio 야. ASGI 와 FastAPI 통합을 지원하고 수평 확장을 위한 Redis 관리자도 제공해. REST 엔드포인트용 FastAPI 앱과 AsyncServer 기반 Socket.IO 서버를 만든 뒤 둘을 ASGIApp 으로 감싸 Uvicorn 에서 함께 제공하는 패턴을 써.
Socket.IO 의 connect 처리기는 쿼리 문자열이나 헤더와 별도로 auth 딕셔너리를 받아. 클라이언트가 SDK 의 auth 선택지로 전달하는 값이야. 검증에 실패하면 ConnectionRefusedError 를 일으켜 클라이언트가 연결 성공 상태를 전혀 보지 않게 해. 사용자 신원은 sio.save_session(sid, ...) 에 저장해 다음 처리기에서 사용하면 돼.
import socketio
from fastapi import FastAPI
# REST app
api = FastAPI()
@api.get('/health')
async def health():
return {'ok': True}
# Socket.IO server
sio = socketio.AsyncServer(async_mode='asgi', cors_allowed_origins='*')
@sio.event
async def connect(sid, environ, auth):
token = (auth or {}).get('token')
if not token or not validate_token(token):
raise socketio.exceptions.ConnectionRefusedError('unauthorized')
user = decode(token)
await sio.save_session(sid, {'user': user})
print('connected', sid, user['name'])
@sio.event
async def disconnect(sid):
print('disconnected', sid)
@sio.on('chat:message')
async def chat_message(sid, data):
sess = await sio.get_session(sid)
user = sess['user']
await sio.emit('chat:message', {
'from': user['name'],
'text': data['text'],
}, room=data['room'])
# Mount everything together
app = socketio.ASGIApp(sio, other_asgi_app=api)
# Run: uvicorn main:app --host 0.0.0.0 --port 8000connect_error 를 표시하는지 확인해. Socket.IO SDK 가 없으면 연결 자체가 되지 않는 것도 확인해.아직 댓글이 없어요. 첫 댓글을 남겨보세요.