[flask sockets] boilerplate for writing Flask services with socket.io functionality #python #fullstack #javascript #flask

api_cmds-base.py

from cmds.base import FlaskView as CmdFlaskView


class FlaskView(CmdFlaskView):
    route_prefix = '/api'

    @classmethod
    def on_socket_event(cls, event_name):
        """Registers a listener for websocket events.
        
        We cache all socket listeners in `_socket_rule_cache` within the parent class.
        Then, on register, we iterate through these listeners and assign them to the socket.

        NOTE: We need to make this part of the class, so it's aware of itself.
        """
        def decorator(f):
            # Format: 'function_name': [<function>, <tagA>, <tagB>, ...]
            # NOTE: This is how you layout a decorator that takes in input.

            if not hasattr(cls, cls.socket_cache_name) or \
                    getattr(cls, cls.socket_cache_name) is None:
                # Initialize cache
                setattr(cls, cls.socket_cache_name, {
                    f.__name__: [f, event_name]
                })
            else:
                cache = getattr(cls, cls.socket_cache_name)

                if f.__name__ in cache:
                    cache[f.__name__].append(event_name)
                else:
                    cache[f.__name__] = [f, event_name]

            def returned_func(*args, **kwargs):
                return f(*args, **kwargs)

            return returned_func

        return decorator

api_cmds-page.py

from functools import wraps

from flask_socketio import emit

from api_cmds.base import FlaskView
from cmds.base import route


def example_decorator(fn):
    @wraps(fn)
    def _decorator(*args, **kwargs):
        return fn(*args, 'value', **kwargs)
    
    return _decorator


class Controller(FlaskView):
    route_base = '/example'
  
    def normal(self):
        return "This path can be accessed via `/example/normal` due to FlaskClassy."
    
    @route(path='/routeA')
    def route_decorator(self):
        return "This path can be accessed via `/example/routeA` due to the route decorator."
    
    @route(methods=['POST'])
    @example_decorator
    def route_decorator_default_to_function_name(self, param):
        """This is especially needed, when you have other decorators on this function,
        otherwise, FlaskClassy's routing won't work.
        
        I've also noticed that the decorator can't be two layers deep (eg. a decorator that takes input,
        to create a dynamic decorator). I'm not sure why for now.
        """
        return "This path can be accessed via `/example/route_decorator_default_to_function_name` (POST only)."
    
    @FlaskView.on_socket_event('join')
    def socket_can_be_called_anything_as_long_as_it_is_prefixed_by_socket(self, data):
        """
        :param data: anything that is sent via JS socket.io.
                     In this example, it's going to be `2`, then `{"a": 1}`
        """
        emit('event_name', 'sockets can also emit values')
        emit('broadcast_event', 'this message to everyone', broadcast=True)
        return 'sockets can return values'

cmds-base.py

import flask_classy


class IgnoredFlaskClassyFunction(Exception):
    pass


class FlaskView(flask_classy.FlaskView):
    # Set default setting for trailing_slash (so routes don't need to end with /)
    trailing_slash = False

    socket_cache_name = '_socket_rule_cache'    # To store socket event listeners

    @classmethod
    def build_rule(cls, rule, method=None):
        """Currently, this is the only way to add custom rules for ignoring
        function names within flask_classy views."""

        if rule.startswith('/socket_'):
            # To support socket functions in the class, we denote this prefix
            # to be purely socket functions.
            # Furthermore, raising this exception is the only way we can get
            # around vendor code of ignoring creating a route.
            raise IgnoredFlaskClassyFunction

        return super(FlaskView, cls).build_rule(rule, method)

    @classmethod
    def socket_shim(cls, function):
        """A shim function is needed, because `socket.on_event` passes all inputs to
        function parameters (ignoring the class' self parameter). Therefore, use
        a shim function to inject this.

        :param function: function
        :return: function
        """
        def callback(*args, **kwargs):
            return function(cls, *args, **kwargs)

        return callback

    @classmethod
    def add_socket_routes(cls, socket):
        """Register socket routes.
        
        :param socket: socketio object.
        """
        methods = getattr(cls, cls.socket_cache_name)
        for function_name in methods:
            function = methods[function_name][0]
            for event_name in methods[function_name][1:]:
                socket.on_event(
                    event_name,
                    cls.socket_shim(function),
                    namespace = cls.route_prefix + cls.route_base,
                )
        
    @classmethod
    def register(cls, *args, **kwargs):
        """Called in app.py to register endpoints"""

        # Socket integration
        socket = kwargs.get('socket', None)
        if socket and hasattr(cls, cls.socket_cache_name):
            cls.add_socket_routes(socket)
        
        del kwargs['socket']
        
        try:
            super(FlaskView, cls).register(*args, **kwargs)
        except IgnoredFlaskClassyFunction:
            pass


def route(path="", **options):
    """Decorator for flask-classy, that allows you to specify options, without
    specifying the path."""
    def decorator(f):
        local_path = path   # needed for scope reasons
        if local_path == '':
            local_path = f.__name__
        
        return flask_classy.route(local_path, **options)(f)

    return decorator

index.html

<!-- This is only a HTML file, so I can show that you need to import the dependency scripts -->
<script type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/2.0.4/socket.io.js"></script>
<script>
  function initializeSocket() {
    var socket = io('/api/example');
    socket.on('disconnect', function() {
      //  Without this, it will keep polling till forever.
      socket.disconnect();
    });
    
    socket.connect('http://' + document.domain + ':' + location.port);
    
    //  Handy command for chained functionality
    socket.addListener = function(eventName, callback) {
      socket.on(eventName, (data) => {
        callback(data);
      });
      
      return socket;
    };
    
    return socket;
  }
  
  var socket = initializeSocket();
  socket.addListener('event_name', function(data) {
    console.assert(data, 'sockets can also emit values');
  }).addListener('broadcast_event', function(data) {
    console.assert(data, 'this message to everyone');
  });
  
  socket.emit('join', 2);
  socket.emit('join', {a:1}, function(data) {
    console.assert(data === 'sockets can return values');
  });
</script>

requirements.txt

Flask==0.12.2
Flask-Classy==0.6.10
Flask-SocketIO==2.9.2