Flask框架（五） —— session源碼分析
目錄python
session源碼分析
一、請求來了，執行__call__方法
# 請求來了執行 __call__方法
if __name__ == '__main__':
app.__call__
app.run()
二、__call__方法
def __call__(self, environ, start_response):
"""The WSGI server calls the Flask application object as the WSGI application. This calls :meth:`wsgi_app` which can be wrapped to applying middleware."""
return self.wsgi_app(environ, start_response)
三、調用__call__方法
def wsgi_app(self, environ, start_response):
# 1.獲得request，和空的session
# ctx是RequestContext的對象，對象裏面有空的session和request
ctx = self.request_context(environ)
error = None
try:
try:
# 2.從request中的cookie中取出value,解密過轉成session對象 --> open_session
ctx.push()
# 3.路由映射到函數，執行函數,而後保存session --> save_session，請求結束
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except:
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)
3.一、ctx = self.request_context(environ) --- 獲得空的session
# 1. 得到RequestContext對象 ---> ctx
def request_context(self, environ):
return RequestContext(self, environ)
class RequestContext(object):
def __init__(self, app, environ, request=None):
self.app = app
if request is None:
request = app.request_class(environ) # request_class = Request
self.request = request
self.url_adapter = app.create_url_adapter(self.request)
self.flashes = None
self.session = None
3.二、ctx.push() --- 調用open_session方法
# 2.ctx.push() --- 調用open_session方法
def push(self):
"""Binds the request context to the current context."""
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop(top._preserved_exc)
# Before we push the request context we have to ensure that there
# is an application context.
app_ctx = _app_ctx_stack.top
if app_ctx is None or app_ctx.app != self.app:
app_ctx = self.app.app_context()
app_ctx.push()
self._implicit_app_ctx_stack.append(app_ctx)
else:
self._implicit_app_ctx_stack.append(None)
if hasattr(sys, 'exc_clear'):
sys.exc_clear()
_request_ctx_stack.push(self)
# Open the session at the moment that the request context is available.
# This allows a custom open_session method to use the request context.
# Only open a new session if this is the first time the request was
# pushed, otherwise stream_with_context loses the session.
# 從request中的cookie中取出value,(有解密過程)轉成session對象
# 正常狀況下此時session中有值了
# 若是request中沒有cookie，那麼session中仍然沒有值
if self.session is None:
# session_interface = SecureCookieSessionInterface()
session_interface = self.app.session_interface
# 2.1 opensession(),SecureCookieSessionInterface類中
self.session = session_interface.open_session(
self.app, self.request
)
if self.session is None:
self.session = session_interface.make_null_session(self.app)
3.2.一、session_interface.open_session() --- 從經過session的名字取出cookie的值轉成session
# 2.1.session_interface.open_session()
class SecureCookieSessionInterface(SessionInterface):
def open_session(self, app, request):
s = self.get_signing_serializer(app)
if s is None:
return None
# 經過session的名字取出cookie的值，key:value形式，獲得的是json格式的字符串
val = request.cookies.get(app.session_cookie_name)
if not val:
return self.session_class()
max_age = total_seconds(app.permanent_session_lifetime)
try:
# 將json格式字符串轉成字典
data = s.loads(val, max_age=max_age)
# 返回一個特殊的字典
return self.session_class(data)
except BadSignature:
return self.session_class()
3.三、self.full_dispatch_request() --- 路由分發，執行函數，寫入session
# 3.self.full_dispatch_request()
1 執行before_request
2 執行視圖函數
3 把session寫入
if not self.session_interface.is_null_session(ctx.session):
self.session_interface.save_session(self, ctx.session, response)
def full_dispatch_request(self):
self.try_trigger_before_first_request_functions()
try:
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
# 路由分發，執行視圖函數
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
# 3.1 把session保存寫入cookie
return self.finalize_request(rv)
3.3.一、self.finalize_request(rv) --- 調用save_session()保存寫入session
# 3.1.self.finalize_request(rv)
def finalize_request(self, rv, from_error_handler=False):
response = self.make_response(rv)
try:
# 利用save_session 保存寫入session
response = self.process_response(response)
request_finished.send(self, response=response)
except Exception:
if not from_error_handler:
raise
self.logger.exception('Request finalizing failed with an '
'error while handling an error')
return response
3.3.1.一、self.process_response(response) --- 調用save_session方法
# 保存寫入session
def process_response(self, response):
ctx = _request_ctx_stack.top
bp = ctx.request.blueprint
funcs = ctx._after_request_functions
if bp is not None and bp in self.after_request_funcs:
funcs = chain(funcs, reversed(self.after_request_funcs[bp]))
if None in self.after_request_funcs:
funcs = chain(funcs, reversed(self.after_request_funcs[None]))
for handler in funcs:
response = handler(response)
if not self.session_interface.is_null_session(ctx.session):
# 調用save_session方法保存寫入session
self.session_interface.save_session(self, ctx.session, response)
return response
save_session()方法
class SecureCookieSessionInterface(SessionInterface):
def save_session(self, app, session, response):
domain = self.get_cookie_domain(app)
path = self.get_cookie_path(app)
# If the session is modified to be empty, remove the cookie.
# If the session is empty, return without setting the cookie.
if not session:
if session.modified:
response.delete_cookie(
app.session_cookie_name,
domain=domain,
path=path
)
return
# Add a "Vary: Cookie" header if the session was accessed at all.
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
)
博客內容僅供參考，部分參考他人優秀博文，僅供學習使用