实现简易session

本文博客地址：https://www.df100.ltd/387.asp

环境：python3.10（主要库：sanic，sqlalchemy），mysql

需求：在一个函数内，给一个字典，直接更改它就相当于更改session

例子：

def helloWorld(get_or_post,session):
    lst = int(get_or_post('lst')) #获取http参数
    try:print(session['lst'])
    except:session['lst'] = None
    if(session['lst'] == None):
        session['lst'] = lst
    else:
        session['lst'] = session['lst']*0.8
    return {'async':False,
    'data':text(str(get_or_post('hello'))+'hello world'+str(session)),
    'cookie':{'Session_key':''}
    }

需要将session保存在mysql里面，表如下（sqlalchemyORM）：

class BaseMixin:
    """model的基类,所有model都必须继承"""
    id = Column(Integer, primary_key=True)
    created_at = Column(Integer, nullable=False)
    updated_at = Column(Integer, nullable=False, index=True)
    deleted_at = Column(Integer)  # 可以为空, 如果非空, 则为软删
class session(BaseMixin): #会话
    __tablename__ = "session" #表名
    # key = Column(String(64),nullable=False,primary_key=True)
    id = Column(String(64),nullable=False,primary_key=True)
    data = Column(String(64)) #会话内容(json键值对)

首先需要实现session操作类：

框架：

class session():
    data = {}
    def __init__(self,session_key):
      pass
    def create(self): #创建sessionkey
        '''
        Create session
        Create a session key is available
        -> session key
        '''
    def get(self): #得到数据库里面的session
        pass
    def refresh(self): 
        '''刷新session,将内存里面的session提交到数据库'''

@contextlib.contextmanager
def get_session():
    s = Session()
    try:
        yield s
        s.commit()
    except Exception as e:
        s.rollback()
        raise e
    finally:
        s.close()

创建session，本质上就是创建一个sessionkey

with get_session() as s:
    session = db.table.session()
    # print(str(uuid.uuid4()))
    session.id = str(uuid.uuid4()) #生成唯一id
    session.created_at = time.time()
    session.updated_at = time.time()
    s.add(session)
    s.commit()
    self.key = session.id
    #return session.id

得到数据库里面的session其实就是查询

with get_session() as s:
    self.data = s.query(session_db).filter(session_db.id == self.key,
    session_db.deleted_at==None).first().dobule_to_dict()['data'] # 查询数据库
    if(self.data != None):
        self.data = json.loads(self.data) #编码成json
    else:
        self.data = {}

刷新session,将内存里面的session提交到数据库也还是查询

    def refresh(self): 
        '''刷新session,将内存里面的session提交到数据库'''
        with get_session() as s:
            s.query(session_db).filter(session_db.id == self.key).update({'data': json.dumps(self.data),
                                                                             'updated_at':time.time()})

然后就可以想办法获取到cookie里面的sessionkey，如果不存在可以创建session，然后就设置cookie

async def api(request,name): 
    data = main(request,name)
    ret = data['data']
    for k,v in data['cookie'].items(): #设置cookie
        ret.cookies[k] = v
    if(data['async']):
        return await ret
    else:
        return ret
app.add_route(api,f'/api/<name:path>',methods=['GET','POST'])

def main(request,name):
    '''
    ->
        {'async': False, 
        'data': <HTTPResponse: 200 text/plain; charset=utf-8>,
        'cookie' : {}
        }
    '''
    def get_or_post(key): #如果没有GET参数就用post
        if(request.args.get(key) != None):
            return request.args.get(key)
        elif(request.form.get(key) != None):
            return request.form.get(key)
        return None
    cookie_session_key = request.cookies.get("Session_key")
    session_odj = session(cookie_session_key)
    if(cookie_session_key == None or cookie_session_key == ''):session_odj.create() #创建Session_key
    session_odj.get() # 将数据库的session同步到内存
    # try:
    ret = helloWorld(get_or_post,session_odj.data) #执行helloWorld函数并且传入session对象的字典
    session_odj.refresh() # 提交内存的session
    ret['cookie']['Session_key'] = session_odj.key # 更新cookie的session_key
    return ret
    # except:
    #     return empty(status=404)