python tornado 路由 web API式路由前后分离

您所在的位置：网站首页 › tornado框架原理 › python tornado 路由 web API式路由前后分离

python tornado 路由 web API式路由前后分离

#python tornado 路由 web API式路由前后分离| 来源: 网络整理| 查看: 265

目前流行前后端分离, 前端一般用Vue或React 做前端, 后端用C#,python,java,php的都可以.

而python 系的web框架有 , Django 和Flask, 还有 Tornado.

其中 Tornado 特别适合Api开发, 灵活轻量级, 自非阻塞式带服务器 , 据说性能强劲, 而且速度相当快。得益于其非阻塞的方式和对 epoll 的运用，Tornado 每秒可以处理数以千计的连接，这意味着对于实时 Web 服务来说，Tornado 是一个理想的 Web 框架. 更适合做前后端分离的后端. 其它两个框架都有点臃肿的感觉. Flask不能支持多端口. Django支持的是过时的Mvc, 很多用不到的东西影响性能, 而且上手要点时间. 所以我选择用Tornado做后端,

更多的好处咱就不多说了. 请移步到这里看 https://www.cnblogs.com/liujianzuo888/articles/5996114.html

Tornado官方给的默认路由示例是按照RestFull 形式来实现的. 按照get,post,方法来区分调用不同的方法.

# 实现一个最简单的tornado服务 import tornado.ioloop #监听循环 import tornado.web #tornado包核心模块 import tornado.httpserver #非阻塞服务器 import tornado.options # class MainHandler(tornado.web.RequestHandler): def get(self): self.write('hello world get') def post(self): self.write('hello world post') application = tornado.web.Application( handlers=[ (r'/',MainHandler), #路由表 ] #列表，列表中包含元组。配置路径信息 ) #实例化操作 #可以传入很多参数 #配置参数项 if __name__ == '__main__': #用来做测试，自己调用 tornado.options.parse_command_line() #打印请求信息(日志信息),不加这一行后端不会有日志信息 http_server = tornado.httpserver.HTTPServer(application) http_server.listen(8080) #监听，绑定端口，服务器由主动变为被动 tornado.ioloop.IOLoop.current().start() #开启监听循环 #单进程单线程epoll #epoll ,主动监听，把监听的任务交给操作系统，不断询问。监听循环

如上的代码是一般常见的代码.默认使用方法. 在工程使用中. 一般会将MainHandler 放到一个单独的py文件中. 一般没人会把所有的代码都放在一个启动文件中吧…

但是即便分离了代码, 仍然是一个 url地址对应一个Handler Class 能不能地址映射到 Handler中的不同的函数呢? 例如 /api/user/login 对应 UserHandler中的 login 函数 /api/user/logout 对应 UserHandler中的 logout 函数 /api/user/userinfo 对应 UserHandler中的 userinfo 函数

经过半天的百度查找,始终没有找到合适的方案, (我发现百度真是越来越不中用了) 甚至到最后都打算自己写一个路由器了…

下班后, 又尝试着再github中找, 结果从好几种路由方式中, . 最后终于找到了比较合适的方案,

https://github.com/ginking/tornado_router 其核心代码只有一个文件,router.py

于是我把它复制到我的项目根目录下,改名成 ApiRouter.py 在它的 router.py 代码之上, 我只增加了一行. 更改后的源代码如下

# 此类来自于 https://github.com/ginking/tornado_router import logging import base64 import traceback import functools import json import tornado.gen import tornado.web import tornado.escape logger = logging.getLogger(__name__) class Router(): def __init__(self, base_handler, redirect=True): self._handlers = [] self._requests = [] self._base_handler = base_handler # redirect if true, not return redirect_url self._redirect = redirect @property def handlers(self): return self._handlers @property def requests(self): return self._requests # authentication wrapper def _auth_wrap(self, f): @functools.wraps(f) @tornado.gen.coroutine def auth_request(handler): user = yield handler.get_cur_user() if not user: redirect_url = handler.get_login_url() + '?referrer=' + str(base64.b64encode(handler.request.uri.encode('ascii')))[2:-1] if self._redirect: # redirect handler.redirect(redirect_url) else: # return json response handler.set_header('Content-Type', 'application/json') handler.write(tornado.escape.json_encode({'redirect': redirect_url})) return yield f(handler) return auth_request # json wrapper def _json_wrap(self, f): @functools.wraps(f) @tornado.gen.coroutine def json_request(handler): # todo: there's no need to prepare dict input?? try: resp = yield f(handler) handler.set_header('Content-Type', 'application/json') handler.write(tornado.escape.json_encode(resp)) except Exception as e: raise tornado.web.HTTPError(500, str(e)) # todo: raise HTTPError #handler.logger.warn(str(e) + '\n' + repr(traceback.format_stack())) #traceback.print_exc() return json_request # route def route(self, method='get', url=None, auth=False, json=False, xsrf=True): # self.application.settings.get("xsrf_cookies") if method.upper() not in tornado.web.RequestHandler.SUPPORTED_METHODS: raise ValueError('invalid HTTP method {} found! tornado only supports HTTP methods in {}'.format( method, tornado.web.RequestHandler.SUPPORTED_METHODS)) def req_wrap(f): if not self._base_handler: raise RuntimeError('base_handler must be initialized!') class InnerHandler(self._base_handler): pass req = self._json_wrap(tornado.gen.coroutine(f)) if json else tornado.gen.coroutine(f) req = self._auth_wrap(req) if auth else req setattr(InnerHandler, method.lower(), req) if not xsrf: setattr(InnerHandler, "check_xsrf_cookie", lambda self: None) InnerHandler.logger = logging.getLogger(f.__name__) f_url = url if not f_url: f_url = '/' + f.__name__ self._handlers.append((f_url, InnerHandler)) @functools.wraps(f) def request(): pass self._requests.append(request) return request return req_wrap class BaseHandler(tornado.web.RequestHandler): _login_url = '/login' def get_login_url(self): return self._login_url @tornado.gen.coroutine def get_cur_user(self): if not hasattr(self, '_cur_user'): self._cur_user = None current_user = self._cur_user if not current_user: user_cookie = self.get_secure_cookie('user', max_age_days=1) if not user_cookie: return None current_user = json.loads(user_cookie.decode('utf-8')) self._cur_user = current_user return self._cur_user @tornado.gen.coroutine def set_cur_user(self, user): self.set_secure_cookie('user', json.dumps(user).encode('utf-8'), expires_days=1, httponly=True) self._cur_user = user @tornado.gen.coroutine def clear_cur_user(self): self.clear_cookie('user') self._cur_user = None #全局路由注册器 GloabelRouter = Router(base_handler=BaseHandler)

我只增加了最后的一句话

#全局路由注册器 GloabelRouter = Router(base_handler=BaseHandler)

这样在使用的时候就很方便了.

auth.py文件是具体业务逻辑的代码如下.

# -*- coding: UTF-8 -*- import os import os.path import tornado.httpserver import tornado.ioloop import tornado.options import tornado.web import json from ApiRouter import GloabelRouter # class LoginHandler(tornado.web.RequestHandler): @GloabelRouter.route(method='post', url='/api/auth/login', auth=False) def Login_Post(handler): username= handler.get_argument("username",default="") password= handler.get_argument("password",default="") .... return " 登录成功" @GloabelRouter.route(method='post', url='/api/auth/logout', auth=False) def Logout_Post(handler): username= handler.get_argument("username",default="") password= handler.get_argument("password",default="") ...... return " 退出成功" ..... ..... .....

如此简单的定义很多函数即可. 在每个函数的上面加上注解. 方便又轻松. 其他的页面也是如此的道理, 只需要引入然后加上注解

from ApiRouter import GloabelRouter @GloabelRouter.route(method='post', url='/api/auth/logout', auth=False)

至于如何在主程序中使用还需要改成下面的代码 StartWeb.py

#完整打分流程V2 import os import os.path # import sys # sys.path.append('../') import tornado.httpserver import tornado.ioloop import tornado.options import tornado.web import tornado.autoreload from ApiRouter import GloabelRouter import cv2 import json import yolov5.models import common.Grade import uuid import base64 import ssl #解决yolov5的反序列化问题 import os,sys root_path = os.getcwd() sys.path.insert(0,root_path+"/yolov5") #解决 tornado 取得中文出现多余字符的问题. # print(sys.getdefaultencoding()) #reload(sys) #sys.setdefaultencoding('utf-8') from handlers import Auth#这里的Auth是文件名 from tornado.options import define, options # windows 系统下 tornado 使用使用 SelectorEventLoop 解决一个bug start import platform if platform.system() == "Windows": import asyncio asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) # windows 系统下 tornado 使用使用 SelectorEventLoop 解决一个bug end # print(context.options) # 定义端口用于指定HTTP服务监听的端口 # 如果命令行中带有port同名参数则会称为全局tornado.options的属性，若没有则使用define定义。 define("port", type=int, default=443, help="run on the given port") # 调试模式 define("debug", type=bool, default=True, help="debug mode") static_path = os.path.join(os.path.dirname(__file__), 'static') static_path = static_path.replace("\\","/") chainpath = os.getcwd()+ "/https/ai.***.com_chain.crt" crtpath = os.getcwd()+ "/https/ai.***.com_public.crt" keypath = os.getcwd()+ "/https/ai.***.com.key" #print(crtpath,keypath) if __name__ == '__main__': tornado.options.parse_command_line() settings = { 'cookie_secret': base64.b64encode(uuid.uuid4().bytes + uuid.uuid4().bytes) } app = tornado.web.Application( handlers = GloabelRouter.handlers, **settings, # handlers=[ # (r'/', OnlineRecognition.OnlineRecognitionHandler), # (r'/OnlineRecognition', OnlineRecognition.OnlineRecognitionHandler), # (r'/OnlineRecognitionYolo5', OnlineRecognitionYolo5.OnlineRecognitionYolo5Handler), # (r'/XiuZheng', XiuZheng.XiuZhengHandler),#修整AI算法的识别名称 # (r'/api/DatasetManage', DatasetManage.DatasetManageHandler),# # (r'/api/auth/login', Auth.LoginHandler),# # (r'/api/auth/logout', Auth.LogoutHandler),# # (r'/api/user/info', Auth.UserInfoHandler),# # ], template_path=os.path.join(os.path.dirname(__file__), "templates"), settings = { 'static_path' : static_path ,'debug' : True, "compiled_template_cache":False }, static_path =static_path ) ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH, cafile=chainpath) # ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2) #这种是强制使用TLSV1.2协议 ssl_ctx.verify_mode = ssl.CERT_NONE ssl_ctx.check_hostname =False ssl_ctx.load_cert_chain(crtpath, keypath) http_server = tornado.httpserver.HTTPServer(app, ssl_options = ssl_ctx # ssl_options={ # "certfile": crtpath, # "keyfile": keypath # } ) print("web https 服务正在监听",options.port) http_server.listen(options.port) print("web https 服务已启动...") #8181 再起一个服务器 http_server2 = tornado.httpserver.HTTPServer(app) print("web http 服务正在监听",8181) http_server2.listen(8181) print("web http 服务已启动...") # tornado.ioloop.IOLoop.instance().start() instance = tornado.ioloop.IOLoop.instance() tornado.autoreload.start(5) instance.start()

这段代码里面包含了, 1.如何启用https, 2.如何同时监听https和http两个端口… 3.如果你想, 甚至可以通知监听10 几个以上的端口. 4.如何设置静态资源路径

与本文相关的关键代码是

from handlers import Auth#这里的Auth是文件名 app = tornado.web.Application( handlers = GloabelRouter.handlers, #GloabelRouter.handlers 中包含了很多的注册信息 **settings, ..... )

【本文地址】

python tornado 路由 web API式路由前后分离

python tornado 路由 web API式路由前后分离

今日新闻

推荐新闻