基于Tornado自定制仿Django的Session以及Form组件

一、session

1.Session本质是保存在服务器端的数据，可以看作是键值对。

用户第一次打开网站页面
   - 生成一段随机字符串，作为value发给客户端浏览器,客户端带着字符串获取对应的session
   - 在session中保存，随机字符串作为key，value={'user':'Mitsui','pwd':123....}

2.session模块代码：

import time
import hashlib
import settings

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

class RedisSession(object):
    def __init__(self,handler):
        """
        基于Redis在服务端存储session
        :param handler:
        """
        self.handler = handler
        self.session_id = settings.SESSION_ID
        self.expires = settings.EXPIRERS
        self.initial()

    @property
    def conn(self):
        import redis
        conn = redis.Redis(host='192.168.xx.xx',port=6379)
        return conn

    def initial(self):
        client_random_str = self.handler.get_cookie(self.session_id)

        if client_random_str and self.conn.exists(client_random_str):
            self.random_str = client_random_str
        else:
            self.random_str = get_random_str()

        expires = time.time() + self.expires
        self.handler.set_cookie(self.session_id,self.random_str,expires=expires)
        #除了对浏览器cookies设置超时时间，也需要对redis数据设置超时时间，可以定时清除数据节省空间
        self.conn.expire(self.random_str,self.expires)

    def __getitem__(self, item):
        """
        取session,item为key,如self.session['user']
        :param item:
        :return:
        """
        import json
        #由于Python数据类型直接存入redis后受到转换，因此在存储时会先dumps,相应的取值时会先loads。
        data_str = self.conn.hget(self.random_str,item)
        if data_str:
            return json.loads(data_str)
        else:
            return None

    def __setitem__(self, key, value):
        """
        设置session
        :param key:
        :param value:
        :return:
        """
        import json
        self.conn.hset(self.random_str,key,json.dumps(value))

    def __delitem__(self, key):
        """
        删除session
        :param key:
        :return:
        """
        self.conn.hdel(self.random_str)

class CacheSession(object):

    container = {}

    def __init__(self,handler):
        """
        普通的内存存取session
        :param handler: 视图函数传进的self,可以使用get_cookie等方法
        """
        self.handler = handler
        self.session_id = settings.SESSION_ID   #存储在浏览器的cookies的key
        self.expires = settings.EXPIRERS        #超时时间
        self.initial()

    def initial(self):
        """

        :return:
        """
        client_random_str = self.handler.get_cookie(self.session_id)
        if client_random_str and client_random_str in self.container:
            #如果session中已经有值，赋原值刷新超时时间
            self.random_str = client_random_str
        else:
            #没有则获取随机字符串 作为session的key存储在内存中container,并设置超时时间，
            self.random_str = get_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):
        """
        获取session
        :param item:key,  session['user']
        :return:
        """
        return self.container[self.random_str].get(item)

    def __setitem__(self, key, value):
        """
        设置session
        :param key:  session['user']
        :param value:    =user
        :return:
        """
        self.container[self.random_str][key] = value

    def __delitem__(self, key):
        """
        删除session   del self.session时触发
        :param key:
        :return:
        """
        if key in self.container[self.random_str]:
            del self.container[self.random_str][key]



class SessionFactory(object):
    """
    读取配置文件，根据配置文件返回定制的session类
    """
    @staticmethod
    def get_session():
        import settings
        import importlib
        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

　

3.基于Tornado的使用:

import tornado.ioloop
import tornado.web
from tornado.web import RequestHandler
from session_code import SessionFactory

class SessionHandler(object):
    def initialize(self,*args,**kwargs):
        cls = SessionFactory.get_session()
        # cls是CacheSession对象，RedisSession对象
        #执行cls的init方法
        self.session = cls(self)


class LoginHandler(SessionHandler,RequestHandler):
    """
    init方法执行之后，执行get 跟post方法之前，会先执行initialize方法,
    这里用多继承的方法继承自SessionHandler，执行initialize方法,获取
    CacheSession或者RedisSession对象的session方法
    """

    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 == "Mitsui" and pwd == '123':
            self.session['user'] = user
            self.redirect('/index')
        else:
            self.render('login.html')


class IndexHandler(SessionHandler,RequestHandler):

    def get(self, *args, **kwargs):
        user = self.session['user']
        if user:
            self.write('欢迎登录！')
        else:
            self.redirect('/login')


sett = {
    'template_path':'views',
    #XSRF
    'xsrf_cookies':True,
}

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

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

　　

配置文件settings:

#所选的session处理类
SESSION_ENGINE = "session_code.CacheSession"
#客户端cookie中保存的key
SESSION_ID = "__session__id__"
#超时时间
EXPIRERS = 300

　　

 二、自定制Form组件:

1.Form组件核心原理：

import re
import copy

# ##################### 定制插件（HTMl）  #####################
class TextInput(object):
    """
    定制前端页面的标签：
    :return: <input type='text' class="c1" ID='I1' ..../>"
    """
    def __init__(self,attrs=None):
        """
        标签自定制属性功能
        :param attrs: {'class':'c1', .....}
        """
        if attrs:
            self.attrs = attrs
        else:
            self.attrs = {}

    def __str__(self):

        data_list = []
        for k,v in self.attrs.items():
            tmp = "{0}='{1}'".format(k,v)
            data_list.append(tmp)
        tpl = "<input type='text' {0}>".format(" ".join(data_list))
        return tpl


class EmailInput(object):

    def __init__(self, attrs=None):
        if attrs:
            self.attrs = attrs
        else:
            self.attrs = {}

    def __str__(self):

        data_list = []
        for k, v in self.attrs.items():
            tmp = "{0}='{1}'".format(k, v)
            data_list.append(tmp)
        tpl = "<input type='email' {0} />".format(" ".join(data_list))
        return tpl


class PasswordInput(object):

    def __init__(self, attrs=None):
        if attrs:
            self.attrs = attrs
        else:
            self.attrs = {}

    def __str__(self):

        data_list = []
        for k, v in self.attrs.items():
            tmp = "{0}='{1}'".format(k, v)
            data_list.append(tmp)
        tpl = "<input type='password' {0} />".format(" ".join(data_list))
        return tpl

# ##################### 定制字段（正则）  #####################
class Field(object):

    def __str__(self):
        """
        保存用户输入的值，当用户调用过is_valid,则self.value有值，
        在插件中增加属性 value = 用户提交过来的值
        :return: 插件的str值，验证过则新增value = 输入值的属性
        """
        if self.value:
            self.widget.attrs['value'] = self.value

        return str(self.widget)


class CharField(Field):
    default_widget = TextInput
    regex = "w+"

    def __init__(self,widget=None):
        """
        初始化的时候，设置对应的插件，如果传入widget,则使用widget传入的插件对象，
        如果未传入则使用默认的插件对象。
        :param widget: 插件对象，TextInput()、EmailInput()....
        """
        self.value = None
        self.widget = widget if widget else self.default_widget()

    def valid_field(self,value):
        self.value = value
        if re.match(self.regex,value):
            return True
        else:
            return False


class EmailField(Field):
    default_widget = EmailInput
    regex = "w+@w+"

    def __init__(self,widget=None):
        self.value = None
        self.widget = widget if widget else self.default_widget()

    def valid_field(self,value):
        self.value = value
        if re.match(self.regex,value):
            return True
        else:
            return False

# ##################### 定制Form  #####################
class BaseForm(object):

    def __init__(self,data):
        """
        获取在类中生成的所有插件，设置到对象中，并添加到self.fields字典中，
        供is_valid方法对所有注册的插件进行数据验证。
        :param data:
        """
        self.fields = {}
        self.data = data   #用户form表单提交值 {"user":'Mitsui','email':'Mitsui@live.com'}
        #需要使用Form表单时，会继承BaseForm类，实例化生成对象时，self即需要在前端展示的form对象
        #通过type(self)找到Form类，Form类__dict__中包含所有的类的静态字段，即使用form时创建的插件，
        #user = CharField() 插件都是继承自Field类，由此获取所有的插件
        for name,field in type(self).__dict__.items():
            #name:user,    field:CharField()
            if isinstance(field,Field):
                #由于是静态字段，所以使用的是同一个对象，如果对其进行修改，会影响其它的form对象，
                #所以这里通过深拷贝防止对其进行修改
                new_field = copy.deepcopy(field)
                #将类的这些静态字段设置到对象中，方便调用
                setattr(self,name,new_field)
                self.fields[name] = new_field

    def is_valid(self):
        """
        将form组件设置的所有字段循环，交给每一个Field字段验证，如果有一个错误
        返回False,否则返回True
        :return:
        """
        flag = True
        for name,field in self.fields.items():
        # name:user,    field:CharField()
            user_input_val = self.data.get(name)
            result = field.valid_field(user_input_val)
            if not result:
                flag = False
        return flag

# #####################  使用Form #####################

class LoginForm(BaseForm):
    user = CharField()
    email = EmailField(widget=EmailInput())

#Django:
# if request == "GET":
#     form = LoginForm()
#     return render('login.html',{'form':form})
# else:
#     form = LoginForm(request.POST)
#     if form.is_valid():
#         pass
#     else:
#         pass

# Tornado:
# def get(self, *args, **kwargs):
#     form = LoginForm()
#     self.render("login.html",form=form)
#
# def post(self, *args, **kwargs):
#     post_data = {}
#     for key in self.request.arguments:
#         if key == '_xsrf': continue
#         post_data[key] = self.get_arguments(key)[0]
#     form = LoginForm(post_data)
#     if form.is_valid():
#         pass
#     else:
#         self.render('login.html',form=form)

　　

2.支持多个WEB框架的Form表单验证 - Tyrion