معرفی کتابخانه python-socketio

python-socketio یک پیاده‌سازی کامل پروتکل Socket.IO برای پایتون است که امکان ارتباط بلادرنگ (real-time) بین کلاینت‌ها و سرور را فراهم می‌کند. این کتابخانه از هر دو حالت synchronous و asynchronous پشتیبانی می‌کند و با فریم‌ورک‌هایی مثل Flask، Django و FastAPI قابل ادغام است. در این مقاله به مفاهیم پایه، نمونه‌های کد، نکات عملیاتی و روش‌های مقیاس‌پذیری خواهیم پرداخت.

ویژگی‌های کلیدی

• پشتیبانی از رویدادها (events) و ارسال پیام و Broadcast
• قابلیت کار با namespaces و rooms
• همکاری با پروتکل‌های مختلف حمل‌ونقل (WebSocket و HTTP long-polling)
• انتقال حالت بین چند فرایند/سرور با استفاده از Redis یا RabbitMQ
• نسخه‌های sync و async (asyncio)

چرا از python-socketio استفاده کنیم؟

برای برنامه‌هایی که نیاز به همگام‌سازی لحظه‌ای مانند چت، داشبوردهای زنده، بازی‌ها یا نوتیفیکیشن دارند، استفاده از socket.io به دلیل سادگی ارسال/دریافت رویدادها و مدیریت اتصالات انتخاب مناسبی است. python-socketio سازگاری زیادی با کلاینت‌های جاوااسکریپت و موبایل دارد و به توسعه‌دهنده امکان تمرکز بر منطق برنامه را می‌دهد.

نمونه ساده: سرور synchronous با Flask

from flask import Flask, render_template
import socketio

sio = socketio.Server()
app = Flask(__name__)
app.wsgi_app = socketio.WSGIApp(sio, app.wsgi_app)

@sio.event
def connect(sid, environ):
    print('Client connected:', sid)

@sio.event
def message(sid, data):
    print('Message from', sid, data)
    sio.send(sid, 'Received: ' + data)

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

if __name__ == '__main__':
    import eventlet
    eventlet.wsgi.server(eventlet.listen(('', 5000)), app)

توضیح: این مثال یک سرور پایه با Flask و python-socketio نشان می‌دهد. ما از eventlet برای اجرای WSGI استفاده کرده‌ایم که اجازهٔ کار با WebSocket و long-polling را می‌دهد. توابع {@code connect}، {@code message} و {@code disconnect} به ترتیب برای اتصال، دریافت پیام و قطع اتصال تعریف شده‌اند.

نسخه‌ی بهینه (async با FastAPI)

import socketio
from fastapi import FastAPI
import uvicorn

sio = socketio.AsyncServer(async_mode='asgi')
app = FastAPI()
sio_app = socketio.ASGIApp(sio, other_asgi_app=app)

@sio.event
async def connect(sid, environ):
    print('Connected', sid)

@sio.event
async def chat_message(sid, data):
    await sio.emit('reply', {'msg': 'Echo: ' + data}, to=sid)

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

if __name__ == "__main__":
    uvicorn.run(sio_app, host="0.0.0.0", port=8000)

توضیح: این نسخه از AsyncServer و ASGIApp استفاده می‌کند که با FastAPI و Uvicorn سازگار است. کدهای async باعث افزایش کارایی در بارهای همزمان می‌شوند. تابع {@code chat_message} با await پیام را به همان کلاینت بازمی‌گرداند.

مثال کلاینت پایتون

import socketio

sio = socketio.Client()

@sio.event
def connect():
    print('Connected to server')

@sio.event
def reply(data):
    print('Reply:', data)

@sio.event
def disconnect():
    print('Disconnected from server')

sio.connect('http://localhost:5000')
sio.send('Hello server')
sio.wait()

توضیح: این کد یک کلاینت سینکرون است که به سرور محلی وصل می‌شود، پیام ارسال می‌کند و منتظر دریافت پاسخ می‌ماند. برای کلاینت‌های غیرهمزمان می‌توان از {@code socketio.AsyncClient} استفاده کرد.

نحوهٔ کار با Namespaces و Rooms

Namespaces به شما اجازه می‌دهد تا مجموعه‌ای از رویدادها را در مسیرهای مختلف جدا کنید (مثل /chat یا /notifications). Rooms یا اتاق‌ها برای گروه‌بندی اتصالات مفیدند—برای مثال ارسال پیام به همه کاربران یک کانال خاص.

• استفاده از {@code sio.enter_room(sid, room)} و {@code sio.leave_room(sid, room)}
• ارسال به اتاق با {@code sio.emit(‘event’, data, room=room)}

نمونه: ارسال به یک اتاق

@sio.event
def join(sid, data):
    room = data.get('room')
    sio.enter_room(sid, room)
    sio.emit('status', f'{sid} joined {room}', room=room)

توضیح: وقتی کاربر درخواست پیوستن دارد، با {@code enter_room} در آن اتاق قرار می‌گیرد و سپس پیام وضعیت به همه اعضای همان اتاق فرستاده می‌شود.

مقیاس‌پذیری: استفاده از Redis یا Message Queue

در محیطی که چند فرایند یا چند نمونهٔ سرور اجرا می‌شوند، باید وضعیت رویدادها بین این نمونه‌ها همگام‌سازی شود. python-socketio از چندین backend برای این کار پشتیبانی می‌کند؛ یکی از متداول‌ترین‌ها Redis است.

import socketio
from socketio import RedisManager
import uvicorn

mgr = RedisManager('redis://localhost:6379/0')
sio = socketio.AsyncServer(client_manager=mgr, async_mode='asgi')
# ادامهٔ تنظیمات به مانند مثال FastAPI

توضیح: با استفاده از RedisManager، پیام‌ها و اطلاعات room بین تمام نمونه‌های سرور منتشر/مشترک می‌شود، بنابراین Broadcastها در تمام نمونه‌ها کار می‌کنند.

ملاحظات امنیتی و عملکرد

• اعتبارسنجی اتصال: در هاندشِیک (connect) اطلاعات هدر یا توکن را بررسی کنید.
• محدودسازی نرخ (rate limiting): برای جلوگیری از حملات flood از محدودسازی استفاده کنید.
• TLS: همیشه برای ترافیک WebSocket از wss:// با TLS استفاده کنید.
• محدودیت اندازه پیام: از محدود کردن سایز پیام برای جلوگیری از مصرف بیش از حد حافظه استفاده کنید.

مقایسه نشانه‌ای: Sync vs Async

نکات پیشرفته و بهترین شیوه‌ها

• برای بارهای بالا، از ASGI و serverهایی مثل Uvicorn/Gunicorn با workers مناسب استفاده کنید.
• برای احراز هویت، از توکن JWT در هدر اتصال استفاده کنید و در handler connect بررسی کنید.
• از ابزارهای مانیتورینگ و محدودسازی اتصالات استفاده کنید تا از نشتی منابع جلوگیری شود.
• در صورت نیاز به ordering قوی، از فیچرهای صف بندی و تایید دریافت (acknowledgements) استفاده کنید.

جمع‌بندی و منابع

python-socketio یک ابزار قدرتمند برای ساخت اپلیکیشن‌های بلادرنگ در پایتون است؛ از پروژه‌های کوچک تا سامانه‌های مقیاس‌پذیر. با دانستن تفاوت‌های sync و async، استفاده از Redis برای مقیاس‌پذیری و رعایت نکات امنیتی می‌توانید اپلیکیشن قابل اعتمادی بسازید.

برای اطلاعات بیشتر و مثال‌های پیشرفته به مستندات رسمی کتابخانه در GitHub و راهنمای استفاده با فریم‌ورک‌های مورد علاقه‌تان مراجعه کنید.

لطفاً از کمبود ها و مشکلات این محتوا برای ما بنویسید

ارسال