请求进来后： app.__call__ app.wsgi_app wsgi_app----ctx.push()-----RequestContext.push()----上下文这里不多说了省略 ---这样一段代码就是session--- if self.session is None: session_interface = self.app.session_interface
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
httponly = self.get_cookie_httponly(app)
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,
httponly=httponly,
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-sessionFlask中的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组件使用和原理
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 '''
<form action="" method="post">
<p><input type=text name=username>
<p><input type=submit value=Login>
</form>
'''
@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'
基本使用
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)
httponly = self.get_cookie_httponly(app)
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, httponly=httponly,
domain=domain, path=path, secure=secure)
自定义Session
第三方session
#!/usr/bin/env python# -*- coding:utf-8 -*-
"""
pip3 install redis
pip3 install flask-session
"""
from flask import Flask, session, redirect
from flask.ext.session import Session
app = Flask(__name__)
app.debug = True
app.secret_key = 'asdfasdfasd'
app.config['SESSION_TYPE'] = 'redis'
from redis import Redis
app.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 name
if __name__ == '__main__':
app.run()