Python web框架 Tornado自定义session组件

Posted 2020-10-09

　　我们在学习Django框架的过程中，内部封装了session组件，以方便于我们使用进行验证。但是Tornado框架是没有session的，所以如果想使用session的话，就需要我们自己定制相对应的组件。在Tornado框架中是存在cookie机制的，所以我们可以利用加密的cookie以实现验证机制。通过配置文件决定数据信息存放的位置。是在缓存还是在内存中。具体代码如下：

#!/usr/bin/env python
# _*_ coding:utf-8 _*_
import time
import json
import redis
import hashlib
import importlib
import settings

def gen_random_str():
    """
        获取随机字符串，作为session的key
    """
    md5 = hashlib.md5()
    md5.update(str(time.time()).encode("utf-8"))
    return md5.hexdigest()

    
class CacheSession(object):
    """
    内存中获取
    """
    container = {}  # 保证类实例化一次，仅有一个session，同一个对象也是一个session

    def __init__(self,handler):
        self.handler = handler #接收handler对象
        self.session_id = settings.SESSION_ID #从配置文件中获取session的id
        self.expires = settings.EXPIRERS #从配置文件中获取session的超时时间
        self.initial() #初始化

    def initial(self):

        #先从cookies获取随机字符串，能拿到并且在container中证明之前访问过，session 不变，若是不满足就重新生成
        client_random_str = self.handler.get_cookie(self.session_id) #从cookies中获取session_id
        if client_random_str and client_random_str in self.container: #判断
            self.random_str = client_random_str #存在就赋值
        else:#不存在就创建，或是错误
            self.random_str = gen_random_str() #生成一个新的随机字符串
            self.container[self.random_str] = {}
       #每次请求来，在超时时间之内，都需要回写一次，以延长超时时间
        expires = time.time()+self.expires
        self.handler.set_cookie(self.session_id,self.random_str,expires=expires)

    def __getitem__(self, item):
        return self.container[self.random_str].get(item)

    def __setitem__(self, key, value):
        self.container[self.random_str][key]=value

    def __delitem__(self, key):
        if key in self.container[self.random_str]:
            del self.container[self.random_str][key]

            
class RedisSession(object):
    """
    缓存中获取
    """

    def __init__(self,handler):
        self.handler = handler
        self.session_id = settings.SESSION_ID
        self.expires = settings.EXPIRERS
        self.initial()

    @property
    def conn(self):
        """
        连接redis
        :return:
        """
        conn = redis.Redis(host=settings.RedisHost,port=settings.RedisPort)
        return conn

    def initial(self):
        """
        初始化
        :return:
        """
        client_random_str = self.handler.get_cookie(self.session_id)
        if client_random_str and self.conn.exists(client_random_str): #检测随机字符串及redis中的随机字符串是否存在
            self.random_str = client_random_str
        else:
            self.random_str = gen_random_str()

        expires = time.time()+self.expires
        self.handler.set_cookie(self.session_id,self.random_str,expires=expires)
        self.conn.expire(self.random_str,self.expires) #设置缓存数据额度超时时间

    def __getitem__(self, item):
        #注意在redis中获取的数据全是字节类型，需要转成字符串再json反序列化
        data_str = self.conn.hget(self.random_str,item).decode("utf-8")
        data_dict = json.loads(data_str) if data_str else None
        return data_dict

    def __setitem__(self, key, value):
        self.conn.hset(self.random_str,key,json.dumps(value))

    def __delitem__(self, key):
        self.conn.hdel(self.random_str,key)

        
class SessionFactory(object):
    """
    工厂模式，通过导入的配置文件，实例化对应方法的
    """
    @staticmethod
    def get_session():
        engine = settings.SESSION_ENGINE
        module_path,cls_name = engine.rsplit(‘.‘,maxsplit=1)
        md = importlib.import_module(module_path) #以字符串的形式导入
        cls = getattr(md,cls_name) #反射验证，获取有没有这个方法
        return cls #有的话返回这个类

session_code.py

#!/usr/bin/env python
# _*_ coding:utf-8 _*_
#配置文件

# session 存储类型

# 存储在内存中
SESSION_ENGINE = "session_code.CacheSession"

# 存储在redis缓存中
# SESSION_ENGINE = "session_code.RedisSession"
# RedisHost = ""
# RedisPort = 6379

SESSION_ID = "__session__id__"
EXPIRERS = 300

settings.py

#!/usr/bin/env python
# _*_ coding:utf-8 _*_

import tornado.ioloop
import tornado.web
from tornado.web import RequestHandler
from session_code import SessionFactory #导入自定义session文件的工厂方法

#创建可以调用session方法的类，让其他需要使用的类去继承
class SessionHandler(object):

    def initialize(self,*args,**kwargs):
        """
        重写实例化中的钩子函数，用于自定义实现某些方法
        :param args:
        :param kwargs:
        :return:
        """
        cls = SessionFactory.get_session() #获取使用的类方法
        self.session = cls(self) #构造方法，把当前的请求类(handler)的对象(self)作为参数传递给session__code进行构造session。

class LoginHandler(SessionHandler,RequestHandler):

    def get(self, *args, **kwargs):
        self.render("login.html")

    def post(self, *args, **kwargs):
        
        user = self.get_argument("user")
        pwd = self.get_argument("pwd")
        
        # 简单实现用户登录认证，可以从数据库中获取数据
        if user=="aaa" and pwd == "123":
            # 验证，在跳转之前，把信息写入session中
            self.session["user"]=user # 写入session
            self.redirect("/index") # 跳转到index页面
        else:
            self.render("login.html") # 否则的话，就跳转到登录页面

class IndexHandler(SessionHandler,RequestHandler):

    def get(self, *args, **kwargs):
        user = self.session["user"] # 从session中获取信息
        if user:
            self.write("欢迎登录")
        else:
            self.redirect("/login")

set = {
    "template_path":"views",
}

application = tornado.web.Application([
    (r"/login",LoginHandler),
    (r"/index",IndexHandler),
],**set)

if __name__ == ‘__main__‘:
    application.listen(8888)
    tornado.ioloop.IOLoop.instance().start()

app.py（应用）

　　遵循MVC模式使用Tornado框架，创建用于存储页面的views文件夹。在app.py文件应用中，设置对于的配置信息。views文件夹下简单页面代码如下：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
    <h1 class="c1">登录</h1>
<form method="POST">
    <input type="text" name="user">
    <input type="text" name="pwd">
    <input type="submit" value="提交">
</form>
</body>
</html>

login.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>

</body>
</html>

index.html