請求進來後：
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-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 '''
<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' 基本使用
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)
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
#!/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()