minimal tornado setup with rest and websocket api through redis pubsub

tornado_redis_pubsub_api.py

#!/usr/bin/env python3

import json
import signal
import sys

import redis
import tornado.options as opt
from tornado.ioloop import IOLoop, PeriodicCallback
from tornado.log import app_log
from tornado.web import Application, RequestHandler
from tornado.websocket import WebSocketHandler


class PubSubHandler(WebSocketHandler):

    def __init__(self, *args, **kw):
        super().__init__(*args, **kw)
        self.redis = redis_session()
        self.pubsub = self.redis.pubsub(ignore_subscribe_messages=True)
        self.pubsub.subscribe(keep_alive=lambda: True)
        self.callback = PeriodicCallback(self.pubsub.get_message, 100, 0.1)
        self.callback.start()

    def open(self):
        super().open()
        self.pubsub.subscribe(channel=self.forward)

    def forward(self, message):
        self.write_message(message['data'])

    def check_origin(self, origin):
        return True

    def on_message(self, message):
        self.write_message(json.dumps({'OK': 200}))

    def on_close(self):
        self.callback.stop()
        self.pubsub.close()


class APIHandler(RequestHandler):

    def set_default_headers(self):
        self.set_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
        self.set_header('Access-Control-Allow-Headers', 'Content-Type')
        self.set_header('Access-Control-Allow-Origin', '*')

    def get(self):
        self.write(json.dumps({'OK': 200}))

    def post(self):
        redis = redis_session()
        redis.publish('channel', self.request.body)
        self.write(json.dumps({'OK': 200}))

    def options(self):
        self.set_status(200)


class FallbackHandler(RequestHandler):

    def get(self, *args):
        self.set_status(200)


def define_options():
    opt.define('port', type=int, default=8040)
    opt.define('debug', type=bool, default=True, group='app')
    opt.parse_command_line()
    opt.define('websocket_ping_interval', type=int, default=1, group='app')
    opt.define('redis_host', type=str, default='127.0.0.1', group='redis')
    opt.define('redis_port', type=int, default=6379, group='redis')
    opt.define('redis_db', type=int, default=0, group='redis')


def redis_session():
    return redis.Redis(
        **{k.replace('redis_', ''): v for k, v in
           opt.options.group_dict('redis').items()})


def main():
    define_options()
    app = Application(
        [(r'/websocket', PubSubHandler),
         (r'/api', APIHandler),
         (r'.*', FallbackHandler)],
        **opt.options.group_dict('app'))
    app.listen(opt.options.port)
    app_log.info('tornado listening at 127.0.0.1:%s', opt.options.port)
    ioloop = IOLoop.current()

    def shutdown(*args):
        app_log.info('tornado shutting down')
        ioloop.stop()
        app_log.info('exit')
        sys.exit(0)

    signal.signal(signal.SIGTERM, shutdown)

    try:
        ioloop.start()
    except KeyboardInterrupt:
        shutdown()


if __name__ == '__main__':
    main()