如果你的 python 环境还没有安装 tornado,请直接使用 pip 安装:
pip install tornado
demo.py
# -*- coding: utf-8 -*-
import tornado.ioloop
import tornado.web
class HomeHandler(tornado.web.RequestHandler):
def get(self): # 响应以get方式发送的请求
self.write("hello, world") # 向请求者(浏览器)应答hello, world
app = tornado.web.Application([ (r"/", HomeHandler), ]) # URL映射
app.listen(80) # 绑定侦听端口
tornado.ioloop.IOLoop.instance().start() # 启动服务
如果多少了解一点 http 协议,知道 get / post 方法,相信你一定能够读懂。也许你的项目规划了很多的url,也许你的服务需要监听非80端口,没有关系,在这个代码上扩展就行。仅仅六行!!!请让我们向犀利的、简洁的、无所不能的 python 致敬!
重点:tornado.web.RequestHandler.write() 不只可以接受字符串参数,还可以接受列表或字典参数——如果应答类型为json时,这个重载特性非常高效
假定我们有这样一个 web 服务需求:
以上面的代码为基础,我们首先要做的工作是 URL 和 对应的处理类之间的关联。这件工作实际上是非常轻松愉快的:
app = tornado.web.Application([
(r"/", HomeHandler),
(r"/login", LoginHandler),
(r"/me", MeHandler)
])
接下来,我们要实现 HomeHandler、LoginHandler 和 MeHandler 这三个类了。通常,我们习惯把这些和URL 对应的处理类,保存为一个独立的文件,比如文件名为 handlers.py,然后在服务器脚本 demo.py 中导入它们。
handlers.py
# -*- coding: utf-8 -*-
import tornado.web
class HomeHandler(tornado.web.RequestHandler):
"""响应主页请求"""
def get(self): # 以get方式请求
self.write("""<!DOCTYPE html><html><body><a href="login">点此登录</a></body></html>""")
class LoginHandler(tornado.web.RequestHandler):
"""响应登录页请求"""
def get(self): # 以get方式请求
self.write(
"""
<!DOCTYPE html><html><body><form method="POST" action="/login">
账号:<input type="text" name="account" value="" /><br />
密码:<input type="password" name="passwd" value="" />
<input type="submit" value="确定" />
</form></body></html>
"""
)
def post(self): # 以post方式请求(本例为提交表单)
account = self.get_argument('account', None)
passwd = self.get_argument('passwd', None)
if account == 'xufive' and passwd == 'dgdgwstd':
self.redirect('/me?name=%s'%account)
else:
self.redirect('/')
class MeHandler(tornado.web.RequestHandler):
"""响应个人信息页请求"""
def get(self): # 以get方式请求
name = self.get_argument('name', None)
if name:
self.write(
"""
<!DOCTYPE html><html><head><meta charset="UTF-8" /></head>
<body>欢迎你来到这里,%s</body></html>
"""%name
)
else:
self.redirect('/')
相应地,服务脚本变成了这样:
demo.py
# -*- coding: utf-8 -*-
import os
import tornado.ioloop
import tornado.web
from tornado.options import parse_command_line
from handlers import *
parse_command_line()
app = tornado.web.Application(
handlers=[
(r"/", HomeHandler),
(r"/login", LoginHandler),
(r"/me", MeHandler)
],
template_path = os.path.join(os.path.dirname(__file__), 'templates')
)
app.listen(80) # 绑定侦听端口
tornado.ioloop.IOLoop.instance().start() # 启动服务
重点:tornado.web.RequestHandler.get_argument() 可以读取通过表单和QueryString传递的参数
读到这里,你一定会觉得奇怪:为什么服务端程序里面混杂了一大堆的 html 代码?Don’t worry,以上的代码仅仅是帮助你建立基本概念的,实际上,tornado 是为数不多的支持模板技术很到位的框架之一,其模板技术不仅支持继承,支持子模版。让我们一步一步讨论如何使用模板。
第1步:模板保存在哪儿?
第2步:怎样写模板?
base.html
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<!-- 此处省略各种样式表文件 -->
</head>
<body>
{% block block_body %}{% end %}
</body>
<html>
基类模板 base.html 定义了一个 block_body 容器,如果有必要,我们在基类模板的任意位置定义更多的容器。假定我们需要一个个人信息页模板,可以直接继承 base.html,然后只填写 block_body 这一部分就行了。
me.html
{% extends "base.html" %}
{% block block_body %}
<h1>欢迎你来到这里,{{name}}</h1>
{% end %}
个人信息页模板引中,我们使用 {{}} 引用了一个变量 name
第3步:如何使用模板?
class MeHandler(tornado.web.RequestHandler):
"""响应个人信息页请求"""
def get(self): # 以get方式请求
name = self.get_argument('name', None)
if name:
self.render('me.html', name=name )
else:
self.redirect('/')
常用的模板语法汇总如下:
tornado.web.RequestHandler 的 cookie 操作非常灵活,下面的 handler 展示了 cookie 的基本读写方法:
class CookieHandler(tornado.web.RequestHandler):
def get(self):
visit_num = self.get_cookie('visit_num')
if not visit_num:
visit_num = '0'
visit_num = str(int(visit_num)+1)
#self.set_cookie('visit_num', visit_num, expires=None) # 内存cookie
self.set_cookie('visit_num', visit_num, expires=time.time()+1000) # 持久化的cookie
self.write("这是您第%s次访问本页面"%visit_num)
如果我们要使用持久化的 Cookie(硬盘 Cookie),为了防止被破解,一般是要加密的,那么,在 tornado.web.Application 中需要设置 cookie_secret 项(加密因子)。
定义tornado.web.Application,这是常用的一个模式:
class Application(tornado.web.Application):
def __init__(self):
handlers = [
(r"/", WelcomeHandler), # 欢迎信息
(r"/server_time",ServerTimeHandler) # 显示服务器时间
]
settings = dict(
title = u"网站名称",
template_path = os.path.join(os.path.dirname(__file__), 'templates'),
static_path = os.path.join(os.path.dirname(__file__), 'static'),
cookie_secret = 'rewqr4gfd654fdsg@$%34dfs',
session_expiry = 0,
login_url = "/",
debug = 1
)
tornado.web.Application.__init__(self, handlers, **settings)
这段代码在 py2 环境 tornado 2.x 以后的各种版本下都可运行。
# -*- coding: utf-8 -*-
# Tornado Session For py2
#---------------------------------------
# version 3.2 / Update: 2018-12-24
#
# 1. 重写get_current_user方法,读取session_id
# version 3.1 / Update: 2018-12-21
#
# 1. 以文件作为session的后端
# 2. 私有方法命名规范化
# version 3.0 / Update: 2018-12-19
#
# 1. 以cookie形式实现session
# 2. 用装饰器引入db
# 3. 类方法命名规范化(不兼容之前的版本)
import os, time
import hashlib
import json
import tornado.web
class SessionRH(tornado.web.RequestHandler):
"""为RequestHandler增加session机制"""
def initialize(self):
"""初始化(构造函数完成之后)"""
self._init_session()
def _init_session(self):
"""session初始化"""
self.session = dict()
self.session_id = self.get_secure_cookie('session_id')
if self.session_id:
session_file = os.path.join(self.application.settings["session_path"], self.session_id)
if os.path.isfile(session_file):
with open(session_file, 'r') as fp:
self.session.update(json.loads(fp.read().strip()))
if not self.session:
self.session_id = None
self.clear_cookie('session_id')
def _save_session(self):
"""保存session"""
try:
expiry = self.application.settings["session_expiry"]
except:
expiry = 0
if not expiry:
expiry = None
self.set_secure_cookie('session_id', self.session_id, expires_days=expiry)
with open(os.path.join(self.application.settings["session_path"], self.session_id), 'w') as fp:
fp.write(json.dumps(self.session))
def create_session(self, **kwargs):
"""生成唯一的session_id及其它需要保存的数据项"""
rand = os.urandom(16)
now = time.time()
self.session_id = hashlib.sha1("%s%s%s" %(rand, now, self.request.remote_ip)).hexdigest()
self.session.update(kwargs)
self.session.update({'last_active':str(now)})
self._save_session()
def destroy_session(self):
"""注销session"""
session_file = os.path.join(self.application.settings["session_path"], self.session_id)
if os.path.isfile(session_file):
os.remove(session_file)
self.session_id = None
self.session = dict()
self.clear_cookie('session_id')
def append_session_item(self, **kwargs):
"""追加session项"""
self.session.update(kwargs)
self._save_session()
def delete_session_item(self, name):
"""删除session项"""
if name in self.session:
self.session.pop(name)
self._save_session()
def get_current_user(self):
"""重写get_current_user方法,读取session_id"""
return self.get_secure_cookie("session_id")
@property
def db(self):
"""用装饰器引入db(application.db)"""
return self.application.db
if __name__ == '__main__':
pass