tornado登陆装饰器

tornado作为鼎鼎大名的web异步框架，用来作为高性能服务器以及web框架都是首选。自从python3.4加入了asyncio原生协程后，tornado的最新版本也开始使用了原生的协程。定义协程函数的时候就很简单了，也可以像sanic一样使用async def了。

class LoginHandler(RequestHandler):

    async def post(self, *args, **kwargs):

        ...

        pass

一般的web项目都会有登陆的功能，这就涉及到了登陆验证，在做这一部分功能的时候，我使用的是 jwt -- json web token的方法验证是否登陆。当然也可以使用类似于django的session验证。各自均有优缺点。

通常需要验证是否登陆的模块很多，一般都会写一个登陆验证装饰器，tornado实现了登陆验证装饰器：

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

method(self,*args,**kwargs)即为要装饰的函数。但是如果像上面定义的post方法 async def post(self, *args, **kwargs),作为一个协程函数，就不能再使用这个装饰器了，就需要改写一下这个装饰器。def authenticated_async(method):

初始化app的过程：

import tornado

from peewee_async import Manager

from YourApp.urls import urlpattern

from YourApp.settings import settings, database

if __name__ == "__main__":

    #集成json到wtforms

    import wtforms_json

    wtforms_json.init()

    app = web.Application(urlpattern, debug=True, **settings)

    app.listen(80)

    objects = Manager(database)

    database.set_allow_sync(False)

    app.objects = objects

    tornado.ioloop.IOLoop.current().start()

登陆装饰器改写

    @functools.wraps(method)

    async def wrapper(self, *args, **kwargs):

        tsessionid = self.request.headers.get("tsessionid", None)

        if tsessionid:

            try:

                send_data = jwt.decode(tsessionid, self.settings["secret_key"], leeway=self.settings["jwt_expire"], options={"verify_exp": True})

                user_id = send_data["id"]

                # User 的model类，根据实际情况调整

                #从数据库中获取到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: # 验证jwt 是否过期

                self.set_status(401)

        else:

            self.set_status(401)

        self.finish({})

    return wrapper

巴特西

tornado登陆装饰器

最新文章

热门文章