Middleware

~12 min · library, middleware, rate-limit

Level 0Poller

0 XP0/60 lessons0/10 achievements

0/120 XP to next level120 XP to go0% complete

Connect-time middleware

The connect handler is the natural place to run middleware: auth, version checks, geo-blocks, rate-limit registration. Raise ConnectionRefusedError(code, message) with a useful payload — Socket.IO surfaces both fields to the client's connect_error event.

Per-event rate limiting

Build a rate-limit decorator and apply it to the events that need it (chat sends, file uploads). Skip it on read-only events. Store windows keyed by sid; a small deque of timestamps per connection is enough.

Code

Auth middleware in connect·python

@sio.event
async def connect(sid, environ, auth):
    token = (auth or {}).get('token')
    user = decode_jwt(token) if token else None
    if not user:
        raise socketio.exceptions.ConnectionRefusedError(
            'unauthorized',
            'token missing or invalid',
        )
    if user.get('disabled'):
        raise socketio.exceptions.ConnectionRefusedError(
            'forbidden',
            'account disabled',
        )
    await sio.save_session(sid, {'user': user})

Per-event rate-limit decorator·python

from collections import defaultdict, deque
import time, functools

windows = defaultdict(deque)  # sid -> deque of timestamps

def rate_limit(per_second=20):
    def deco(fn):
        @functools.wraps(fn)
        async def wrapper(sid, *args, **kwargs):
            now = time.time()
            w = windows[sid]
            while w and now - w[0] > 1.0:
                w.popleft()
            if len(w) >= per_second:
                await sio.emit('error',
                    {'code': 'rate_limited', 'message': 'slow down'},
                    to=sid)
                return
            w.append(now)
            return await fn(sid, *args, **kwargs)
        return wrapper
    return deco

@sio.on('chat:message')
@rate_limit(per_second=20)
async def chat(sid, data):
    # ...
    pass

External links

python-socketio: Connection events

Exercise

Add the auth middleware and a rate limit of 5 messages/sec on chat. Connect with a bad token: see connect_error. With a good token, send 100 messages in a tight loop: confirm only the first 5 succeed each second and the rest receive an error event.

Progress

Progress is local-only — sign in to sync across devices.

← PreviousAuto-Reconnection & Acknowledgments Next →Decision Matrix: Raw WS vs Libraries vs Managed

Spotted a bug or have feedback on this page?Report an Issue

Comments 0

🔔 Reply notifications (sign in)

No comments yet — be the first.