Last active
April 22, 2018 08:55
-
-
Save yen3/cbaf189f73c2c0d00eec6ea0daae1f23 to your computer and use it in GitHub Desktop.
tornado memo
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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 | |
""" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment