Implementing Socket IO event handling via Blueprint in Flask-SocketIO - without using a single global socketio object.

__init__.py

"""
Flask web application factory. Creates and returns the flask app object.

@see http://flask.pocoo.org/docs/1.0/
@author Brian Wojtczak
"""

import logging
import os

from flask import Flask
from flask_socketio import SocketIO


def create_app(test_config: dict = None) -> Flask:

    # Application instance
    app = Flask(__name__)

    # SocketIO extension
    socketio = SocketIO()
    socketio.init_app(app)

    # Functionality via blueprints
    from . import auth, events
    app.register_blueprint(auth.bp)
    app.register_blueprint(events.bp)

    return app

auth.py

"""
Dummy Flask Authentication Blueprint.

@author Brian Wojtczak
"""

import functools
import logging

from flask import Blueprint, request, session, abort

logger = logging.getLogger(__name__)
bp = Blueprint('auth', __name__)


def is_logged_in():
    return True


def login_required(view):
    """
    View decorator that sends an error to anonymous users.
    Use the error handler to provide a friendly user experience.
    """

    @functools.wraps(view)
    def wrapped_view(**kwargs):
        if not is_logged_in():
            abort(401)

        return view(**kwargs)

    return wrapped_view


@bp.before_app_request
def load_logged_in_user():
    pass  # Not implemented

events.py

"""
Flask Blueprint for handling events on the SocketIO stream.

@author Brian Wojtczak
"""

import functools
import logging
import time

from flask import request
from flask_socketio import emit, ConnectionRefusedError, disconnect

from .auth import is_logged_in
from .io_blueprint import IOBlueprint

logger = logging.getLogger(__name__)
bp = IOBlueprint('events', __name__)


def authenticated_only(f):
    @functools.wraps(f)
    def wrapped(*args, **kwargs):
        if not is_logged_in():
            disconnect()
        else:
            return f(*args, **kwargs)

    return wrapped


@bp.on('connect')
def connect():
    if not is_logged_in():
        raise ConnectionRefusedError('unauthorized!')
    emit('flash', 'Welcome ' + request.remote_user)  # context aware emit


@bp.on('echo')
@authenticated_only
def on_alive(data):
    logger.debug(data)
    emit('echo', data)  # context aware emit


@bp.on('broadcast')
@authenticated_only
def on_broadcast(data):
    logger.debug(data)
    bp.emit('broadcast', data)  # bp.emit same as socketio.emit

io_blueprint.py

"""
A Flask Blueprint class to be used with Flask-SocketIO.

This class inherits from the Flask Blueprint class so that
 we can use the standard Blueprint interface.

Derived from https://github.com/m-housh/io-blueprint
Original work by Michael Housh, [email protected]
Modified by Brian Wojtczak

@author Brian Wojtczak
"""

# noinspection PyPackageRequirements
import socketio

from flask import Blueprint


class IOBlueprint(Blueprint):

    def __init__(self, *args, **kwargs):
        super().__init__(self, *args, **kwargs)
        self.namespace = self.url_prefix or '/'
        self._socketio_handlers = []
        self.socketio = None
        self.record_once(self.init_socketio)

    def init_socketio(self, state):
        self.socketio: socketio.Client = state.app.extensions['socketio']
        for f in self._socketio_handlers:
            f(self.socketio)

        return self.socketio

    def on(self, key):
        """ A decorator to add a handler to a blueprint. """

        def wrapper(f):
            def wrap(sio):
                @sio.on(key, namespace=self.namespace)
                def wrapped(*args, **kwargs):
                    return f(*args, **kwargs)

                return sio

            self._socketio_handlers.append(wrap)

        return wrapper

    def emit(self, *args, **kwargs):
        self.socketio.emit(*args, **kwargs)

LICENSE

Copyright 2020 Brian Wojtczak

Permission is hereby granted, free of charge, to any person obtaining a
copy of this software and associated documentation files (the "Software"),
to deal in the Software without restriction, including without limitation
the rights to use, copy, modify, merge, publish, distribute, sublicense,
and/or sell copies of the Software, and to permit persons to whom the
Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included
in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF 
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY 
CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, 
TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE 
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

License?

License?

Added MIT license declaration for you.

I know this is old post, but still. Is this still working? Do you maybe have any idea why this error happens:
if "." in name:
TypeError: argument of type 'IOBlueprint' is not iterable

I get same error with your code and with integrating IOBlueprint into my test project. Thanks in advance.

I used it in a few projects - it works there, but I've not worked on those projects for a while though, so still using older versions of libraries.

eventlet==0.25.1
Flask==1.0.3
Flask-SocketIO==4.1.0
python-socketio==4.2.0

Also please make sure you're using Python 3.

Thank you Brian. It works with those versions. I guess something was changed in some of those modules later I guess. Will deep dive and try to find a fix.

@avijoni

I believe I have found a fix for this. On line 23, delete the first argument to super(). So:

# go from this
class IOBlueprint(Blueprint):

    def __init__(self, *args, **kwargs):
        super().__init__(self, *args, **kwargs)
        # <rest of __init__ is the same>

# to this
class IOBlueprint(Blueprint):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)                   # <= notice the lack of the first `self` argument here
        # <rest of __init__ is the same>

Thanks @enjoythecode

@enjoythecode Thanks. I tested it today and I can confirm fix works.

Is there anything else that needs to be taken into consideration if this is using https/nginx/gunicon, still can't seem to catch the emits on the server coming from the client, although server->client works as it should. Great code though, thank you.

Is there anything else that needs to be taken into consideration if this is using https/nginx/gunicon, still can't seem to catch the emits on the server coming from the client, although server->client works as it should. Great code though, thank you.

Just the normal considerations for web sockets.
https://www.nginx.com/blog/websocket-nginx/