Tornado框架——2
目录：
简介
本文介绍了cookies和session，xss和csrf以及模板系统
tornad内置cookies功能
cookies可以保存到浏览器端，下次访问的时候带着cookie来访问。
tornado代码
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
import time
class LoginHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
self.render("login.html")
def post(self, *args, **kwargs):
username = self.get_argument('username',None)
passwd = self.get_argument('passwd',None)
print '%(username)s %(passwd)s' % locals()
if username == 'why' and passwd == '123456':
r = time.time() + 10
self.set_cookie('auto', '1',expires=r)
self.redirect('/home')
else:
self.render('login.html')
class LogoutHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
self.set_cookie('auto','0')
self.redirect('/login')
class HomeHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
cookie = self.get_cookie('auto')
if cookie == '1':
self.render('home.html')
else:
self.redirect('/login')
settings = {
'template_path': 'template', #配置模板路径
'static_path': 'static', #静态文件路径
}
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/login", LoginHandler),
(r"/logout", LogoutHandler),
(r"/home", HomeHandler),
], **settings) #加载settings
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
login.html代码
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form action="/login" method="post">
<input type="text" name="username">
<input type="password" name="passwd">
<input type="submit" value="tijiao">
</form>
</body>
</html>
home.html代码
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<a href="/logout">退出</a>
保密内容
</body>
</html>
登录时获取的cookies
退出时获取的cookies
方法源码
def set_cookie(self, name, value, domain=None, expires=None, path="/",
expires_days=None, **kwargs):
"""Sets the given cookie name/value with the given options.
Additional keyword arguments are set on the Cookie.Morsel
directly.
See https://docs.python.org/2/library/cookie.html#Cookie.Morsel
for available attributes.
"""
设置cookies立即过期
self.set_cookie('auto', '1', expires=)
加密使用
设置和获取cookies的方法为set_secure_cookie()和get_secure_cookie()
在setting中配置
'cookie_secret':'ashjkfhkjajsflkaj'
加密时登录获取的cookies
加密的本质
写入cookies
- 将值进行base64加密(可以被反解)
- 对除值以外的内容(当前时间和加密字符串)进行签名，哈希算法（无法逆向解析）
- 拼接加密值和签名
读取cookies
- 读取加密值和签名
- 对签名进行验证
- base64解密，获取值内容
- 对比检测合法性
cookies的功能
cookies除了能实现用户身份的验证，还能实现很多功能，例如用户名保存，浏览习惯保存等
- 当用户登录的时候浏览器就会读取cookies，将上次登录的用户名填入；
- 读取浏览习惯，例如一些侧边栏是否打开，都会记录到cookies中，打开侧边栏和关闭侧边栏的时候，都会设置cookies的值的。
>document.cookie
<""
>document.cookie = "k1=1;path=/;"
<"k1=1;path=/;"
>document.cookie
<"k1=1"
可以看到的path是不显示的，只是一个条件，同理的还有domain和expires等
通过DOM设置cookies超时时间
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<script>
function setCookie(name,value,expires){
var current_date = new Date();
current_date.setSeconds(current_date.getSeconds() + 10);
document.cookie = "name=why;expires=" + current_date.toUTCString();
}
</script>
</body>
</html>
session功能
tornado并不提供session功能，不过还好提供session依赖的cookies功能。
session是存放在服务器端，可以在内存，缓存或者数据库中，不过一般
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
container = {}
class IndexHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
if self.get_argument('u',None) in ['why','wanghongyu']:
import hashlib
import time
obj = hashlib.md5()
obj.update(bytes(str(time.time())))
random_str = obj.hexdigest()
container[random_str] = {}
container[random_str]['k1'] = 123
container[random_str]['k2'] = self.get_argument('u',None)
container[random_str]['is_login'] = True
self.set_cookie('iiiii',random_str)
else:
self.write('请登录')
class ManagerHandler(tornado.web.RequestHandler):
def get(self):
random_str = self.get_cookie('iiiii')
current_user_info = container.get(random_str, None)
if not current_user_info:
self.redirect('index')
else:
if current_user_info.get('is_login', None):
temp = "%s %s" % (current_user_info.get('k1'), current_user_info.get('k2'))
self.write(temp)
else:
self.redirect('/index')
settings = {
'template_path': 'template', #配置模板路径
'static_path': 'static', #静态文件路径
}
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index", IndexHandler),
(r"/manager", ManagerHandler),
], **settings) #加载settings
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
访问http://127.0.0.1:8888/index?u=why
然后访问http://127.0.0.1:8888/manager可以看到k1和k2，这些k1和k2就是session的信息。
session进行封装
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
container = {}
class Session:
def __init__(self, handler):
self.handler = handler
def set_value(self, key, value):
random_str = self.__genarate_random_str()
container[random_str] = {}
container[random_str][key] = value
self.handler.set_cookie("iiiii", random_str)
def get_value(self, key):
random_str = self.handler.get_cookie("iiiii")
user_info_dict = container[random_str]
value = user_info_dict[key]
return value
def __genarate_random_str(self):
import hashlib
import time
obj = hashlib.md5()
obj.update(bytes(str(time.time())))
random_str = obj.hexdigest()
return random_str
class IndexHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
if self.get_argument('u',None) in ['why','wanghongyu']:
s = Session(self)
s.set_value('is_login',True)
else:
self.write('请登录')
class ManagerHandler(tornado.web.RequestHandler):
def get(self):
s = Session(self)
val = s.get_value('is_login')
if val:
self.write('成功')
else:
self.write('失败')
settings = {
'template_path': 'template', #配置模板路径
'static_path': 'static', #静态文件路径
}
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index", IndexHandler),
(r"/manager", ManagerHandler),
], **settings) #加载settings
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
解决session重复
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
container = {}
class Session:
def __init__(self, handler):
self.handler = handler
self.random_str = None
def set_value(self, key, value):
# 生成随机字符串做key
# 定义自己的数据
# 在客户端写入
if not self.random_str:
random_str = self.handler.get_cookie('iiiii')
# 判断session是否存在
# 不存在的话创建一个新的session
if not random_str:
random_str = self.__genarate_random_str()
container[random_str] = {}
# 存在的话
else:
# 判断session是否在服务器session库
if random_str in container.keys():
pass
# 不存在创建
else:
random_str = self.__genarate_random_str()
container[random_str] = {}
self.random_str = random_str
container[self.random_str][key] = value
self.handler.set_cookie("iiiii", self.random_str)
def get_value(self, key):
# 获取随机字符串
random_str = self.handler.get_cookie("iiiii")
# 如果随机字符串不存在
if not random_str:
return None
user_info_dict = container[random_str]
# 如果字典不存在
if not user_info_dict:
return None
value = user_info_dict.get(key, None)
return value
def __genarate_random_str(self):
# 生成随机字符串
import hashlib
import time
obj = hashlib.md5()
obj.update(bytes(str(time.time())))
random_str = obj.hexdigest()
return random_str
class IndexHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
if self.get_argument('u',None) in ['why','wanghongyu']:
s = Session(self)
s.set_value('is_login',True)
else:
self.write('请登录')
class ManagerHandler(tornado.web.RequestHandler):
def get(self):
s = Session(self)
val = s.get_value('is_login')
if val:
self.write('成功')
else:
self.write('失败')
settings = {
'template_path': 'template', #配置模板路径
'static_path': 'static', #静态文件路径
}
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index", IndexHandler),
(r"/manager", ManagerHandler),
], **settings) #加载settings
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
在其中不创建对象
在请求的时候，tornado内部创建对象原理是
obj = ManagerHandler()
func = getattr(obj, "get")
func()
这些都不是关键，这边主要需要创建对象的时候需要执行构造函数，在ManagerHandler中没有，需要去父类中寻找，可以
class RequestHandler(object):
def __init__(self, application, request, **kwargs):
省略部分
self.initialize(**kwargs)
可以看到这样一个函数
def initialize(self):
"""Hook for subclass initialization. Called for each request.
A dictionary passed as the third argument of a url spec will be
supplied as keyword arguments to initialize().
Example::
class ProfileHandler(RequestHandler):
def initialize(self, database):
self.database = database
def get(self, username):
...
app = Application([
(r'/user/(.*)', ProfileHandler, dict(database=database)),
])
"""
pass
这个函数是一个钩子函数，所以只需要在继承RequestHandler其之前，继承的类中有该方法就可以创建一个钩子函数，这是tornado特意留下来的。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
container = {}
class Session:
def __init__(self, handler):
self.handler = handler
self.random_str = None
def set_value(self, key, value):
# 生成随机字符串做key
# 定义自己的数据
# 在客户端写入
if not self.random_str:
random_str = self.handler.get_cookie('iiiii')
# 判断session是否存在
# 不存在的话创建一个新的session
if not random_str:
random_str = self.__genarate_random_str()
container[random_str] = {}
# 存在的话
else:
# 判断session是否在服务器session库
if random_str in container.keys():
pass
# 不存在创建
else:
random_str = self.__genarate_random_str()
container[random_str] = {}
self.random_str = random_str
container[self.random_str][key] = value
self.handler.set_cookie("iiiii", self.random_str)
def get_value(self, key):
# 获取随机字符串
random_str = self.handler.get_cookie("iiiii")
# 如果随机字符串不存在
if not random_str:
return None
user_info_dict = container[random_str]
# 如果字典不存在
if not user_info_dict:
return None
value = user_info_dict.get(key, None)
return value
def __genarate_random_str(self):
# 生成随机字符串
import hashlib
import time
obj = hashlib.md5()
obj.update(bytes(str(time.time())))
random_str = obj.hexdigest()
return random_str
class BaseHandler(tornado.web.RequestHandler):
def initialize(self):
self.session = Session(self)
class IndexHandler(BaseHandler):
def get(self, *args, **kwargs):
if self.get_argument('u',None) in ['why','wanghongyu']:
self.session.set_value('is_login',True)
self.session.set_value('u',self.get_argument('u',None))
else:
self.write('请登录')
class ManagerHandler(BaseHandler):
def get(self):
val = self.session.get_value('is_login')
if val:
self.write(self.session.get_value('u'))
else:
self.write('失败')
settings = {
'template_path': 'template', #配置模板路径
'static_path': 'static', #静态文件路径
}
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index", IndexHandler),
(r"/manager", ManagerHandler),
], **settings) #加载settings
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
如果想使用session['is_login']进行取值
可以将方法名改为__getitem__和__setitem__。
路由系统
支持基本路由，基础正则路由，二级域名路由
基本路由和基础正则路由
基础路由就是根据准确的url进行匹配，而基础正则路由就封装了re模块进行url匹配。
tornado代码
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
class IndexHandler(tornado.web.RequestHandler):
def get(self,nid, num):
print(nid,num)
self.render("index.html")
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index/(?P<num>\d*)/(?P<nid>\d*)", IndexHandler),
], ) #加载settings
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
通过re的正则进行分组，直接指定分组的名称，如果不指定名称是按照第一个参数，第二个参数的进行传递。
通过基础正则路由进行分页
第一步——调通路由与代码
tornado代码
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
list_info = []
class IndexHandler(tornado.web.RequestHandler): #继承tornado.web.RequestHandler
def get(self, page, *args, **kwargs): #直接访问的时候
self.render("index.html", list_info = list_info) #返回html
def post(self, *args, **kwargs):
username = self.get_argument('username')
passwd = self.get_argument('passwd')
temp = {'username': username, 'passwd': passwd}
list_info.append(temp)
self.redirect("/index/1")
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index/(?P<page>\d*)", IndexHandler), #当发送url请求的时候，对检查是否匹配，然后路由到对应的类的get，post等方法(路由映射)
],)
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
index.html代码
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form action="/index/1" method="post">
<input name="username" type="text">
<input name="passwd" type="text">
<input type="submit" value="提交">
</form>
<h1>show</h1>
<table border="1">
<thead>
<tr>
<th>用户名</th>
<th>密码</th>
</tr>
</thead>
<tbody>
{% for i in list_info %}
<tr>
<td>{{ i['username'] }}</td>
<td>{{ i['passwd'] }}</td>
</tr>
{% end %}
</tbody>
</table>
</body>
</html>
第二步——分页显示
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
list_info = []
class IndexHandler(tornado.web.RequestHandler): #继承tornado.web.RequestHandler
def get(self, page, *args, **kwargs): #直接访问的时候
try:
page = int(page)
except:
page = 1
if page < 1:
page = 1
start = (page - 1) * 5
end = start + 5
current_list = list_info[start: end]
self.render("index.html", list_info = current_list) #返回html
def post(self, *args, **kwargs):
username = self.get_argument('username')
passwd = self.get_argument('passwd')
temp = {'username': username, 'passwd': passwd}
list_info.append(temp)
self.redirect("/index/1")
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index/(?P<page>\d*)", IndexHandler), #当发送url请求的时候，对检查是否匹配，然后路由到对应的类的get，post等方法(路由映射)
],)
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
分页显示
第一页显示
第二页显示
不过有一个问题，我们提交后会一直显示第一页，理论上应该显示提交数据所在位置。
第三步
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
list_info = []
class IndexHandler(tornado.web.RequestHandler): #继承tornado.web.RequestHandler
def get(self, page, *args, **kwargs): #直接访问的时候
try:
page = int(page)
except:
page = 1
if page < 1:
page = 1
start = (page - 1) * 5
end = start + 5
current_list = list_info[start: end]
current_page = page
self.render("index.html", list_info = current_list, current_page = current_page) #返回html
def post(self, page, *args, **kwargs):
username = self.get_argument('username')
passwd = self.get_argument('passwd')
temp = {'username': username, 'passwd': passwd}
list_info.append(temp)
self.redirect("/index/" + page)
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index/(?P<page>\d*)", IndexHandler), #当发送url请求的时候，对检查是否匹配，然后路由到对应的类的get，post等方法(路由映射)
],)
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
html代码
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form action="/index/{{ current_page }}" method="post">
<input name="username" type="text">
<input name="passwd" type="text">
<input type="submit" value="提交">
</form>
<h1>show</h1>
<table border="1">
<thead>
<tr>
<th>用户名</th>
<th>密码</th>
</tr>
</thead>
<tbody>
{% for i in list_info %}
<tr>
<td>{{ i['username'] }}</td>
<td>{{ i['passwd'] }}</td>
</tr>
{% end %}
</tbody>
</table>
</body>
</html>
第四步——防止xss攻击
如果传入了一个JavaSrcipt代码
<script>alert(123);</script>
可以看到
那是tornado加载模板的时候进行对这些进行转义了，如果想取消转义。
<tbody>
{% for i in list_info %}
<tr>
<td>{% raw i['username'] %}</td>
<td>{{ i['passwd'] }}</td>
</tr>
{% end %}
</tbody>
可以看到
如果js获取用户名，密码的话就问题大了，这就是跨站脚本攻击XSS
第五步————通过分页的方式显示
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
list_info = []
class IndexHandler(tornado.web.RequestHandler): #继承tornado.web.RequestHandler
def get(self, page, *args, **kwargs): #直接访问的时候
try:
page = int(page)
except:
page = 1
if page < 1:
page = 1
start = (page - 1) * 5
end = start + 5
current_list = list_info[start: end]
current_page = page
all_pager,e = divmod(len(list_info), 5)
if e > 0:
all_pager += 1;
list_page = []
for p in range(all_pager):
if p + 1 == page:
temp = '<a class="active" href="/index/%s">%s</a>' % (p+1, p+1)
else:
temp = '<a href="/index/%s">%s</a>' % (p + 1, p + 1)
list_page.append(temp)
str_page = ''.join(list_page)
self.render("index.html", list_info = current_list, current_page = current_page, str_page=str_page) #返回html
def post(self, page, *args, **kwargs):
username = self.get_argument('username')
passwd = self.get_argument('passwd')
temp = {'username': username, 'passwd': passwd}
list_info.append(temp)
self.redirect("/index/" + page)
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index/(?P<page>\d*)", IndexHandler), #当发送url请求的时候，对检查是否匹配，然后路由到对应的类的get，post等方法(路由映射)
],)
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
html代码
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
<style>
.pager a{
display: inline-block;
padding: 5px;
margin: 3px;
background-color: blue;
}
.pager a.active{
background-color: red;
color: white;
}
</style>
</head>
<body>
<form action="/index/{{ current_page }}" method="post">
<input name="username" type="text">
<input name="passwd" type="text">
<input type="submit" value="提交">
</form>
<h1>show</h1>
<table border="1">
<thead>
<tr>
<th>用户名</th>
<th>密码</th>
</tr>
</thead>
<tbody>
{% for i in list_info %}
<tr>
<td>{{ i['username'] }}</td>
<td>{{ i['passwd'] }}</td>
</tr>
{% end %}
</tbody>
</table>
<div class="pager">
{% raw str_page %}
</div>
</body>
</html>
问题是页数多会把所有的页都显示出来
第六步——进行少数页进行展示
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
list_info = []
i = 0
while i < 300:
i += 1
temp = {'username': i, 'passwd': i}
list_info.append(temp)
class IndexHandler(tornado.web.RequestHandler): #继承tornado.web.RequestHandler
def get(self, page, *args, **kwargs): #直接访问的时候
try:
page = int(page)
except:
page = 1
if page < 1:
page = 1
start = (page - 1) * 5
end = start + 5
current_list = list_info[start: end]
current_page = page
all_pager,e = divmod(len(list_info), 5)
if e > 0:
all_pager += 1;
'''
总页数 <= 11
总页数 > 11
当前页 <= 6
当前页 > 6
当前页 + 5 > 总页数
'''
if all_pager <= 11:
s = 1
t = all_pager
else:
if page <= 6:
s = 1
t = 11
else:
if (page + 5) > all_pager:
s = page - 5
t = all_pager
else:
s = page - 5
t = page + 5
list_page = []
for p in range(s, t + 1):
if p == page:
temp = '<a class="active" href="/index/%s">%s</a>' % (p, p)
else:
temp = '<a href="/index/%s">%s</a>' % (p, p)
list_page.append(temp)
str_page = ''.join(list_page)
self.render("index.html", list_info = current_list, current_page = current_page, str_page=str_page) #返回html
def post(self, page, *args, **kwargs):
username = self.get_argument('username')
passwd = self.get_argument('passwd')
temp = {'username': username, 'passwd': passwd}
list_info.append(temp)
self.redirect("/index/" + page)
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index/(?P<page>\d*)", IndexHandler), #当发送url请求的时候，对检查是否匹配，然后路由到对应的类的get，post等方法(路由映射)
],)
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
第七步
添加首尾页上下页
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
list_info = []
i = 0
while i < 300:
i += 1
temp = {'username': i, 'passwd': i}
list_info.append(temp)
class IndexHandler(tornado.web.RequestHandler): #继承tornado.web.RequestHandler
def get(self, page, *args, **kwargs): #直接访问的时候
try:
page = int(page)
except:
page = 1
if page < 1:
page = 1
start = (page - 1) * 5
end = start + 5
current_list = list_info[start: end]
current_page = page
all_pager,e = divmod(len(list_info), 5)
if e > 0:
all_pager += 1;
'''
总页数 <= 11
总页数 > 11
当前页 <= 6
当前页 > 6
当前页 + 5 > 总页数
'''
if all_pager <= 11:
s = 1
t = all_pager
else:
if page <= 6:
s = 1
t = 11
else:
if (page + 5) > all_pager:
s = page - 5
t = all_pager
else:
s = page - 5
t = page + 5
list_page = []
# 首页
first_page = '<a href="/index/%s">%s</a>' % ('1', '首页')
list_page.append(first_page)
# 上一页
if page == 1:
prev_page = '<a href="javascript:void(0);">上一页</a>'
else:
prev_page = '<a href="/index/%s">上一页</a>' % (page - 1)
list_page.append(prev_page)
# 中间页
for p in range(s, t + 1):
if p == page:
temp = '<a class="active" href="/index/%s">%s</a>' % (p, p)
else:
temp = '<a href="/index/%s">%s</a>' % (p, p)
list_page.append(temp)
# 下一页
if page == all_pager:
next_page = '<a href="javascript:void(0);">下一页</a>'
else:
next_page = '<a href="/index/%s">下一页</a>' % (page + 1)
list_page.append(next_page)
# 尾页
last_page = '<a href="/index/%s">%s</a>' % (all_pager, '尾页')
list_page.append(last_page)
str_page = ''.join(list_page)
self.render("index.html", list_info = current_list, current_page = current_page, str_page=str_page) #返回html
def post(self, page, *args, **kwargs):
username = self.get_argument('username')
passwd = self.get_argument('passwd')
temp = {'username': username, 'passwd': passwd}
list_info.append(temp)
self.redirect("/index/" + page)
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index/(?P<page>\d*)", IndexHandler), #当发送url请求的时候，对检查是否匹配，然后路由到对应的类的get，post等方法(路由映射)
],)
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
第八步——封装成类
封装成类
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
list_info = []
i = 0
while i < 300:
i += 1
temp = {'username': i, 'passwd': i}
list_info.append(temp)
# 封装成类
class Pagination:
def __init__(self, page, all_paper):
try:
page = int(page)
except:
page = 1
if page < 1:
page = 1
self.page = page
self.all_pager = all_paper
@property
def start(self):
return (self.page - 1) * 5
@property
def end(self):
return self.page * 5
def page_str(self):
list_page = []
if self.all_pager <= 11:
s = 1
t = self.all_pager
else:
if self.page <= 6:
s = 1
t = 11
else:
if (self.page + 5) > self.all_pager:
s = self.page - 5
t = self.all_pager
else:
s = self.page - 5
t = self.page + 5
# 首页
first_page = '<a href="/index/%s">%s</a>' % ('1', '首页')
list_page.append(first_page)
# 上一页
if self.page == 1:
prev_page = '<a href="javascript:void(0);">上一页</a>'
else:
prev_page = '<a href="/index/%s">上一页</a>' % (self.page - 1)
list_page.append(prev_page)
# 中间页
for p in range(s, t + 1):
if p == self.page:
temp = '<a class="active" href="/index/%s">%s</a>' % (p, p)
else:
temp = '<a href="/index/%s">%s</a>' % (p, p)
list_page.append(temp)
# 下一页
if self.page == self.all_pager:
next_page = '<a href="javascript:void(0);">下一页</a>'
else:
next_page = '<a href="/index/%s">下一页</a>' % (self.page + 1)
list_page.append(next_page)
# 尾页
last_page = '<a href="/index/%s">%s</a>' % (self.all_pager, '尾页')
list_page.append(last_page)
# 页面跳转
jump = """<input type='text' /><a onclick="Jump(this);">Go</a>"""
script = """
<script>
function Jump(ths) {
var val = ths.previousElementSibling.value;
if(val.trim().length>0){
location.href = '/index/' + val;
}
}
</script>
"""
# trim去空格
list_page.append(jump)
list_page.append(script)
str_page = ''.join(list_page)
return str_page
class IndexHandler(tornado.web.RequestHandler): #继承tornado.web.RequestHandler
def get(self, page, *args, **kwargs): #直接访问的时候
all_pager,e = divmod(len(list_info), 5)
if e > 0:
all_pager += 1
page_obj = Pagination(page,all_pager)
current_list = list_info[page_obj.start: page_obj.end]
str_page = page_obj.page_str()
self.render("index.html", list_info = current_list, current_page = page, str_page=str_page) #返回html
def post(self, page, *args, **kwargs):
username = self.get_argument('username')
passwd = self.get_argument('passwd')
temp = {'username': username, 'passwd': passwd}
list_info.append(temp)
self.redirect("/index/" + page)
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index/(?P<page>\d*)", IndexHandler), #当发送url请求的时候，对检查是否匹配，然后路由到对应的类的get，post等方法(路由映射)
],)
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
二级域名路由
这个是tornado框架独有的路由系统
class IndexHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
self.render("index.html")
class WWWHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
self.render("www.html")
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index/(?P<page>\d*)", IndexHandler), #当发送url请求的时候，对检查是否匹配，然后路由到对应的类的get，post等方法(路由映射)
],)
application.add_handlers('www.whysdomain.com',[
(r"/index/(?P<page>\d*)", WWWHandler),
],)
验证码
依赖模块
[C:\Python27\Scripts]$ pip.exe install pillow
Collecting pillow
Downloading Pillow-4.2.1-cp27-cp27m-win_amd64.whl (1.4MB)
Collecting olefile (from pillow)
Downloading olefile-0.44.zip (74kB)
Installing collected packages: olefile, pillow
Running setup.py install for olefile: started
Running setup.py install for olefile: finished with status 'done'
Successfully installed olefile-0.44 pillow-4.2.1
tornado代码
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop #导入模块
import tornado.web
import io
import check_code
container = {}
class Session:
def __init__(self, handler):
self.handler = handler
self.random_str = None
def set_value(self, key, value):
# 生成随机字符串做key
# 定义自己的数据
# 在客户端写入
if not self.random_str:
random_str = self.handler.get_cookie('iiiii')
# 判断session是否存在
# 不存在的话创建一个新的session
if not random_str:
random_str = self.__genarate_random_str()
container[random_str] = {}
# 存在的话
else:
# 判断session是否在服务器session库
if random_str in container.keys():
pass
# 不存在创建
else:
random_str = self.__genarate_random_str()
container[random_str] = {}
self.random_str = random_str
container[self.random_str][key] = value
self.handler.set_cookie("iiiii", self.random_str)
def get_value(self, key):
# 获取随机字符串
random_str = self.handler.get_cookie("iiiii")
# 如果随机字符串不存在
if not random_str:
return None
user_info_dict = container[random_str]
# 如果字典不存在
if not user_info_dict:
return None
value = user_info_dict.get(key, None)
return value
def __genarate_random_str(self):
# 生成随机字符串
import hashlib
import time
obj = hashlib.md5()
obj.update(bytes(str(time.time())))
random_str = obj.hexdigest()
return random_str
class BaseHandler(tornado.web.RequestHandler):
def initialize(self):
self.session = Session(self)
class IndexHandler(BaseHandler):
def get(self, *args, **kwargs):
if self.get_argument('u',None) in ['why','wanghongyu']:
self.session.set_value('is_login',True)
self.session.set_value('u',self.get_argument('u',None))
else:
self.write('请登录')
class ManagerHandler(BaseHandler):
def get(self):
val = self.session.get_value('is_login')
if val:
self.write(self.session.get_value('u'))
else:
self.write('失败')
class CheckCodeHandler(BaseHandler):
def get(self):
# 生成图片并返回
mstream = io.BytesIO()
# 创建图片，写入验证码
img, code = check_code.create_validate_code()
# 将图片写入mstream对象
img.save(mstream, "GIF")
self.write(mstream.getvalue())
self.session.set_value("CheckCode", code)
class LoginHandler(BaseHandler):
def get(self, *args, **kwargs):
self.render('index.html',status='')
def post(self, *args, **kwargs):
user = self.get_argument('user',None)
pwd = self.get_argument('pwd', None)
code = self.get_argument('code',None)
check_code = self.session.get_value('CheckCode')
if code.upper() == check_code.upper():
self.write('验证码正确')
else:
self.write('验证码错误')
settings = {
'template_path': 'template', #配置模板路径
'static_path': 'static', #静态文件路径
}
application = tornado.web.Application([ #创建tornado.web.Application对象
(r"/index", IndexHandler),
(r"/manager", ManagerHandler),
(r"/login", LoginHandler),
(r"/check_code", CheckCodeHandler),
], **settings) #加载settings
if __name__ == "__main__":
application.listen(8888) #创建一个socket
tornado.ioloop.IOLoop.instance().start() #使用epoll，IO多路复用循环
html代码
<!DOCTYPE html>
<html>
<head lang="en">
<meta charset="UTF-8">
<title></title>
</head>
<body>
<h1>hello</h1>
<form action="/index" method="post" enctype="multipart/form-data">
<p><input name="user" type="text" placeholder="用户名" /></p>
<p><input name="pwd" type="password" placeholder="密码" /></p>
<p>
<input name='code' type="text" placeholder="验证码" />
<img src="/check_code" onclick='ChangeCode();' id='imgCode'>
</p>
<input type="submit" />
</form>
<script type="text/javascript">
function ChangeCode() {
var code = document.getElementById('imgCode');
code.src += '?';
}
</script>
</body>
</html>
目录结构
下载链接 check_code.py Monaco.ttf
csrf
csrf限制post请求
get请求的时候，csrf会给一个随机字符串，而post请求的时候带着这个随机字符串，可以判断这个csrf是否正确，相当于token，所以直接发post请求就不行
class CsrfHandler(BaseHandler):
def get(self, *args, **kwargs):
self.render('index.html')
def post(self, *args, **kwargs):
self.write('csrf.post')
进行访问的时候没有问题
settings = {
'template_path': 'template', #配置模板路径
'static_path': 'static', #静态文件路径
'xsrf_cookies': True
}
进行post请求的时候会出现403: Forbidden
{{ xsrf_form_html() }}
会出现<input type="hidden" name="_xsrf" value="2|b31776ca|7b08a3a6f7c20f5d3845982f06422a4c|1504381671"/>
{% raw xsrf_form_html() %}
对于ajax请求
function getCookie(name) {
var r = document.cookie.match("\\b" + name + "=([^;]*)\\b");
return r ? r[1] : undefined;
}
jQuery.postJSON = function(url, args, callback) {
args._xsrf = getCookie("_xsrf");
$.ajax({url: url, data: $.param(args), dataType: "text", type: "POST",
success: function(response) {
callback(eval("(" + response + ")"));
}});
};
原理是从cookies中获取_xsrf
模板引擎
Tornao中的模板语言和django中类似，模板引擎将模板文件载入内存，然后将数据嵌入其中，最终获取到一个完整的字符串，再将字符串返回给请求者。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.ioloop
import tornado.web
class IndexHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
self.render("2.html")
application = tornado.web.Application([
(r"/", IndexHandler),
],)
if __name__ == "__main__":
application.listen(8888)
tornado.ioloop.IOLoop.instance().start()
1.html，用作模板
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<p>1</p>
{% block first %}
{% end %}
<p>1</p>
{% block second %}
{% end %}
<p>1</p>
</body>
</html>
2.html，引入模板
{% extends "1.html" %}
{% block first %}
<p>2</p>
{% include "3.html" %}
{% end %}
{% block second %}
<p>2</p>
{% include "3.html" %}
{% end %}
3.html，被导入模板
<p>3</p>
示例效果
也支持if判断，for循环，更多可以参考tornado官方
