tornado memo

api_doc_gen.py

# Credit: https://github.com/hfaran/Tornado-JSON/blob/master/tornado_json/api_doc_gen.py
import re
import inspect
import types
import itertools

import tornado.web

from tornado.web import RequestHandler

HTTP_METHODS = ["get", "put", "post", "patch", "delete", "head", "options"]


def get_api_methods(route):
    """Return [(pattern, method, handler_class, handle_function)] from ``route``

    :type route: tuple|tornado.web.URLSpec
    :rtype: tuple
    :raises TypeError: If ``route`` is not a tuple or URLSpec
    """
    if isinstance(route, tuple):
        assert len(route) >= 2
        pattern, handler_class = route[:2]
    elif isinstance(route, tornado.web.URLSpec):
        pattern, handler_class = route.regex.pattern, route.handler_class
    else:
        raise TypeError("Unknown route type '{}'"
                        .format(type(route).__name__))

    methods = []
    route_re = re.compile(pattern)
    route_params = set(list(route_re.groupindex.keys()) + ['self'])
    for http_method in HTTP_METHODS:
        method = getattr(handler_class, http_method, None)
        if method:
            method_params = set(inspect.signature(method).parameters)

            if method_params == route_params:
                methods.append((pattern, http_method, handler_class, method))

    return methods


def _add_indent(lines, indent=4):
    if isinstance(lines, str):
        lines = lines.splitlines()

    indent_line = lambda l: " "*indent + l if l else ''

    return "\n".join([indent_line(l) for l in lines])


def generate_api_section(url, http_method, handle_fun):
    api_doc = inspect.getdoc(handle_fun)

    doc = (".. http:{http_method}:: {url}\n\n{api_doc}".format(
        url=url, http_method=http_method, api_doc=_add_indent(api_doc)))

    return doc


def get_api_docs(routes):
    """
    Generates Sphinx formatted API documentation using method's docstring

    :type  routes: [(url, RequestHandler), ...]
    :param routes: List of routes (this is ideally all possible routes of the
        app)
    :rtype: str
    :returns: generated GFM-formatted documentation
    """

    apis = list(itertools.chain(*[get_api_methods(r) for r in routes]))

    api_docs = []
    for url, http_method, handle_cls, handle_fun in apis:
        if issubclass(handle_cls, RequestHandler):
            api_docs.append(generate_api_section(url, http_method, handle_fun))

    return "\n\n\n".join(api_docs)


def api_doc_gen(routes):
    """Get and write API documentation for ``routes`` to file"""
    docs = get_api_docs(routes)
    return docs

async_http_client_unix_socket.py

# If you know the source, please notify me. I will add the source in the begin of the file.
import socket
from tornado.concurrent import TracebackFuture, return_future
from tornado import gen
from tornado.httpclient import AsyncHTTPClient
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from tornado.netutil import bind_unix_socket, Resolver
from tornado.web import RequestHandler, Application

class HelloHandler(Reimport re
import inspect
import types
import itquestHandler):
    def get(self):
        self.write('hello')

SOCKPATH = '/tmp/test.sock'

class UnixResolver(Resolver):
    def initialize(self, resolver):
        self.resolver = resolver

    def close(self):
        self.resolver.close()

    @gen.coroutine
    def resolve(self, host, port, *args, **kwargs):
        if host == 'unixsocket':
            raise gen.Return([(socket.AF_UNIX, SOCKPATH)])
        result = yield self.resolver.resolve(host, port, *args, **kwargs)
        raise gen.Return(result)

@gen.coroutine
def main():
    app = Application([('/', HelloHandler)])
    server = HTTPServer(app)
    server.add_socket(bind_unix_socket(SOCKPATH))

    resolver = UnixResolver(resolver=Resolver())
    AsyncHTTPClient.configure(None, resolver=resolver)

    response = yield AsyncHTTPClient().fetch('http://unixsocket/')
    print response.body

IOLoop.current().run_sync(main)

py3_async_await_test.py

import tornado.ioloop


async def add(a, b):
    return a + b

async def add_twice():
    c = await add(1, 2)
    d = await add(3, 4)
    return c + d


async def main():
    ans = await add_twice()
    print(ans)


if __name__ == "__main__":
    io_loop = tornado.ioloop.IOLoop.current()
    io_loop.run_sync(main)

py3_tornado5_aio_async_await.py

#!/usr/bin/env python3

# tornado == 5.0.0
# aioredis == 1.1.0

from tornado import gen
import tornado.ioloop
import aioredis


@gen.coroutine
def add_gen(a, b):
    return a + b


async def add(a, b):
    return a + b


async def add_twice():
    c = await add(1, 2)
    d = await add(3, 4)
    e = await add_gen(5, 6)
    return c + d + e


async def go():
    conn = await aioredis.create_connection(
        'redis://localhost')
    await conn.execute('set', 'my-key', 'value')
    val = await conn.execute('get', 'my-key')
    print(val.decode('utf8'))
    conn.close()
    await conn.wait_closed()


@gen.coroutine
def test_call_go():
    yield go()


async def main():
    ans = await add_twice()
    print(ans)
    await go()
    await test_call_go()


if __name__ == "__main__":
    io_loop = tornado.ioloop.IOLoop.current()
    io_loop.run_sync(main)

"""
$ docker run --rm -p 6379:6379 --name some-redis -d redis
e37e7a7b76a38e1a3e366e50153ba7c6c79d691d15e0964e8957add56bd5deed
$ python server.py
21
value
value
$ docker run -it --link some-redis:redis --rm redis redis-cli -h redis -p 6379
redis:6379> get my-key
"value"
redis:6379> exit
$ docker stop some-redis
some-redis
"""

server.py

#!/usr/bin/env python3

import logging

from tornado import gen
import tornado.ioloop
import tornado.web
import tornado.options
import tornado.websocket

app_log = logging.getLogger('tornado.application')


class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("Hello World!\n")


class EchoWebSocketHandler(tornado.websocket.WebSocketHandler):
    def check_origin(self, origin):
        return True

    def open(self, qid=None):
        app_log.debug("Open new websocket")
        if qid:
            app_log.info("New qid: " + str(qid))

    @gen.coroutine
    def on_message(self, message):
        app_log.debug("type: " + str(type(message)))
        app_log.debug("New websocket msg:" + str(message))
        self.write_message("Echo: " + message)

    def on_close(self):
        app_log.debug("Close websocket")


def init_log():
    import tornado.log
    tornado.log.enable_pretty_logging()
    for name in ['tornado.access', 'tornado', 'tornado.application',
                 'tornado.general']:
        logger = logging.getLogger(name)
        logger.setLevel(logging.DEBUG)


def make_app():
    handlers = [
        (r"/", MainHandler),
        (r"/websocket", EchoWebSocketHandler),
        (r"/websocket/(?P<qid>\d+)", EchoWebSocketHandler),
    ]

    return tornado.web.Application(handlers , debug=True)


application = make_app()


def main():
    PORT = 11000

    init_log()
    application.listen(PORT)
    app_log.info("Start the server to listen: " + str(PORT))
    tornado.ioloop.IOLoop.current().start()


if __name__ == "__main__":
    main()