Last active
April 14, 2017 08:52
-
-
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.
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.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) |
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.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() |
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 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