Tornado是使用Python编写的一个强大的、可扩展的Web服务器。
它非常适合 长时间轮询， WebSocket和其他需要与每个用户建立长期连接的应用程序。常用于编写可扩展的社交应用、实时分析引擎，或RESTful API，微服务
1.一个简单的tornado应用
from typing import Optional, Awaitable, Any
import tornado.httpserver
import tornado.web
from tornado import ioloop
class Application(tornado.web.Application):
def __init__(self):
handlers = [
(r'/', IndexHandlers),
]
tornado.web.Application.__init__(self, handlers, debug=True)
class IndexHandlers(tornado.web.RequestHandler):
def initialize(self) -> None: # 在 Application 的初始化配置参数下被调用
print('123')
def prepare(self) -> Optional[Awaitable[None]]: # 预处理方法, 在执行对应的请求方法之前调用.
print('456')
def get(self):
# self.write_error(400) # 错误
self.write('hello')
def on_finish(self) -> None:
# 在请求处理结束后调用, 在该方法中可进行资源回收或日志处理等一些操作.
# 注意不要在该方法中进行数据的返回
print('finish...')
def write_error(self, status_code: int, **kwargs: Any) -> None:
# 输出一个HTML的出错信息
print('555')
def on_connection_close(self) -> None:
# 对端连接关闭，或者在socket上读写错误的时候被调用，可以让服务器做一些清理。
print('与客户端断开时会被调用')
def get_current_user(self) -> Any:
print('获取当前user')
def get_user_locale(self) -> Optional[tornado.locale.Locale]:
print('给当前用户返回一个 Locale 对象')
def set_default_headers(self) -> None:
self.set_header('custom', 'val')
print('可以用来设置 在响应时的附加首部 (例如可以定制 Server 首部)')
if __name__ == '__main__':
app = Application()
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(8000)
抛开底层，在request请求中，调用顺序
重定向
- self.redirect(permanent=False)可以在RequestHandler中重定向,默认302 Found(临时性转移)，否则就是301 Moved Permanently(永久性转移)
- tornado.web.RedirectHandler同样可以重定向， 默认使用的永久重定向.
handlers = [
(r'/book/(\d+)/?$', BookHandler),
(r'/book/detail/(\d+)$', tornado.web.RedirectHandler, dict(url=r"/book/1/?$"))
]
认证与安全
set_cookie(key, value),可以用来设置cookie，获取cookie，使用get_cookie()
如果你的cookie被用来保存用户重要信息，你应该使用secure_cookie，你需要在配置Application时设置cookie_secret。
设置安全cookie使用set_secure_cookie(),获取时使用get_secure_cookie()
另外，你可以使用expires_days来设置cookie的有效期，默认30天
val = self.get_secure_cookie('username')
print(val)
if not val:
self.set_secure_cookie('username', 'aabc', expires_days=1)
用户认证
self.current_user获取当前用户
实现用户认证, 你需要覆盖请求控制器中的 get_current_user() 方法 来确认怎样获取当前登陆的用户
class BaseHandler(tornado.web.RequestHandler):
def get_current_user(self):
return self.get_secure_cookie("user")
class MainHandler(BaseHandler):
def get(self):
if not self.current_user:
self.redirect("/login")
return
name = tornado.escape.xhtml_escape(self.current_user)
self.write("Hello, " + name)
class LoginHandler(BaseHandler):
def get(self):
self.write('<html><body><form action="/login" method="post">'
'Name: <input type="text" name="name">'
'<input type="submit" value="Sign in">'
'</form></body></html>')
def post(self):
self.set_secure_cookie("user", self.get_argument("name"))
self.redirect("/")
application = tornado.web.Application([
(r"/", MainHandler),
(r"/login", LoginHandler),
], cookie_secret="__TODO:_GENERATE_YOUR_OWN_RANDOM_VALUE_HERE__")
你可以使用 Python 装饰器 (decorator) tornado.web.authenticated 来获取登陆的用户. 如果你的方法被这个装饰器所修饰, 若是当前的用户没有登陆, 则用户会被重定向到 login_url
class MainHandler(BaseHandler):
@tornado.web.authenticated
def get(self):
name = tornado.escape.xhtml_escape(self.current_user)
self.write("Hello, " + name)
settings = {
"cookie_secret": "__TODO:_GENERATE_YOUR_OWN_RANDOM_VALUE_HERE__",
"login_url": "/login",
}
application = tornado.web.Application([
(r"/", MainHandler),
(r"/login", LoginHandler),
], **settings)
如果你的 post() 方法被 authenticated 修饰, 而且用户还没有登陆, 这时服务器会产生一个 403 错误.
第三方认证
tornado.auth 模块既实现了认证, 而且还支持许多知名网站的认证协议, 这其中包括 Google/Gmail, Facebook, Twitter, 和 FriendFeed. 模块内包含了通过这些网站登陆用户的方法, 并在允许的情况下访问该网站的服务. 下面是一个github的例子：
class Application(tornado.web.Application):
def __init__(self):
handlers = [
(r'/oauth2/github', GithubOauth2LoginHandler),
]
tornado.web.Application.__init__(self, handlers, debug=True, autoreload=True,
github_oauth={"key": "",
"secret": ""},
cookie_secret='123secret')
class GithubOauth2Mixin(tornado.auth.OAuth2Mixin):
_OAUTH_AUTHORIZE_URL = "https://github.com/login/oauth/authorize"
_OAUTH_ACCESS_TOKEN_URL = "https://github.com/login/oauth/access_token"
_OAUTH_USERINFO_URL = "https://api.github.com/user"
_OAUTH_NO_CALLBACKS = False
_OAUTH_SETTINGS_KEY = "github_oauth"
async def get_authenticated_user(
self, redirect_uri: str, code: str
) -> Dict[str, Any]:
handler = cast(tornado.web.RequestHandler, self)
http = self.get_auth_http_client()
import json
body = json.dumps(
{
"redirect_uri": redirect_uri,
"code": code,
"client_id": handler.settings[self._OAUTH_SETTINGS_KEY]["key"],
"client_secret": handler.settings[self._OAUTH_SETTINGS_KEY]["secret"],
"grant_type": "authorization_code",
}
)
response = await http.fetch(
self._OAUTH_ACCESS_TOKEN_URL,
method="POST",
headers={"Content-Type": "application/json"},
body=body,
)
from urllib.parse import parse_qs
text = response.body.decode('utf-8')
dic = dict([(k, v[0]) for k, v in parse_qs(text).items()])
res = await http.fetch(
self._OAUTH_USERINFO_URL,
method='GET',
headers={"Authorization": "token {0}".format(dic.get('access_token')), "User-Agent":
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko)"
" Chrome/84.0.4147.89 Safari/537.36"}
)
# 必须设置User-Agent，否则会403，太坑了
return json.loads(res.body)
# return escape.json_decode(response.body)
class GithubOauth2LoginHandler(tornado.web.RequestHandler, GithubOauth2Mixin):
async def get(self):
if self.get_argument('code', default=None): # 获取token和获取用户信息
user = await self.get_authenticated_user(
redirect_uri='http://127.0.0.1:8000/oauth2/github',
code=self.get_argument('code'))
# Save the user with e.g. set_secure_cookie
print(user)
else:
await self.authorize_redirect( # 获取code
redirect_uri='http://127.0.0.1:8000/oauth2/github',
client_id=self.settings['github_oauth']['key'],
scope='',
response_type='code',
extra_params={})
跨站请求伪造防护
你只需要在Application中添加"xsrf_cookies": True,就可以让tornado启用XSRF 保护
评论列表
已有0条评论