Использование веб-сокетов в Python с Socket.IO

GET /socket HTTP/1.1
Host: example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: x3JJHMbDL1EzLkh9GBhXDw==
Sec-WebSocket-Protocol: chat, superchat
Sec-WebSocket-Version: 13

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: HSmrc0sMlYUkAGmm5OPpG2HaGWk=
Sec-WebSocket-Protocol: chat

# Серверная часть Socket.IO с поддержкой фоллбэков
from flask import Flask
from flask_socketio import SocketIO
 
app = Flask(__name__)
socketio = SocketIO(app, async_mode='eventlet', 
                   transport_options={
                       'websocket': True,
                       'polling': True
                   })
 
@socketio.on('connect')
def on_connect():
    print(f"Клиент подключился через {request.environ.get('socketio.transport')}")

@socketio.on('join')
def on_join(data):
    username = data['username']
    room = data['room']
    join_room(room)
    socketio.emit('status', f'{username} присоединился к комнате {room}', room=room)

<тип пакета>[<номер пакета>][<пространство имён>][<данные>]

# Отправка бинарных данных через Socket.IO
@socketio.on('request_image')
def send_image():
    with open('image.jpg', 'rb') as f:
        image_data = f.read()
    socketio.emit('image_response', {'image': True, 'buffer': image_data})

@socketio.on('connect')
def connect_handler():
    token = request.args.get('token')
    if not validate_token(token):
        return False  # Отклонить соединение
    current_user.sid = request.sid  # Сохранить идентификатор сессии

# Настройка Socket.IO с Redis в качестве брокера сообщений
socketio = SocketIO(app, message_queue='redis://localhost:6379/0')

socketio = SocketIO(app, cors_allowed_origins="https://frontend.example.com")

socketio = SocketIO(app, cors_allowed_origins=["https://production.example.com", 
                                           "https://staging.example.com"])

from flask import Flask, render_template
from flask_socketio import SocketIO
 
app = Flask(__name__)
app.config['SECRET_KEY'] = 'ваш-секретный-ключ'
socketio = SocketIO(app)
 
@app.route('/')
def index():
    return render_template('index.html')
 
@socketio.on('message')
def handle_message(message):
    print('Получено сообщение: ' + message)
    socketio.emit('response', {'data': 'Сервер получил: ' + message})
 
if __name__ == '__main__':
    socketio.run(app, debug=True)

@socketio.on('chat_message')
def handle_chat_message(data):
    # data — это словарь с данными от клиента
    username = data.get('username', 'Anonymous')
    message = data.get('message', '')
    timestamp = datetime.now().strftime('%H:%M:%S')
    
    socketio.emit('new_message', {
        'username': username,
        'message': message,
        'timestamp': timestamp
    })

@socketio.on('важное_сообщение')
def handle_important_message(data, callback):
    # Обрабатываем данные...
    
    # Вызываем обратный вызов с подтверждением
    callback({'status': 'success'})

from flask_socketio import join_room, leave_room
 
@socketio.on('join')
def on_join(data):
    username = data['username']
    room = data['room']
    join_room(room)
    socketio.emit('status', f'{username} присоединился к комнате {room}', room=room)
 
@socketio.on('leave')
def on_leave(data):
    username = data['username']
    room = data['room']
    leave_room(room)
    socketio.emit('status', f'{username} покинул комнату {room}', room=room)
 
@socketio.on('message_to_room')
def message_to_room(data):
    room = data['room']
    socketio.emit('message', data['msg'], room=room)

socketio = SocketIO(app, cors_allowed_origins="*")  # В продакшене используйте конкретные домены!

import threading
 
@socketio.on('запрос_данных')
def handle_data_request():
    # Неправильно - блокирует обработку других клиентов
    # result = requests.get('https://api.example.com/data').json()
    # socketio.emit('данные', result)
    
    # Правильно - выполняем долгую операцию в отдельном потоке
    def fetch_data():
        result = requests.get('https://api.example.com/data').json()
        socketio.emit('данные', result)
    
    threading.Thread(target=fetch_data).start()

@socketio.on('connect')
def connect_handler():
    token = request.args.get('token')
    if not validate_token(token):
        raise ConnectionRefusedError('unauthorized!')

/app
  /static
    /js
      socketio_client.js
  /templates
    index.html
  /socket_events
    __init__.py
    chat_events.py
    notification_events.py
  /services
    chat_service.py
    user_service.py
  __init__.py
  models.py
  routes.py

# chat_events.py
from flask_socketio import emit, join_room, leave_room
from app.services.chat_service import save_message, get_room_history
from app import socketio
 
@socketio.on('send_message')
def handle_send_message(data):
    # Обработка сообщения с использованием сервисного слоя
    message = save_message(data['user_id'], data['room'], data['message'])
    emit('new_message', message.to_dict(), room=data['room'])
 
@socketio.on('join_room')
def handle_join_room(data):
    # Логика присоединения к комнате...

import pytest
from flask import Flask
from flask_socketio import SocketIO
 
# Импортируем наше приложение
from app import create_app, socketio as app_socketio
 
@pytest.fixture
def app():
    app = create_app(testing=True)
    return app
 
@pytest.fixture
def socketio_client(app):
    return app_socketio.test_client(app)
 
def test_connect(socketio_client):
    assert socketio_client.is_connected()
    
def test_chat_message(socketio_client):
    # Отправляем сообщение
    socketio_client.emit('chat_message', {'username': 'Tester', 'message': 'Hello!'})
    
    # Получаем ответное событие
    received = socketio_client.get_received()
    assert len(received) > 0
    assert received[0]['name'] == 'new_message'
    assert received[0]['args'][0]['username'] == 'Tester'

def test_delayed_response(socketio_client):
    socketio_client.emit('slow_operation', {'param': 'value'})
    
    # Ждем ответа до 2 секунд
    socketio_client.get_received()  # Очищаем буфер полученых сообщений
    socketio_client.sleep(2)  
    received = socketio_client.get_received()
    
    assert len(received) > 0
    assert received[0]['name'] == 'operation_result'

def test_room_messages(app, socketio_client):
    # Создаем второго клиента
    client2 = app_socketio.test_client(app)
    
    # Оба клиента присоединяются к комнате
    socketio_client.emit('join', {'username': 'User1', 'room': 'test_room'})
    client2.emit('join', {'username': 'User2', 'room': 'test_room'})
    
    # Очищаем полученные события от присоединения
    socketio_client.get_received()
    client2.get_received()
    
    # Первый клиент отправляет сообщение в комнату
    socketio_client.emit('message_to_room', {'room': 'test_room', 'msg': 'Hi everyone!'})
    
    # Проверяем, что второй клиент получил сообщение
    received = client2.get_received()
    assert len(received) > 0
    assert received[0]['name'] == 'message'
    assert received[0]['args'][0] == 'Hi everyone!'

@socketio.on('get_dashboard_data')
def send_dashboard_data(client_data):
    client_hash = client_data.get('data_hash', '')
    current_data = get_current_dashboard_data()
    current_hash = calculate_hash(current_data)
    
    if client_hash == current_hash:
        # Данные не изменились, отправляем только подтверждение
        return {'status': 'unchanged', 'hash': current_hash}
    
    # Данные изменились, отправляем только дельту
    if client_hash:
        delta = generate_delta(client_hash, current_hash)
        return {'status': 'delta', 'delta': delta, 'hash': current_hash}
    
    # Первый запрос или клиент сильно отстал, отправляем полные данные
    return {'status': 'full', 'data': current_data, 'hash': current_hash}

import msgpack
 
@socketio.on('binary_request')
def handle_binary_request():
    data = {"values": [1, 2, 3, 4, 5], "text": "Пример данных"}
    
    # Сериализуем в MessagePack (компактнее JSON)
    binary_data = msgpack.packb(data)
    
    # Отправляем как бинарные данные
    emit('binary_response', binary_data, binary=True)

# Выбор асинхронного режима при инициализации
socketio = SocketIO(app, async_mode='eventlet')  # Варианты: 'eventlet', 'gevent', 'threading'

if __name__ == '__main__':
    socketio.run(app, host='0.0.0.0', port=5000)

@socketio.on('async_operation')
async def handle_async_operation(data):
    # Эта функция не блокирует основной поток
    result = await perform_async_operation(data)
    emit('operation_result', result)

import logging
logger = logging.getLogger('socketio')
 
@socketio.on('connect')
def on_connect():
    user_id = get_user_id_from_session()
    logger.info(f"User {user_id} connected, sid: {request.sid}")
 
@socketio.on('disconnect')
def on_disconnect():
    user_id = get_user_id_from_session()
    logger.info(f"User {user_id} disconnected, sid: {request.sid}")

from prometheus_client import Counter, Gauge
 
connection_gauge = Gauge('socketio_active_connections', 'Number of active Socket.IO connections')
message_counter = Counter('socketio_messages_total', 'Total number of Socket.IO messages processed')
 
@socketio.on('connect')
def on_connect():
    connection_gauge.inc()
 
@socketio.on('disconnect')
def on_disconnect():
    connection_gauge.dec()
 
@socketio.on('chat_message')
def handle_message(message):
    message_counter.inc()
    # Обработка сообщения...

@socketio.on('send_message')
def handle_message(data):
  # Сохраняем сообщение в БД
  message_id = save_to_database(data)
  
  # Отправляем в комнату с подтверждением
  socketio.emit('new_message', 
               {'id': message_id, 'content': data['content']}, 
               room=data['room'])
  
  return {'status': 'sent', 'id': message_id}
 
@socketio.on('message_seen')
def handle_seen(data):
  # Обновляем статус в БД
  update_message_status(data['message_id'], 'seen')
  
  # Уведомляем отправителя о прочтении
  socketio.emit('message_status', 
               {'id': data['message_id'], 'status': 'seen'}, 
               room=get_sender_room(data['message_id']))

@socketio.on('place_card')
def handle_card_placement(data):
  game_id = data['game_id']
  player_id = get_player_from_session()
  card = data['card']
  
  # Проверка валидности хода
  if not is_valid_move(game_id, player_id, card):
      return {'status': 'error', 'message': 'Invalid move'}
  
  # Обновляем состояние игры
  update_game_state(game_id, player_id, card)
  
  # Сообщаем всем игрокам о ходе
  game_state = get_game_state(game_id)
  socketio.emit('game_update', game_state, room=f'game_{game_id}')
  
  # Проверяем условия окончания игры
  if is_game_over(game_id):
      socketio.emit('game_over', get_game_results(game_id), room=f'game_{game_id}')

# Фоновый поток, собирающий данные с датчиков
def sensor_polling_thread():
  last_values = {}
  while True:
      current_values = read_sensors()
      
      # Отправляем только изменения
      for sensor_id, value in current_values.items():
          if sensor_id not in last_values or last_values[sensor_id] != value:
              socketio.emit('sensor_update', 
                         {'id': sensor_id, 'value': value}, 
                         namespace='/monitoring')
      
      last_values = current_values.copy()
      time.sleep(0.5)  # Опрашиваем датчики 2 раза в секунду
 
# Запускаем в отдельном потоке
threading.Thread(target=sensor_polling_thread, daemon=True).start()

@socketio.on('add_sticky')
def handle_add_sticky(data):
  board_id = data['board_id']
  sticky_id = generate_unique_id()
  sticky_data = {
      'id': sticky_id,
      'content': data['content'],
      'position': data['position'],
      'author': get_user_info()
  }
  
  # Сохраняем в БД
  save_sticky(board_id, sticky_data)
  
  # Оповещаем всех участников доски
  socketio.emit('sticky_added', sticky_data, room=f'board_{board_id}')
  
  return {'status': 'success', 'id': sticky_id}

# На каждом сервере настраиваем Socket.IO с Redis
socketio = SocketIO(app, message_queue='redis://redis-server:6379/0')

# Вместо хранения данных в памяти для каждого сокета
socket_data = {}  # Bad practice!
 
@socketio.on('connect')
def connect():
    socket_data[request.sid] = {'huge': 'data structure'}
 
# Используйте внешнее хранилище вроде Redis
@socketio.on('connect')
def connect():
    redis_client.hset(f"socket:{request.sid}", "last_active", time.time())

@socketio.on('get_data')
def handle_get_data(data):
    page = data.get('page', 1)
    per_page = data.get('per_page', 50)
    
    # Возвращаем только порцию данных
    results = fetch_data_from_db(page=page, per_page=per_page)
    emit('data_result', {
        'items': results,
        'total': count_total_results(),
        'page': page
    })

import eventlet
eventlet.monkey_patch()  # Патчим стандартную библиотеку
 
app = Flask(__name__)
socketio = SocketIO(app, async_mode='eventlet')
 
if __name__ == '__main__':
    # Используем встроенный в eventlet сервер
    socketio.run(app, host='0.0.0.0', port=5000)

@socketio.on('connect')
def connect():
    jwt_token = request.args.get('token')
    try:
        payload = jwt.decode(jwt_token, app.config['SECRET_KEY'], algorithms=['HS256'])
        # Сохраняем информацию о пользователе
        session['user_id'] = payload['sub']
        return True
    except jwt.PyJWTError:
        # Отклоняем соединение при невалидном токене
        return False

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
 
limiter = Limiter(app, key_func=get_remote_address)
 
@socketio.on('chat_message')
@limiter.limit("10/minute")  # Ограничение: 10 сообщений в минуту
def handle_chat_message(data):
    # Обработка сообщения

@socketio.on('connect')
def connect():
    origin = request.headers.get('Origin', '')
    if origin not in ['https://your-app.com', 'https://www.your-app.com']:
        return False

import socketio
from fastapi import FastAPI
 
app = FastAPI()
sio = socketio.AsyncServer(async_mode='asgi')
socket_app = socketio.ASGIApp(sio, app)
 
@sio.event
async def connect(sid, environ):
    print(f"Client connected: {sid}")
 
@sio.event
async def message(sid, data):
    print(f"Message from {sid}: {data}")
    await sio.emit('response', {'data': 'Server received your message!'}, to=sid)

# В channels routing.py
from django.urls import re_path
from channels.routing import ProtocolTypeRouter, URLRouter
from . import consumers
 
websocket_urlpatterns = [
    re_path(r'ws/chat/(?P<room_name>\w+)/$', consumers.ChatConsumer.as_asgi()),
]
 
application = ProtocolTypeRouter({
    'websocket': URLRouter(websocket_urlpatterns),
})