使用tornado，我們可以做什么？

2019-11-14 16:56:16

字體：大中小

來源：轉載

供稿：網友

以下介紹都是建立在python2.x的基礎上面，tornado使用任意版本皆可。

如果我們需要對外提供一個http server(web api)/websocket server時，我們都可以使用tornado，以下是一個基于tornado的rest的應用簡介。

當我們下載好了tornado以后，可以按照tornado的文檔demo，復制一份監聽代碼過來，代碼如下：

import tornado.ioloopimport tornado.webclass MainHandler(tornado.web.RequestHandler):    def get(self):        self.write("Hello, world")application = tornado.web.Application([    (r"/", MainHandler),])if __name__ == "__main__":    application.listen(8888)    tornado.ioloop.IOLoop.instance().start()

View Code

這一段代碼在運行時就已經可以開始監聽并返回信息了，然后開始按照rest的規范開始改寫，我們首先需要對url的path進行定位，根據path執行相對應的代碼，則有了以下代碼：

import os.pathimport tornado.escapeimport tornado.ioloopimport tornado.optionsimport tornado.webimport tornado.websocketimport tornado.httpserverimport toolsimport applogicfrom config import configclass Application(tornado.web.Application):    def __init__(self):        handlers = [            (r"/.*?", WebHandler),        ]        settings = dict(            cookie_secret="__TODO:_GENERATE_YOUR_OWN_RANDOM_VALUE_HERE__",            template_path=os.path.join(os.path.dirname(__file__), "templates"),            static_path=os.path.join(os.path.dirname(__file__), "static"),            xsrf_cookies=False        )        tornado.web.Application.__init__(self, handlers, **settings)class WebHandler(tornado.web.RequestHandler):    def get(self):        path = self.request.path        command = path[path.rfind('/') +1:]        result = applogic.execute(self, command, self.get_argument)        if result:            to_message = tools.dumps(result)            self.write(to_message)    def post(self):        path = self.request.path        command = path[path.rfind('/') +1:]        obj = tools.loads(self.request.body)        result = applogic.execute(self, command, obj)        if result:            to_message = tools.dumps(result)            self.write(to_message)def main():    app = Application()    http_server = tornado.httpserver.HTTPServer(app)    http_server.bind(config.default['main_port'], config.default['host'])    tornado.ioloop.IOLoop.instance().start()

View Code

我們對path做了最后一級目錄的截取，由于客觀原因，多級目錄的則需要看客們自行實現。

applogic是我們的具體的邏輯代碼門面類，因為是腳本語言的關系，所以這里使用一個相對巧妙的ioc，代碼如下：

def execute(handler, command, obj):    if main_map.has_key(command):        result = main_map[command](handler, obj)        if result:            response.send(handler, result)    else:        handler.send_error(404)def create(handler, obj):    flag = papercache.push(obj)    result = {}    if flag:        result = {            'code':0        }    else:        result = {            'code':1,            'errorMsg':'param is error'        }    return resultdef grab(handler, obj):    id = obj.has_key('id') and obj['id'] or None    result = {        'code':1,        'errorMsg':'redpaper is empty',        'money':0    }    if id:        money = papercache.pop_queue(id)        if money:            result = {                'money' : money,                'code': 0            }    return resultmain_map = {    'create':create,    'grab':grab}

View Code

我們將具體的執行邏輯，放在定義好的function里面，然后將function的引用放在一個字典里面，然后根據command（最后一級目錄對應的字符串），來執行具體的代碼。

我們將返回的數據做了一個封裝，因為實際應用里面的數據格式，可能采用的是json，或者是其它自定義的協議，所以我們有一個response的封裝，代碼如下：

def send(handler, obj):    if type(handler) is not None:        obj = tools.dumps(obj)        handler.write(obj)

View Code

tools的代碼如下：

import jsondef dumps(obj):    obj = toUnicode(obj)    if obj:        obj = json.dumps(obj)        obj = str(obj)    return objdef loads(obj):    if obj:        obj = json.loads(str(obj))        obj = toUtf8(obj)    return obj

View Code

因為可能存在中文的關系，所以加了一個Utf8的轉換，，基于websocket的也是同理，則在監聽的時候，使用WebSocketHandle，代碼如下：

import os.pathimport tornado.escapeimport tornado.ioloopimport tornado.optionsimport tornado.webimport tornado.websocketimport tornado.httpserverimport toolsimport applogicfrom config import configclass Application(tornado.web.Application):    def __init__(self):        handlers = [            (r"/web", MainHandler)            (r"/.*?", WebHandler),        ]        settings = dict(            cookie_secret="__TODO:_GENERATE_YOUR_OWN_RANDOM_VALUE_HERE__",            template_path=os.path.join(os.path.dirname(__file__), "templates"),            static_path=os.path.join(os.path.dirname(__file__), "static"),            xsrf_cookies=False        )        tornado.web.Application.__init__(self, handlers, **settings)class MainHandler(tornado.websocket.WebSocketHandler):    def allow_draft76(self):        return True    def open(self):        token = self.get_cookie('token')        if not token:            PRint "long not token"            self.close()        else:            print('connect')    def on_close(self):        print('close');    def on_message(self, message):        obj = tools.loads(message)        command = obj.has_key('command') and obj['command'] or None        body = obj.has_key('body') and obj['body'] or None        result = applogic.execute(self, command, body)        if result:            to_message = tools.dumps(result)            self.write(to_message)        returnclass WebHandler(tornado.web.RequestHandler):    def get(self):        path = self.request.path        command = path[path.rfind('/') +1:]        result = applogic.execute(self, command, self.get_argument)        if result:            to_message = tools.dumps(result)            self.write(to_message)    def post(self):        path = self.request.path        command = path[path.rfind('/') +1:]        obj = tools.loads(self.request.body)        result = applogic.execute(self, command, obj)        if result:            to_message = tools.dumps(result)            self.write(to_message)def main():    app = Application()    http_server = tornado.httpserver.HTTPServer(app)    http_server.bind(config.default['main_port'], config.default['host'])    tornado.ioloop.IOLoop.instance().start()