本文博客地址: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)
标签:get,实现,self,简易,session,cookie,key,data
From: https://www.cnblogs.com/xhsz/p/17205621.html