我们在学习Flask的时候学习过flask-login库进行登录管理，在tornado同样存在类似的功能authenticated。我们可以使用这个装饰器进行登录权限验证。
Tornado的原生装饰器
我们看下装饰器authenticated的源码，分析下工作原理。
def authenticated(method):
"""Decorate methods with this to require that the user be logged in.
If the user is not logged in, they will be redirected to the configured
`login url <RequestHandler.get_login_url>`.
If you configure a login url with a query parameter, Tornado will
assume you know what you're doing and use it as-is. If not, it
will add a `next` parameter so the login page knows where to send
you once you're logged in.
"""
@functools.wraps(method)
def wrapper(self, *args, **kwargs):
if not self.current_user:
if self.request.method in ("GET", "HEAD"):
url = self.get_login_url()
if "?" not in url:
if urlparse.urlsplit(url).scheme:
# if login url is absolute, make next absolute too
next_url = self.request.full_url()
else:
next_url = self.request.uri
url += "?" + urlencode(dict(next=next_url))
self.redirect(url)
return
raise HTTPError(403)
return method(self, *args, **kwargs)
return wrapper
源码中我们看到当self.current_user为空的时候将会执行下面的页面跳转。
我们再看下这个self.current_user的源码。
@property
def current_user(self):
"""The authenticated user for this request.
This is set in one of two ways:
* A subclass may override `get_current_user()`, which will be called
automatically the first time ``self.current_user`` is accessed.
`get_current_user()` will only be called once per request,
and is cached for future access::
def get_current_user(self):
user_cookie = self.get_secure_cookie("user")
if user_cookie:
return json.loads(user_cookie)
return None
* It may be set as a normal variable, typically from an overridden
`prepare()`::
@gen.coroutine
def prepare(self):
user_id_cookie = self.get_secure_cookie("user_id")
if user_id_cookie:
self.current_user = yield load_user(user_id_cookie)
Note that `prepare()` may be a coroutine while `get_current_user()`
may not, so the latter form is necessary if loading the user requires
asynchronous operations.
The user object may be any type of the application's choosing.
"""
if not hasattr(self, "_current_user"):
self._current_user = self.get_current_user()
return self._current_user
我们看到文档注释知道current_user为RequestHandler的动态属性有两种方式去赋予初值。
我们再看下get_current_user的源码：
def get_current_user(self):
"""Override to determine the current user from, e.g., a cookie.
This method may not be a coroutine.
"""
return None
这是一个RequestHandler的方法，通过重写这个方法我们可以设置当前登录用户。
注意：这个方法默认返回的是None，并且调用这个方法是同步的，如果重写的时候涉及到去查询数据库就会耗时。如果写成协程，上层调用是不支持的。
我们接着看authenticated的源码部分，当用户为登录的时候回去调用url = self.get_login_url()。我们看下get_login_url()的源码：
def get_login_url(self):
"""Override to customize the login URL based on the request.
By default, we use the ``login_url`` application setting.
"""
self.require_setting("login_url", "@tornado.web.authenticated")
return self.application.settings["login_url"]
会从配置中查找一个login_url进行返回。
这样是符合前后端不分离的单体项目，但是我们要是前后端分离就很不友好了。我们重写改写这个装饰器来完成我们的要求。
我们自己编写的装饰器
我们为了在用户没有登陆的时候返回json串而不是页面。我们重新抒写一个基于jwt的装饰器。
def authenticated_async(method):
@functools.wraps(method)
async def wrapper(self, *args, **kwargs):
tsessionid = self.request.headers.get("tsessionid", None)
if tsessionid:
# 对token过期进行异常捕捉
try:
# 从 token 中获得我们之前存进 payload 的用户id
send_data = jwt.decode(tsessionid, self.settings["secret_key"], leeway=self.settings["jwt_expire"],
options={"verify_exp": True})
user_id = send_data["id"]
# 从数据库中获取到user并设置给_current_user
try:
user = await self.application.objects.get(User, id=user_id)
self._current_user = user
# 此处需要使用协程方式执行 因为需要装饰的是一个协程
await method(self, *args, **kwargs)
except User.DoesNotExist as e:
self.set_status(401)
except jwt.ExpiredSignatureError as e:
self.set_status(401)
else:
self.set_status(401)
self.finish({})
return wrapper
这样只要我们在需要权限的地方加上权限装饰器即可。
class GroupHandler(RedisHandler):
@authenticated_async
async def get(self, *args, **kwargs):
网友评论