Tornado 和现在的主流 Web 服务器框架（包括大多数 Python 的框架）有着明显的区别：它是非阻塞式服务器，而且速度相当快。得利于其非阻塞的方式和对 epoll 的运用
基本操作
tornado_base.py
# Author:song import tornado.ioloop import tornado.web #配置信息 settings = { 'static_path':'static', 'static_url_prefix':'/song/',#别名 'template_path':'templates', } class MainHandler(tornado.web.RequestHandler): def get(self): self.write("Hello, world") #返回数据 class LoginHandler(tornado.web.RequestHandler): def get(self): #self.get_argument('p') 获取请求参数，相当于django的request.GET.get('p') #self.get_body_argument('name') 获取post内容信息 #self.set_cookie('k1','v1')设置cookie #self.set_header('h1','v1')设置请求头 self.render("login.html",**{'k1':'song','k2':'shi','k3':[1,2,3,4],'k4':{'name':'a','age':18}}) #返回模板 def post(self, *args, **kwargs): print(self.get_argument('user')) self.redirect('https://home.cnblogs.com/u/master-song/')#重定向 #路由规则 application = tornado.web.Application([ (r"/index", MainHandler), (r"/login", LoginHandler), ],**settings) if __name__ == "__main__": #创建socket对象，将其加入select application.listen(8888) #开始循环监听 tornado.ioloop.IOLoop.instance().start()
模板login.html
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <link rel="stylesheet" href="/song/comment.css"/> <title>Title</title> </head> <body> <h1>login</h1> {{k1}} {{k2}} <h1>列表</h1> <ul> {% for i in k3 %} <li>{{i}}</li> {% end %} </ul> <h1>字典</h1> {{k4}}<br> {{k4['name']}}<br> {{k4.get('name')}} <ul> {% for k,v in k4.items() %} <li>{{k}}:{{v}}</li> {% end %} <form method="post" action="login"> <input type="text" name="user"/> <input type="submit" value="提交"/> </ul> </body> </html>
comment.css 只写了一句
body{
background-color: blue;
}
login.html渲染图
Tornado默认提供的 UIMethod 和 UIModule，实现类似于Django的simple_tag的功能。
创建两个py文件
# Author:song def test1(self): return '测试前端使用函数'
# Author:song from tornado.web import UIModule from tornado import escape class test2(UIModule): def embedded_css(self): return 'body{color:blue}' #网页引入<style type="text/css">body{color:blue}</style> def css_files(self): return 'a.css'# 引入<link href="/song/a.css" type="text/css" rel="stylesheet"> def javascript_files(self): return 'sk.js' #<script src="/song/sk.js" type="text/javascript"></script> def render(self, *args, **kwargs): print(args)#获取内容（123） print(kwargs)#{} # return escape.xhtml_escape('<h1>测试代码段前端渲染</h1>') #原样显示 <h1>测试代码段前端渲染</h1> return '<h1>测试代码段前端渲染</h1>' #加载效果后显示
在tornado_base.py中
import uimethods as mt
import uimodules as md
settings = {
'static_path':'static',
'static_url_prefix':'/song/',
'template_path':'templates',
'ui_methods':mt,
'ui_modules':md, #加上这两行
}
web组件的定制
1.session实现机制
# Author:song import uuid #基于内存的写法 class Session(object): container = {} def __init__(self,handler): # 获取用户cookie，如果有，不操作，否则，给用户生成随即字符串 # - 写给用户 # - 保存在session nid = handler.get_cookie('session_id') if nid: if nid in Session.container: pass else: nid = str(uuid.uuid4()) Session.container[nid] = {} else: nid = str(uuid.uuid4()) Session.container[nid] = {} handler.set_cookie('session_id', nid, max_age=1000)#设置cookie并设超时时间 # nid当前访问用户的随即字符串 self.nid = nid # 封装了所有用户请求信息 self.handler = handler def __setitem__(self, key, value): Session.container[self.nid][key] =value def __getitem__(self, item): return Session.container[self.nid].get(item) def __delitem__(self, key): del Session.container[self.nid][key] #基于redis class RedisSession(object): def __init__(self,handler): # 获取用户cookie，如果有，不操作，否则，给用户生成随即字符串 # - 写给用户 # - 保存在session nid = handler.get_cookie('session_id') if nid: if nid in Session.container: pass else: nid = str(uuid.uuid4()) # Session.container[nid] = {} # 连接redis写值 # redis 服务器IP，端口（6379） # 根据Nid是字符串 => 6871237123 # 6871237123 % 3 = 0,1,2 # ['10.1.11.2','10.1.11.3','10.1.11.4'] else: nid = str(uuid.uuid4()) # Session.container[nid] = {} # 连接redis写值 handler.set_cookie('session_id', nid, max_age=1000) # nid当前访问用户的随即字符串 self.nid = nid # 封装了所有用户请求信息 self.handler = handler def __setitem__(self, key, value): #Session.container[self.nid][key] =value pass def __getitem__(self, item): #return Session.container[self.nid].get(item) pass def __delitem__(self, key): # del Session.container[self.nid][key] pass
# Author:song # session_module = "RedisSession" # 只允许两种Session,RedisSession session_module = "Session"
# Author:song # Author:song import tornado.ioloop import tornado.web from tornado_session import session_key #配置信息 settings = { 'static_path':'static', 'static_url_prefix':'/song/', 'template_path':'templates', } class BaseHandler(object): def initialize(self): # 获取用户cookie，如果有，不操作，否则，给用户生成随即字符串 # - 写给用户 # - 保存在session from tornado_session import session session_module = getattr(session, session_key.session_module)#反向解析 self.session = session_module(self) super(BaseHandler,self).initialize() class IndexHandler(BaseHandler, tornado.web.RequestHandler): def get(self): if self.session['is_login']: self.write("Hello, world") else: self.redirect('/login') class LoginHandler(BaseHandler,tornado.web.RequestHandler): def get(self, *args, **kwargs): self.render('login.html') def post(self, *args, **kwargs): v = self.get_argument('user') if v == 'song': self.session['is_login'] = True self.redirect('/index') else: self.redirect('/login') application = tornado.web.Application([ (r"/index",IndexHandler ), (r"/login", LoginHandler), ],**settings) if __name__ == "__main__": application.listen(8888) tornado.ioloop.IOLoop.instance().start()
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>tornado-session</title> </head> <body> <form method="post" action="login"> <input type="text" name="user"/> <input type="submit" value="提交"/> </form> </body> </html>
如果选择使用redis，可能需要用到分布式
form简单表单验证
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>tornado_form</title> </head> <body> <form method="post" action="login"> <!--<input type="text" name="user"/>--> <!--<input type="email" name="email"/>--> {% raw obj.user %} {% raw obj.email %} <input type="submit" value="提交"/> </form> </body> </html>
# Author:song import tornado.web import re settings = { 'static_path': 'static', 'static_url_prefix': '/song/', 'template_path':'templates', } class StringField: def __init__(self,name): self.rex = "^w+$" self.name = name self.value = '' self.error = "" def __str__(self): return "<input type='text' name='%s' value='%s' />" %(self.name,self.value,) class EmailField: def __init__(self,name): self.rex = "^w+@.*$" self.name = name self.value = '' self.error = "" def __str__(self): return "<input type='text' name='%s' value='%s' />" %(self.name,self.value,) class LoginForm: def __init__(self): self.user = StringField(name='user') self.email = EmailField(name='email') def is_valid(self,handler): value_dict = {} flag = True for k,v in self.__dict__.items(): inp = handler.get_argument(k) rex = re.match(v.rex,inp) v.value = inp if rex: value_dict[k] = inp else: v.error = '%s 错误了..' %k flag = False return flag,value_dict class LoginHandler(tornado.web.RequestHandler): def get(self, *args, **kwargs): obj = LoginForm() self.render('login.html',**{'obj': obj}) def post(self, *args, **kwargs): obj = LoginForm() valid,value_dict = obj.is_valid(self) print(valid,value_dict) if valid: print(value_dict) self.write('welcome') else: return self.render('login.html',**{'obj': obj}) application = tornado.web.Application([ (r"/login", LoginHandler), ],**settings) if __name__ == "__main__": application.listen(8888) tornado.ioloop.IOLoop.instance().start()