flask-session（flasksession更新）

flask-session（flasksession更新）

请求进来后： app.__call__ app.wsgi_app wsgi_app---->ctx.push()----->RequestContext.push()---->上下文这里不多说了省略 ---这样一段代码就是session---> if self.session is None: session_interface = self.app.session_interface self.session = session_interface.open_session( self.app, self.request ) if self.session is None: self.session = session_interface.make_null_session(self.app) app.session_interface = SecureCookieSessionInterface()# 重点：open_session(self, app, request) save_session(self, app, session, response)class SecureCookieSessionInterface(SessionInterface): serializer = session_json_serializer session_class = SecureCookieSession def get_signing_serializer(self, app): if not app.secret_key: return None signer_kwargs = dict( key_derivation=self.key_derivation, digest_method=self.digest_method ) return URLSafeTimedSerializer(app.secret_key, salt=self.salt, serializer=self.serializer, signer_kwargs=signer_kwargs) def open_session(self, app, request): # app.secret_key s = self.get_signing_serializer(app) if s is None: return None # 获取cookie中随机字符串 val = request.cookies.get(app.session_cookie_name) if not val: return self.session_class() max_age = total_seconds(app.permanent_session_lifetime) try: data = s.loads(val, max_age=max_age) # 其实就是等于SecureCookieSession对象；self.session=SecureCookieSession return self.session_class(data) except BadSignature: return self.session_class() def save_session(self, app, session, response): domain = self.get_cookie_domain(app) path = self.get_cookie_path(app) if not session: if session.modified: response.delete_cookie( app.session_cookie_name, domain=domain, path=path ) return if session.accessed: response.vary.add('Cookie') if not self.should_set_cookie(app, session): return = self.get_cookie_ secure = self.get_cookie_secure(app) samesite = self.get_cookie_samesite(app) expires = self.get_expiration_time(app, session) val = self.get_signing_serializer(app).dumps(dict(session)) response.set_cookie( app.session_cookie_name, val, expires=expires, domain=domain, path=path, secure=secure, samesite=samesite ) open_session（）--返回---> session_class() = class SecureCookieSession(CallbackDict, SessionMixin): pass SecureCookieSession其实就是个（dict）容器综上：self.session其实就是等于SecureCookieSession对象；self.session=SecureCookieSession设置值session['xxx']='xxx'其实就是去空字典（dict）容器中写入值app.session_interface = SecureCookieSessionInterface()响应----->wsgi_app()------>response = self.full_dispatch_request()----->self.finalize_request()---->self.process_response(response)--->--->最后执行了---self.session_interface.save_session(self, ctx.session, response)---------保存值获取值session['xxx']

总结：

flask-session Flask中的session处理机制（内置：将session保存在加密cookie中实现） - 请求刚到来：获取随机字符串，存在则去“数据库”中获取原来的个人数据，否则创建一个空容器。 --> 内存：对象（随机字符串，{放置数据的容器}） # 1. obj = 创建SecureCookieSessionInterface() # 2. obj = open_session(self.request) = SecureCookieSession() # self.session = SecureCookieSession()对象。 self.session = self.app.open_session(self.request) - 视图：操作内存中 对象（随机字符串，{放置数据的容器}） - 响应：内存对象（随机字符串，{放置数据的容器}） - 将数据保存到“数据库” - 把随机字符串写在用户cookie中。 - 自定义 请求刚到来： # 创建特殊字典，并添加到Local中。 # 调用关系： # self.session_interface.open_session(self, request) # 由于默认app中的session_interface=SecureCookieSessionInterface() # SecureCookieSessionInterface().open_session(self, request) # 由于默认app中的session_interface=MySessionInterFace() # MySessionInterFace().open_session(self, request) self.session = self.app.open_session(self.request) 调用： session -> LocalProxy -> 偏函数 -> LocalStack -> Local 请求终止： # 由于默认app中的session_interface=SecureCookieSessionInterface() # SecureCookieSessionInterface().save_session(self, app, session, response) # 由于默认app中的session_interface=MySessionInterFace() # MySessionInterFace().save_session(self, app, session, response) - flask-session组件 - 使用： from flask import Flask,session from flask_session import RedisSessionInterface app = Flask(__name__) app.secret_key = 'suijksdfsd' # 方式一 from redis import Redis conn = Redis() app.session_interface = RedisSessionInterface(conn,key_prefix='__',use_signer=False) # 方式二 from redis import Redis from flask.ext.session import Session app.config['SESSION_TYPE'] = 'redis' app.config['SESSION_REDIS'] = Redis(host='192.168.0.94',port='6379') Session(app) @app.route('/') def index(): session['xxx'] = 123 return 'Index' if __name__ == '__main__': app.run() - 源码： - 流程 PS: 问题：设置cookie时，如何设定关闭浏览器则cookie失效。 response.set_cookie('k','v',exipre=None) 总结： 1. 内置原理 2. 如何进行自定义 3. flask-session组件使用和原理

from flask import Flask, session, redirect, url_for, escape, request app = Flask(__name__) @app.route('/')def index(): if 'username' in session: return 'Logged in as %s' % escape(session['username']) return 'You are not logged in' @app.route('/login', methods=['GET', 'POST'])def login(): if request.method == 'POST': session['username'] = request.form['username'] return redirect(url_for('index')) return '''

''' @app.route('/logout')def logout(): # remove the username from the session if it's there session.pop('username', None) return redirect(url_for('index')) # set the secret key. keep this really secret:app.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'

基本使用

pip3 install Flask-Session run.py from flask import Flask from flask import session from pro_flask.utils.session import MySessionInterface app = Flask(__name__) app.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT' app.session_interface = MySessionInterface() @app.route('/login.html', methods=['GET', "POST"]) def login(): print(session) session['user1'] = 'alex' session['user2'] = 'alex' del session['user2'] return "内容" if __name__ == '__main__': app.run() session.py #!/usr/bin/env python # -*- coding:utf-8 -*- import uuid import json from flask.sessions import SessionInterface from flask.sessions import SessionMixin from itsdangerous import Signer, BadSignature, want_bytes class MySession(dict, SessionMixin): def __init__(self, initial=None, sid=None): self.sid = sid self.initial = initial super(MySession, self).__init__(initial or ()) def __setitem__(self, key, value): super(MySession, self).__setitem__(key, value) def __getitem__(self, item): return super(MySession, self).__getitem__(item) def __delitem__(self, key): super(MySession, self).__delitem__(key) class MySessionInterface(SessionInterface): session_class = MySession container = {} def __init__(self): import redis self.redis = redis.Redis() def _generate_sid(self): return str(uuid.uuid4()) def _get_signer(self, app): if not app.secret_key: return None return Signer(app.secret_key, salt='flask-session', key_derivation='hmac') def open_session(self, app, request): """ 程序刚启动时执行，需要返回一个session对象 """ sid = request.cookies.get(app.session_cookie_name) if not sid: # 创建一个随机字符串 sid = self._generate_sid() # 创建一个特殊的字典（sid,data）,并返回 return self.session_class(sid=sid) signer = self._get_signer(app) try: sid_as_bytes = signer.unsign(sid) sid = sid_as_bytes.decode() except BadSignature: sid = self._generate_sid() return self.session_class(sid=sid) # session保存在redis中 # val = self.redis.get(sid) # session保存在内存中 val = self.container.get(sid) if val is not None: try: data = json.loads(val) return self.session_class(data, sid=sid) except: return self.session_class(sid=sid) return self.session_class(sid=sid) def save_session(self, app, session, response): """ 程序结束前执行，可以保存session中所有的值 如： 保存到resit 写入到用户cookie """ domain = self.get_cookie_domain(app) path = self.get_cookie_path(app) = self.get_cookie_ secure = self.get_cookie_secure(app) expires = self.get_expiration_time(app, session) val = json.dumps(dict(session)) # session保存在redis中 # self.redis.setex(name=session.sid, value=val, time=app.permanent_session_lifetime) # session保存在内存中 self.container.setdefault(session.sid, val) session_id = self._get_signer(app).sign(want_bytes(session.sid)) response.set_cookie(app.session_cookie_name, session_id, expires=expires, domain=domain, path=path, secure=secure)

自定义Session

第三方session

#!/usr/bin/env python# -*- coding:utf-8 -*-"""pip3 install redispip3 install flask-session"""from flask import Flask, session, redirectfrom flask.ext.session import Sessionapp = Flask(__name__)app.debug = Trueapp.secret_key = 'asdfasdfasd'app.config['SESSION_TYPE'] = 'redis'from redis import Redisapp.config['SESSION_REDIS'] = Redis(host='192.168.0.94',port='6379')Session(app)@app.route('/login')def login(): session['username'] = 'alex' return redirect('/index')@app.route('/index')def index(): name = session['username'] return nameif __name__ == '__main__': app.run()

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。