Writing an API with Tornado
September 14, 2015 / Developers, Tutorials, Uncategorized

CALLING ALL DEVELOPERS!
Our API servers at indico are written in Python using Tornado as a server framework. Tornado is a Python web framework and asynchronous networking library originally developed at FriendFeed. By using non-blocking network I/O, Tornado can scale to tens of thousands of open connections. Some of the APIs we provide take some time to process so we take advantage of Tornado’s scalability and non-blocking I/O to support these longer user connections.
Given indico’s use of Tornado, I would simply like to share a boilerplate to spinning up an API server using Tornado.
Overview
Assuming you’ve already setup your environment with Python (2.7) and pip, go ahead and checkout the Github repository. It is worth noting that the package currently has a coverage of 100% with nosetests (branch-inclusive) — meaning you can rest easy using the functionality provided.
Top-level directory
setup.py -> python setup.py develop -> installs the package and requirements that are necessary to run the server. req.txt -> dependencies list setup.cfg -> nosetests -> runs the test suite for the package indico -> package for the server
Follow the README instructions on the repository to get setup.
Modules
Let’s go over the modules that are in the package.
db -> database helper functions for each individual database table / collection
error -> custom errors for error handling and sending appropriate server responses
routes -> handlers that specify routing logic for the server. most of the server logic is here.
tests -> unittest tests
utils -> utilities that make lives easier (more later)
Running the Server
To run the server, run the following:
$ python -m indico.server
Additional configs are in config.py — right now, it simply has a specified port and MongoDB URI.
Deep Dive — Routes
The routes module contains RequestHandlers that the server uses to receive incoming requests and reply with appropriate responses. In the boilerplate source, there exists a RequestHandler for route paths beginning with /authand /user. Each of these handlers extend the general handler found inhandler.py. See the comments attached to the code.
""" Indico Request Handler """ import json, traceback from bson.objectid import ObjectId import tornado.web from indico.error import IndicoError, RouteNotFound, ServerError from indico.utils import LOGGER class JSONEncoder(json.JSONEncoder): def default(self, o): if isinstance(o, ObjectId): return str(o) return json.JSONEncoder.default(self, o) class IndicoHandler(tornado.web.RequestHandler): @tornado.web.asynchronous def post(self, action): try: # Fetch appropriate handler if not hasattr(self, str(action)): raise RouteNotFound(action) # Pass along the data and get a result handler = getattr(self, str(action)) handler(self.request.body) except IndicoError as e: self.respond(e.message, e.code) except Exception as e: LOGGER.error( "\n\n==== INDICO SERVER ERROR ====\n%s\n%s\n", __file__, traceback.format_exc() ) error = ServerError() self.respond(error.message, error.code) def respond(self, data, code=200): self.set_status(code) self.write(JSONEncoder().encode({ "status": code, "data": data })) self.finish()
This general handler abstracts away the logic to parse, route, and respond to POST requests. Child request handlers only need to define functions for each `action` supported and return the response.
Example:
""" IndicoService User Route Creating and Maintaining Users """ import indico.db.user_db as UserDB from indico.utils import unpack, mongo_callback, type_check from indico.routes.handler import IndicoHandler from indico.utils.auth.auth_utils import auth class UserHandler(IndicoHandler): @auth @unpack("update_user") @type_check(dict) def update(self, update_user): @mongo_callback(self) def update_callback(result): self.respond(result) UserDB.update({ "$set": update_user }, self.user_id, update_callback) UserRoute = (r"/user/(?P[a-zA-Z]+)?", UserHandler)
This route defines `/user/update` — which as the naming suggests updates a user with the provided information. The logic is concise and easy to manage thanks to the wrappers developed in the utils module.
Deep Dive — Testing
I wrote an earlier post about testing with Motor, Tornado, and unittest. Since then, I have discovered that the logic I developed there had already been packaged into Tornado’s own testing tools. The current repo here reflects changes using Tornado’s tools. They are quite nice as there are no more runaway processes and unresolved connections.
I have built some additional logic to wrap those tools:
class ServerTest(AsyncHTTPTestCase): def setUp(self): super(ServerTest, self).setUp() name = str(self).split(" ") self.name = name[0].replace("_","") + name[1].split(".")[-1][:-1] indico.db.CLIENT = MotorClient(MONGODB) indico.db.mongodb = indico.db.CLIENT[self.name] def get_app(self): return Application(self.routes, debug = False) def post(self, route, data, headers=HEADERS): result = self.fetch("/%s" % route, method = "POST", body = json.dumps(data), headers=headers).body try: return json.loads(result) except ValueError: raise ValueError(result) def tearDown(self): indico.db.CLIENT.drop_database(self.name) super(ServerTest, self).tearDown()
If you only need to test asynchronous calls to the database, simply extend only AsyncTestCase provided in tornado.testing and remove the accompanying get_app and post.
Questions/comments/random trivia facts? Feel free to email me at chris@indico.io.