编程学习网 > 编程语言 > Python > Python_学习之框架Tornado教程
2023
08-14

Python_学习之框架Tornado教程

清华大佬耗费三个月吐血整理的几百G的资源，免费分享！....>>>

Tornado是一款非阻塞的web服务器框架。


官方文档：https://www.tornadoweb.org/en/stable/guide.html

安装：pip install tornado

一、快速搭建web应用
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.web
import tornado.ioloop
import tornado.options

# 编写视图类
class HealthCheck(tornado.web.RequestHandler):


   def get(self):
       self.write({"code": 0, "err": "ok", "data": {}})


# 路由映射
app = tornado.web.Application([
  (r"/health_check", HealthCheck),  # 访问路径对应的视图类
], debug=True)

# 启动服务
if __name__ == '__main__':
   # 单进程
   # app.listen(9527)
   # current返回当前线程循环实例，然后启动循环，同时开启监听
   # tornado.ioloop.IOLoop.current().start()

   # 多进程 【linux上部署】
   import tornado.httpserver

   http_server = tornado.httpserver.HTTPServer(app)
   http_server.bind(9527)
   http_server.start(4)
   tornado.ioloop.IOLoop.current().start()


二、钩子函数
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import tornado.web
import tornado.ioloop


class HealthCheck(tornado.web.RequestHandler):
   def set_default_headers(self):
       print("执行【set_default_headers】钩子函数：设置默认的响应头")

   def initialize(self, python, life) -> None:
       self.python = python
       self.life = life
       print("执行【initalize】钩子函数：获取路由中的字典，作为参数，供视图http方法使用")

   def prepare(self):
       print("执行【prepare】钩子函数：在处理视图之前，可以处理些问题")

   def get(self):
       try:
           # write并不会马上返回给请求方，而是放在缓存里，如果后面都执行完了，才会返回
           self.write({"code": 0, "err": "ok", "data": {}})
           print(f"执行视图函数……{self.python}|{self.life}")
           import time
           time.sleep(2)
           raise TypeError("值异常")
       except Exception as ex:
           # 如果视图异常了，可以通过send_error来主动抛出异常，然后通过write_error进行处理异常
           self.send_error(status_code=501, error_msg=ex.args[0])

   def on_finish(self) -> None:
       """请求处理完后干的事情，不是异步的，前一个请求虽然很快接到响应，
      但后一个请求进来需要等日志写完后才能处理新的请求"""
       import time
       time.sleep(5)
       print("执行【on_finish】钩子函数：视图处理完后做的事情，比如记录日志")

   def write_error(self, status_code: int, **kwargs) -> None:
       """视图中出现异常时执行"""
       print("执行异常处理【write_error】钩子函数")
       self.write(dict(code=status_code, desc=kwargs["error_msg"]))


# 路由映射
app = tornado.web.Application([
  (r"/", HealthCheck, {"python": "3.7.8", "life": dict(sale_price=20000)}),  # 访问路径对应的视图类
], debug=True)

if __name__ == '__main__':
   """python """
   # 单进程
   try:
       app.listen(9527)
       # current返回当前线程循环实例，然后启动循环，同时开启监听
       tornado.ioloop.IOLoop.current().start()
   except OSError as e:
       print(e)
       tornado.ioloop.IOLoop.current().stop()


       
# 如果视图函数不报异常，钩子函数执行顺序
执行【set_default_headers】钩子函数：设置默认的响应头
执行【initalize】钩子函数：获取路由中的字典，作为参数，供视图http方法使用
执行【prepare】钩子函数：在处理视图之前，可以处理些问题
执行视图函数……3.7.8|{'sale_price': 20000}
执行【on_finish】钩子函数：视图处理完后做的事情，比如记录日志

# 如果视图出现异常，执行顺序为
执行【set_default_headers】钩子函数：设置默认的响应头
执行【initalize】钩子函数：获取路由中的字典，作为参数，供视图http方法使用
执行【prepare】钩子函数：在处理视图之前，可以处理些问题
执行视图函数……3.7.8|{'sale_price': 20000}
执行【set_default_headers】钩子函数：设置默认的响应头
执行异常处理【write_error】钩子函数
执行【on_finish】钩子函数：视图处理完后做的事情，比如记录日志


三、tornado.options 全局参数定义应用
#!/usr/bin/env python
# -*- coding:utf-8 -*-

# 提供了全局参数的定义，储存和转换，如加载命令行参数，自定义config文件
import tornado.options  

if __name__ == '__main__':
   # 方式1. 接收命令行参数
   import argparse
   parser = argparse.ArgumentParser()
   parser.add_argument("-p", "--port", type=int, default=8600)
   args = parser.parse_args()
   print(args.port)

   # 方式2. tornado自带的处理参数
   # 定义1
   tornado.options.define(
       # 自定义的命令行参数名，唯一性,会作为options的属性
       name="port",
       # 变量默认值,默认None
       default=8600,
       # 变量值的类型，支持str,int,float,datetime,bool等,默认转化为对应的类型
       type=int,
       # 变量说明信息
       help="set service port",
       # 变量值举例
       metavar="8000|9000|9100"
  )
   # 定义2,
   tornado.options.define(name="hobby", default=[], type=str, multiple=True,
                          help="开启multiple为True接收多个值，以列表接收",
                          metavar="playBall|watchTv")
   # 定义3，
   # tornado.options.define_logging_options()

   # 获取【命令行方式传入】 python Main.py --port=8000 --hobby=money,watchTv
   tornado.options.parse_command_line()

   # 查看使用方法
   tornado.options.options.print_help()
# 使用
   print(tornado.options.options.port)
   print(tornado.options.options.hobby)


四、request对象及路由参数解析
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# author:SunXiuWen
# datetime:2021/6/17 0017 15:59
import json
import tornado.web
import tornado.ioloop

"""
self.request.body 所有请求方法的原生数据，二进制
self.request.body_arguments 所有请求方法中的body部分带的数据
self.request.query_arguments 所有请求方法url后拼接的数据
"""


class HealthCheck(tornado.web.RequestHandler):

   def get(self):
       # 获取get请求url后拼接的参数:http://localhost:8600?version= 1.0 &version=20000
       # 获取单个参数,如果变量名相同，后面的覆盖前面的
       print(self.get_query_argument(
           # 请求参数变量名
           "version",
           default="定义获取不到时默认值，默认为None",
           # 是否对值进行左右去空,默认为True
           strip=True))

       # 获取相同参数同时存在的多个值，如xxx?version=1&version=2
       print(self.get_query_arguments("version"))

       self.write({"code": 0, "err": "ok-get", "data": {}})

   def post(self):
       # TODO：非json请求的报文
       # print(self.get_body_argument("version"))
       # print(self.get_body_arguments("version"))
       # # 或者,下面两个方法同时支持get和post方法
       # print(self.get_argument("version"))
       # print(self.get_arguments("version"))

       # TODO: json请求报文
       print(self.request.body)  # 获取原生post请求来的参数
       print(self.request.query_arguments)  # 获取的都是原生的url后拼接的参数
       print(self.request.body_arguments)  # 获取非json传过来的原生body参数

       self.write({"code": 0, "err": "ok-post", "data": {}})


class Index(tornado.web.RequestHandler):
   def initialize(self, types) -> None:
       """路由中的字典会传到initalize方法当成参数"""
       self.types = types
       print(self.types)

   def get(self):
       bytes_to_str = self.request.body.decode(encoding="utf-8")

       self.write(bytes_to_str)


class Order(tornado.web.RequestHandler):
   def get(self, num=12, args_name=333, p2=4444):
       """url后面拼接的正则以参数的形式传给视图，同django与flask一样"""
       self.write(f"your want to get {num} page|the {args_name}|{p2}")

   def post(self, page=1):
       """查看request对象的属性"""
       print(self.request.method)  # 获取请求方法名 POST
       print(self.request.host)  # 获取被请求主机名即服务器名 127.0.0.1:32003
       print(self.request.uri)  # 获取完整资源路径，资源状态标志位 /order/222222
       print(self.request.path)  # 请求地址
       print(self.request.full_url()) # 请求完整地址 http://localhost:9527/order/222222?xxx=33
       print(self.request.query)  # 请求参数,包含get的参数
       print(self.request.version)  # HTTP/1.1
       print(self.request.headers)   # 获取http请求头信息
       print(self.request.body)      # 获取请求原始二进制报文
       print(self.request.remote_ip)  # 客服端地址 127.0.0.1
       print(self.request.files)  # 获取上传的文件
       self.write(f"ok：{page}")


app = tornado.web.Application([
  (r"/health_check", HealthCheck),
  (r"/index", Index, {"types": 1}),
  (r"/order/([0-9]+)", Order),
  (r"/order1/(?P<args_name>\w+)/(?P<p2>\w+)", Order)
], debug=True)

if __name__ == '__main__':
   app.listen(9527)
   tornado.ioloop.IOLoop.current().start()


五、response响应
from tornado import ioloop
from tornado import web

# 项目配置
settings = {
   "debug": True,  # 开启debug模式
}

# 视图类必须要直接或者间接继承于 web.RequestHandler
from typing import Union
from tornado.util import unicode_type
from tornado import escape
from tornado.escape import utf8


class HttpRequest(web.RequestHandler):
   def write(self, chunk: Union[str, bytes, dict, list]) -> None:
       if self._finished:
           raise RuntimeError("Cannot write() after finish()")
       if not isinstance(chunk, (bytes, unicode_type, dict, list)):
           message = "write() only accepts bytes, unicode, and dict objects"
           raise TypeError(message)
       if isinstance(chunk, (dict, list)):
           chunk = escape.json_encode(chunk)
           self.set_header("Content-Type", "application/json; charset=UTF-8")
       chunk = utf8(chunk)
       self._write_buffer.append(chunk)


class Home(HttpRequest):
   def get(self):
       # write 会自动识别
       # self.write("<h1>hello world</h1>") # 响应html文档
       # self.write({"msg":"ok"}) # 响应json数据

       # 注意，json的数据格式也可以是列表,tornado中默认不支持返回list，所以如果返回list，则需要重写write
       self.write([1, 2, 3])
       import time
       time.sleep(3)
       self.write("客户端会先接受到上面的【1，2，3】等3秒后又会接到这个数据")
       
       # 重新设置响应头的内容 set_header[修改]
       self.set_header("Content-Type", "text/json; charset=gbk")
       # 自定义响应头 add_header[添加]
       self.add_header("test", "xxxx")
       self.add_header("python", "yyyy")
       # self.clear_header("Server") # 从响应头中删除指定名称的响应头信息

   def post(self):
       # self.set_status(404,"No User") # 第二个参数表示响应描述，如果不设置，则显示原来对应的
       # self.send_error(500,reason="服务器炸了！")
       self.send_error(404, msg="服务器炸了！", info="快报警")  # 要使用send_error必须先声明send_error方法

   def write_error(self, status_code, **kwargs):
       self.write("<h1>完蛋啦...</h1>")
       self.write("<p>错误信息:%s</p>" % kwargs["msg"])
       self.write("<p>错误描述:%s</p>" % kwargs["info"])

   def put(self):
       # 页面响应
       self.redirect("http://www.oldboyedu.com")


# 路由列表
urls = [
  (r"/", Home),
]

if __name__ == "__main__":
   # Application是tornado web框架的核心应用类，是与服务器对应的接口，里面保存了路由映射表
   app = web.Application(
       urls,
       **settings,
  )
   # 设置监听的端口和地址
   # 启动http多进程服务
   app.listen(port=8000, address="0.0.0.0")
   # ioloop，全局的tornado事件循环，是服务器的引擎核心，start表示创建IO事件循环,等待客户端连接
   ioloop.IOLoop.current().start()
   
# tornado对响应数据的处理，有三种模式
self.write(xxx)  将chunk数据块写入缓存，默认允许有多个write(),等视图所有的逻辑处理完了，最后都会将缓存的里数据，发送给前端
self.flush()     表示立即将write写入缓存的数据发给前端，不等待视图是否执行完毕
self.finish()    表示视图发给前端的数据到此为止，后面如果还有write将不支持

# 同时也支持页面渲染
self.rnder("index.html")  返回渲染完的html
self.redirect('/test')    301临时重定向
self.redirect('/test',permanent=True)    302永久重定向


六、应用分发及路由分发
flask、django为了实现应用解耦，都有实现应用与路由分发，tornado默认是没有的，需要自己编写，下面提供一个自己用的

1. 目录结构

├─app_admin
│ ├─controllers
│ │ └─__init__.py
│ │ └─account.py  
│ └─__init__.py
│ └─__urls__.py
├─app_check
│ ├─controllers
│ │ └─__init__.py
│ │ └─bill.py
│ └─__init__.py
│ └─__urls__.py
├─settings.py
├─Main.py


2. 以app_admin应用为例

# 在app_admin/controllers/account.py中编写视图
import tornado.web


class Login(tornado.web.RequestHandler):
   def get(self):
       self.write("welcome to learn tornado!")

# 在app_admin/urls.py文件中配置本应用路由
from app_admin.controllers import account

patterns = [
  (r"/login", account.Login),
]

# 在app_admin/__init__.py中导入urls.py文件的路由配置列表，便于之后总入口的归集
from .urls import patterns


3. settings.py文件中配置各自的二级域名及应用路由地址

# 分发app的二级域名配置
routes = (
  {
      "host_pattern": "www.tornado.com", # 二级域名，阿里云平台可以自己配，本地需要改hosts文件做对应的dns映射
      "route_path": "app_admin.urls",
      "route_name": "patterns"
  },
  {
      "host_pattern": "admin.tornado.com",
      "route_path": "app_check.urls",
      "route_name": "patterns"
  }
)


4. 入口Main.py文件中进行加载就可以实现应用分发，路由分发，且支持二级域名

import tornado.web
import tornado.ioloop


class HealthCheck(tornado.web.RequestHandler):

   def set_default_headers(self):
       """设置默认响应头"""
       self.set_header("Content-type", "application/json; charset=utf-8")

   def get(self):
       self.write({"code": 0, "err": "ok", "data": {}})

   def on_finish(self) -> None:
       """请求处理完后干的事情，不是异步的，前一个请求虽然很快接到响应，
      但后一个请求进来需要等日志写完后才能处理新的请求"""
       # print(self.get_query_argument("a"))
       # print(self.get_arguments("b"))
       import time
       time.sleep(5)
       print("记录日志")


# 路由分发加载方法:二级域名
def load_route(application_obj):
   import settings
   for route in settings.routes:
       print(route)
       host_pattern = route["host_pattern"]
       route_path = route["route_path"]
       route_name = route["route_name"]

       m = __import__(route_path)
       pattern_list = getattr(m, route_name)
       # 动态将路由添加到路由配置字典中
       application_obj.add_handlers(host_pattern, pattern_list)


app = tornado.web.Application([
  (r"/health_check", HealthCheck),
])

load_route(app)

if __name__ == '__main__':
   # 1. 接收命令行参数
   import argparse

   parser = argparse.ArgumentParser()
   parser.add_argument("-p", "--port", type=int, default=8600)
   args = parser.parse_args()

   # 单进程
   app.listen(args.port)
   # current返回当前线程循环实例，然后启动循环，同时开启监听
   tornado.ioloop.IOLoop.current().start()


七、cookie的应用
import tornado.web
import tornado.ioloop


class AuthCheck(tornado.web.RequestHandler):
   def get(self):
       # 获取cookie
       if not self.get_cookie("api-token"):
           # 配置cookie
           self.set_cookie("api-token", "xxxxxxxx")
           self.write("login.html")
       else:
           self.write("index.html")


class AuthCheckPlus(tornado.web.RequestHandler):
   def get(self):
       # 一般来说cookie是明文存放在浏览器不安全，很容易被伪造，因此，可以通过加密签名的方式防止，tornado有提供方法
       # 需要在路由中配置一个密钥cookie_secret="xxxx"
       if not self.get_secure_cookie("api-token"):
           # 配置加密的cookie
           self.set_secure_cookie("api-token", "xxxx")
           self.write("login.html")
       else:
           self.write("index.html")


app = tornado.web.Application([
  ("/login", AuthCheck),
  ("/login/plus", AuthCheckPlus),
], cookie_secret="11111111111")

if __name__ == '__main__':
   app.listen(9527)
   tornado.ioloop.IOLoop.current().start()


八、模板语法
快速构建

main.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# datetime:2021/8/4 0004
import os
import datetime
import tornado.web
import tornado.ioloop
import tornado.options

settings = {
   "debug": True,
   # 静态文件url地址前缀【注意尾部加/】
   "static_url_prefix": "/static/",
   # 静态文件保存路径
   "static_path": os.path.join(os.path.dirname(__file__), 'static'),
   # html文件路径
   "template_path": os.path.join(os.path.dirname(__file__), 'templates'),
}

tornado.options.define("port", default=9527, type=int, help="define a port for this web")


class Index(tornado.web.RequestHandler):
   @staticmethod
   def salary_format(data):
       return "%.2f" % data

   def get(self):
       person_info = dict(name="boss", salary=20000, company=["beijing", "tencent"])
       self.render('index.html', datetime=datetime, info=dict(name="boss"), salary_format=self.salary_format,
                   **person_info)


urls = [
   tornado.web.url(f"/", Index, {}, name="index"),
]

if __name__ == '__main__':
   tornado.options.parse_command_line()
   app = tornado.web.Application(urls, **settings)

   app.listen(tornado.options.options.port)
   tornado.ioloop.IOLoop.current().start()


templates/index.html

<!DOCTYPE html>
<html lang="en">
<head>
   <meta charset="UTF-8">
   <title>Title</title>
</head>
<body>
<p>当前时间:{{datetime.datetime.now()}}</p>
<p>name={{name}}</p>
<p>salary={{salary}}</p>
<p>{{info['name']}}</p>
<p>{{"-".join(company)}}</p>
<p>{{salary_format(salary)}}</p>
</body>
</html>
2. 模板语法


# 变量: 支持切片及python相应的基本方法，也可以传的是函数
{{ 变量 }}
{{ 变量[index] }}

# 内置变量
request

# 控制语句
# 判断
{% 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() %}

以上就是“Python_学习之框架Tornado教程”的详细内容，想要了解更多Python教程欢迎持续关注编程学习网。

扫码二维码 获取免费视频学习资料

Python编程学习

查 看2022高级编程视频教程免费获取