本文共 15011 字，大约阅读时间需要 50 分钟。
tornado.httputil.HTTPServerRequest
from tornado import ioloop from tornado import web from tornado import autoreload from tornado.options import define,options,parse_command_line settings = { 'debug' : True, } define("port", default=8888, type=int,help="设置监听端口号，默认为8888") class Home(web.RequestHandler): def get(self): print(self.request) # 请求处理对象 # HTTPServerRequest(protocol='http', host='127.0.0.1:8888', method='GET', uri='/?name=xiaoming', version='HTTP/1.1', remote_ip='127.0.0.1') # print(self.request.protocol) # 协议 # print(self.request.method) # Http请求方法 # print(self.request.uri) # uri地址 # print(self.request.full_url()) # 完整url地址 # print(self.request.version) # HTTP协议版本 # print(self.request.headers) # 请求头 HTTPHeaders # print(self.request.body) # 请求体[原始数据] # print(self.request.host) # 地址端口 # print(self.request.files) # 上传文件 # print(self.request.cookies) # cookie信息 # print(self.request.remote_ip) # 客户端IP地址 print(self.request.query_arguments) # 地址参数列表 print(self.request.body_arguments) # 请求体参数列表 print(self.request.request_time()) # 请求处理时间 self.write("hello world") # 设置路由列表 urls = [ (r"/", Home), ] if __name__ == "__main__": # 创建应用实例对象 parse_command_line() app = web.Application(urls,**settings) # 设置监听的端口和地址 app.listen(options.port) # ioloop，全局的tornado事件循环，是服务器的引擎核心，start表示创建IO事件循环 ioloop.IOLoop.current().start()## 接收查询字符串from tornado import ioloopfrom tornado import webfrom tornado import autoreloadfrom tornado.options import define,options,parse_command_linesettings = { 'debug' : True,}define("port", default=8888, type=int,help="设置监听端口号，默认为8888")class Home(web.RequestHandler): def get(self): # print(self.request.arguments["name"][0].decode()) # name = self.get_argument("name") # self.get_query_argument("name") # print(name) # xiaoming names = self.get_arguments("name") # # self.get_query_arguments("name") print(names) # ['xiaoming', '123'] # self.write 响应数据 # self.write("hello!") self.write("hello world")# 设置路由列表urls = [ (r"/", Home),]if __name__ == "__main__": # 创建应用实例对象 parse_command_line() app = web.Application(urls,**settings) # 设置监听的端口和地址 app.listen(options.port) # ioloop，全局的tornado事件循环，是服务器的引擎核心，start表示创建IO事件循环 ioloop.IOLoop.current().start() from tornado import ioloopfrom tornado import webfrom tornado import autoreloadfrom tornado.options import define,options,parse_command_linesettings = { 'debug' : True,}define("port", default=8888, type=int,help="设置监听端口号，默认为8888")class Home(web.RequestHandler): def get(self): # print(self.request.arguments["name"][0].decode()) # name = self.get_argument("name") # self.get_query_argument("name") # print(name) # xiaoming names = self.get_arguments("name") # # self.get_query_arguments("name") print(names) # ['xiaoming', '123'] self.write("hello!get") def post(self): print(self.request.arguments) # {'name': [b'xiaoming', b'xiaohong']} print(self.request.body_arguments) # {'name': [b'xiaohong']} print(self.get_argument("name")) # xiaohong print(self.get_body_argument("name")) # xiaohong print(self.get_arguments("name")) # ['xiaoming', 'xiaohong'] print(self.get_body_arguments("name")) # ['xiaohong'] self.write("hello!post") # 设置路由列表urls = [ (r"/", Home),]if __name__ == "__main__": # 创建应用实例对象 parse_command_line() app = web.Application(urls,**settings) # 设置监听的端口和地址 app.listen(options.port) # ioloop，全局的tornado事件循环，是服务器的引擎核心，start表示创建IO事件循环 ioloop.IOLoop.current().start() from tornado import ioloopfrom tornado import webfrom tornado import autoreloadfrom tornado.options import define,options,parse_command_linesettings = { 'debug' : True,}define("port", default=8888, type=int,help="设置监听端口号，默认为8888")class Home(web.RequestHandler): def get(self,name): print(name) self.write("home!get")class Index(web.RequestHandler): def get(self,name): print(name) self.write("index!get")# 路由列表urls = [ (r"/home/(.+)", Home), # 不绑定传参 (r"/index/(?P .+)", Index), # 绑定传参]if __name__ == "__main__": # 创建应用实例对象 parse_command_line() app = web.Application(urls,**settings) # 设置监听的端口和地址 app.listen(options.port) # ioloop，全局的tornado事件循环，是服务器的引擎核心，start表示创建IO事件循环 ioloop.IOLoop.current().start() from tornado import ioloopfrom tornado import webfrom tornado import autoreloadfrom tornado.options import define, options, parse_command_linesettings = { 'debug': True,}define("port", default=8888, type=int, help="设置监听端口号，默认为8888")from datetime import datetimeclass Home(web.RequestHandler): def set_default_headers(self): self.set_header("time", int(datetime.now().timestamp())) def get(self): # self.write("hello
") # 响应html文本信息 self.write({"message":"hello get"}) # 响应json数据 self.set_header("Content-Type","text/json; charset=gbk") self.add_header("Company","OldboyEdu") # 自定义响应头 self.set_cookie("name","xiaohui") # 设置cookie def post(self): self.write({"message": "hello post"}) # 响应json数据 def put(self): self.clear_header("time") # self.set_status(404,"Not Found") # self.send_error(500,reason="服务器炸了！") self.send_error(404, msg="服务器炸了！", info="快报警") def write_error(self, status_code, **kwargs): self.write("完蛋啦...
") self.write("错误信息:%s
" % kwargs["msg"]) self.write("错误描述:%s
" % kwargs["info"]) def patch(self): # 页面跳转 self.redirect("http://www.baidu.com")# 设置路由列表urls = [ (r"/", Home),]if __name__ == "__main__": # 创建应用实例对象 parse_command_line() app = web.Application(urls, **settings) # 设置监听的端口和地址 app.listen(options.port) # ioloop，全局的tornado事件循环，是服务器的引擎核心，start表示创建IO事件循环 ioloop.IOLoop.current().start() 设置cookie self.set_cookie(name, value)
获取cookie self.get_cookie(name)
from tornado import webfrom tornado import ioloopsettings = { "debug": True,}from datetime import datetimeclass Home(web.RequestHandler): def get(self): # 设置cookie self.set_cookie("uname","xiaoming",expires=int(datetime.now().timestamp())+10) self.write("set cookie")class Index(web.RequestHandler): def get(self): # 获取cookie uname = self.get_cookie("uname","") self.write("uname=%s" % uname)urls = [ (r"/cookie/set", Home), (r"/cookie/get", Index),]if __name__ == '__main__': app = web.Application(urls,**settings) app.listen(port=8888) ioloop.IOLoop.current().start() 设置cookie self.set_secure_cookie(name,value)
获取cookie self.get_secure_cookie(name)
删除cookie self.clear_cookie(name)
清空cookie self.clear_all_cookie()
from tornado import webfrom tornado import ioloopsettings = { "debug": True, # import base64, uuid # base64.b64encode(uuid.uuid4().bytes + uuid.uuid4().bytes) "cookie_secret": "WO+JNAJ3QZyOe4SMVXZpXAt3uG9hoU0UokoCBeYn1Y4="}from datetime import datetimeclass Home(web.RequestHandler): def get(self): self.set_secure_cookie("name","xiaoming",expires=int(datetime.now().timestamp())+30) self.set_secure_cookie("age","16",expires=int(datetime.now().timestamp())+30) self.write("set cookie")class Index(web.RequestHandler): def get(self): # 获取cookie[加密] age = self.get_secure_cookie("age") name = self.get_secure_cookie("name") if age is not None: age = age.decode() if name is not None: name = name.decode() self.write("age=%s,name=%s" % (age,name))class Page(web.RequestHandler): def get(self): """删除cookie""" # self.clear_cookie("age") # 删除指定名称的cookie[不管是否有加密] self.clear_all_cookies() # 删除所有cookie[慎用] self.write("del cookie")urls = [ (r"/cookie/set", Home), (r"/cookie/get", Index), (r"/cookie/del", Page),]if __name__ == '__main__': app = web.Application(urls,**settings) app.listen(port=8888) ioloop.IOLoop.current().start() from tornado import webfrom tornado import ioloopimport ossettings = { 'debug': True, # 静态文件保存路径 "static_path": os.path.join(os.path.dirname(__file__), 'static'), # 静态文件url地址前缀 "static_url_prefix":"/static/", # 必须前后有斜杠 # 提供静态文件访问支持的视图类 "static_handler_class": web.StaticFileHandler,}class Home(web.RequestHandler): def get(self): # 项目中使用join拼凑路径时，必须注意第二个参数，必能以斜杠开头，会出现路径穿越(路径穿透)问题 path = os.path.join(os.path.dirname(__file__),"/static") self.write("path=%s" % path)urls = [ (r"/", Home), # 上面settings中关于静态文件的配置，主要是提供给Application应用对象进行初始化生成下面路由时候使用到的。 # (r"/static/(.*)", web.StaticFileHandler, {"path": os.path.join(os.path.dirname(__file__), 'static')}),]if __name__ == '__main__': app = web.Application(urls,**settings) app.listen(port=8888) ioloop.IOLoop.current().start() 加载template文件
from tornado import webfrom tornado import ioloopimport ossettings = { 'debug': True, "template_path": os.path.join(os.path.dirname(__file__), 'templates'),}class Home(web.RequestHandler): def get(self): self.render("index.html",data={"message":"hello world"})urls = [ (r"/", Home),]if __name__ == '__main__': app = web.Application(urls,**settings) app.listen(port=8888) ioloop.IOLoop.current().start() Title {
{data["message"]}}
from tornado import webfrom tornado import ioloopsettings = { 'debug': True,}class Home(web.RequestHandler): def initialize(self,company)-> str: # initialize 初始化方法[钩子方法] self.company = company def get(self): print(self.company) print("uri路径：%s" % self.reverse_url("home") ) # 对路由别名进行 反解析 self.write("hello,get") def post(self): print(self.company)from tornado.web import urlurls = [ # (r"/", Home), # 这个格式的路由其实是简写模式， 在tornaodo.web中内部中最终由 _ApplicationRouter 的 Rule来进行封装和匹配路由和视图的关系 # url(pattern=路由uri地址, handler=视图类,kwargs=提供给视图类的公共参数,name="路由别名，用于反解析"), url(pattern=r"/abc", handler=Home,kwargs={"company":"OldBoyEdu"},name="home"),]if __name__ == '__main__': app = web.Application(urls,**settings) app.listen(port=8888) ioloop.IOLoop.current().start() 在tornado提供的视图类中，我们除了可以编写客户端http请求对应名称的视图方法和初始化方法initialize以外，还提供了一个预处理方法prepare和on_finish，prepare方法会在http请求方法执行之前先执行,on_finish会在http响应完成时进行。
from tornado import ioloopfrom tornado import webfrom tornado.httpserver import HTTPServerfrom tornado.options import define, options, parse_command_linefrom tornado.web import urlsettings = { 'debug': True,}define("port", default=8888, type=int, help="设置监听端口号，默认为8888")class Home(web.RequestHandler): def initialize(self): print("initialize执行了") def prepare(self): print("prepare执行了") def set_default_headers(self): print("set_default_headers执行了") def get(self): self.write("hello,get") print("视图http方法执行了") # self.send_error(200,msg="注意：丢炸弹了") # 此处抛出错误 def write_error(self, status_code, **info): print("write_error执行了,msg=%s" % info["msg"]) def on_finish(self): print("on_finish执行了")# 设置路由列表urls = [ (r"/", Home),]if __name__ == "__main__": # 创建应用实例对象 parse_command_line() app = web.Application(urls, **settings) server = HTTPServer(app) # 设置监听的端口和地址 server.listen(options.port) server.start(1) ioloop.IOLoop.current().start() 没有异常
有异常
事实上，在tornado提供的视图操作中，视图中提供了一个 _write_buffer列表用于暂时缓存提供给客户端的数据, 这个 _write_buffer就是输出缓冲区
self.write()本质上来说是将chunk数据块写到输出缓冲区中。所以才出现在视图中多次调用self.write()输出数据的情况,因为self.write根本没有输出数据,而是把数据写入到了输出缓冲区里面. 如果没有其他操作干预的情况下，则视图方法处理完成以后，会讲输出缓冲区中所有的数据冲刷出来响应给客户端。
除了self.write()方法以外，tornado还提供了2个方法用于在视图中冲刷缓存数据到客户端的。
self.flush() 立刻把数据从输出缓冲区冲刷出去。
self.finish()立刻把数据从输出缓冲区冲刷出去。但是与self.flush()不同的是, self.finish()执行了以后, 后面的所有输出调用都不在支持.也就不能返回给客户端。
from tornado import ioloopfrom tornado import webfrom tornado.httpserver import HTTPServerfrom tornado.options import define, options, parse_command_linesettings = { 'debug': True,}define("port", default=8888, type=int, help="设置监听端口号，默认为8888")class Home(web.RequestHandler): def get(self): import time self.write("hello,get1") self.flush() time.sleep(3) self.write("hello,get2") self.flush() time.sleep(3) self.write("hello,get3") self.flush() self.finish("这里一般不写任何内容，表示视图处理结束") self.write("hello,get4")# 设置路由列表urls = [ (r"/", Home),]if __name__ == "__main__": # 创建应用实例对象 parse_command_line() app = web.Application(urls, **settings) server = HTTPServer(app) # 设置监听的端口和地址 server.listen(options.port) server.start(1) ioloop.IOLoop.current().start() tornado提供了装饰器tornado.web.authenticated与视图内置方法get_current_user允许我们轻松的实现用户认证功能。
装饰器authenticated依赖于请求处理类中的self.current_user属性来进行判断用户是否通过认证，如果self.current_user值为假（None、False、0、""等），任何GET或HEAD请求都将把访客重定向到settings配置中login_url设置的URL，而非法用户的POST请求将返回HTTPError(403)异常, Forbidden。
from tornado import webfrom tornado import ioloopsettings = { 'debug': True, # 登录页面的url地址 "login_url": r"/login"}from tornado.web import authenticatedclass HttpRequest(web.RequestHandler): def get_current_user(self): username = self.get_argument("username", "") password = self.get_argument("password", "") if username == "root" and password == "123": return usernameclass Home(HttpRequest): @authenticated def get(self): self.write("hello,用户个人中心") @authenticated def post(self): self.write("hello,用户中心")class UserLogin(web.RequestHandler): def get(self): self.write("登录页面")urls = [ (r"/", Home), (settings["login_url"], UserLogin),]if __name__ == '__main__': app = web.Application(urls,**settings) app.listen(port=8888) ioloop.IOLoop.current().start() Title {# 模板注释 #}name={
{name}}name={
{info['name']}}{
{"-".join(address)}}{
{money_format(money)}}
# 判断{% if ... %}{% elif ... %}{% else ... %}{% end %}# 遍历{% for ... in ... %}{% end %}# 循环{% while ... %}{% end %} # 导包{% from ... import ... %}{% import ... %}# 加载其他模板{% include ... %}# 输出原始数据{% raw ... %}# 语句/局部变量{% set 变量名=变量值 %}# 异常处理{% try %}...{% except %}...{% else %}...{% finally %}...{% end %}# 模板继承{% extends *filename* %}{% block 模板块名称} {% end %} # 输出转义数据，tornado在配置中允许通过autoescape=None设置全局转义{ { escape(text) }}# 静态文件存储路径{ { static_url("style.css") }}# 路由反解析reverse_url("路由别名")# CSRF防范机制，CSRF也叫XSRF# tornado开启csrf必须在配置中进行设置 xsrf_cookies = True# 补充：# 在前后端分离项目中，客户端可以通过cookie来读取XSRFToken，cookie名称为_xsrf，请求头必须名称：X-XSRFToken# 在视图方法中可以通过 self.xsrf_token 来获取 XSRFTokentoken{% module xsrf_form_html() %} 转载地址：http://dwhzi.baihongyu.com/