Skip to content

Instantly share code, notes, and snippets.

@HarryPehkonen
Last active April 14, 2017 08:52
Show Gist options
  • Save HarryPehkonen/5423354 to your computer and use it in GitHub Desktop.
Save HarryPehkonen/5423354 to your computer and use it in GitHub Desktop.
Writing your own asynchronous library using Tornado. Library.ask is called asynchronously, and just returns a string. Library.check_failure is used to make sure failures are detected when unit-testing.
import tornado.gen
class Library():
@tornado.gen.engine
def ask(self, question, callback=None):
# do a computation that doesn't block for a long time.
result = "Why would you want to know about '{0}'".format(question)
error = None
callback(result, error)
@tornado.gen.engine
def check_failure(self, callback=None):
error = "This is the error"
callback(None, error)
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.gen
import tornado.options
tornado.options.define("port", default=8000, help="Run on the given port", type=int)
import library
mylibrary = library.Library()
class WhateverHandler(tornado.web.RequestHandler):
@tornado.web.asynchronous
@tornado.gen.engine
def get(self):
print("get")
future = tornado.gen.Task(mylibrary.ask, question="What's this?")
response = yield future
(result, error) = response.args
if error:
raise tornado.web.HTTPError(500, error)
else:
self.set_header("Content-type", "text/html")
self.write("<p>result:</p>")
self.write("<pre>{0}</pre>".format(result))
self.write("<p>error:</p>")
self.write("<pre>{0}</pre>".format(error))
self.finish()
tornado.options.parse_command_line()
app = tornado.web.Application(
handlers = [
(r'/', WhateverHandler),
],
)
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(tornado.options.options.port)
ioloop = tornado.ioloop.IOLoop.instance()
ioloop.start()
import unittest
import library
import tornado.gen
class TestLibrary(unittest.TestCase):
def setUp(self):
self.my_library = library.Library()
def tearDown(self):
pass
@tornado.gen.engine
def test_ask(self):
question = "What's up?"
future = tornado.gen.Task(self.my_library.ask, question=question)
response = yield future
(result, error) = response.args
if error:
self.fail("Got an error: {0}".format(error))
else:
self.assertEqual(result, "Why would you want to know about '{0}'".format(question))
@tornado.gen.engine
def test_failure(self):
future = tornado.gen.Task(self.my_library.check_failure)
response = yield future
(result, error) = response.args
if error:
pass
else:
self.fail("No error when expected one!")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment