流程图如下图所示:
调用相关类如下图所示:
相关代码如下:
from flask import Flask, session
app = Flask(__name__)
'''
1. 加密会话数据:在 Flask 中,会话数据存储在客户端的 cookie 中。设置 app.secret_key 可以加密会话 cookie,确保数据的安全性。
2. 签名数据:通过设置 app.secret_key,Flask 可以对客户端发送的数据进行签名,用于验证数据的完整性和来源。
'''
app.secret_key = "1111111"
@app.route("/index1")
def index():
# 调用session的__setitem__方法
# 去ctx中获取session(特殊的字典)
session["k1"] = 123
return "Index"
@app.route("/index2")
def index2():
# 调用session的__getitem__方法
print(session["k1"])
return "Index2"
if __name__ == "__main__":
app.__call__()
app.run()
1、flask实例化之后可以执行的__ call __方法
# self为实例化的app对象
def __call__(
self, environ: WSGIEnvironment, start_response: StartResponse
) -> cabc.Iterable[bytes]:
# 调用wsgi_app方法
return self.wsgi_app(environ, start_response)
2、调用wsgi_app方法:
def wsgi_app(
self, environ: WSGIEnvironment, start_response: StartResponse
) -> cabc.Iterable[bytes]:
# 调用request_context方法,返回ctx对象
'''
request_context方法中返回一个类对象,RequestContext
ctx = RequestContext(self, environ)
结合下面的流程最终可以得到ctx中有request(environ),session;将两者打包到一块
'''
ctx = self.request_context(environ)
error: BaseException | None = None
try:
try:
'''
对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)
'''
ctx.push()
'''
执行视图函数
'''
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
if "werkzeug.debug.preserve_context" in environ:
environ["werkzeug.debug.preserve_context"](_cv_app.get())
environ["werkzeug.debug.preserve_context"](_cv_request.get())
if error is not None and self.should_ignore_error(error):
error = None
ctx.pop(error)
2.1、request_context方法源码如下:
def request_context(self, environ: WSGIEnvironment) -> RequestContext:
# 返回RequestContext对象 self 表示实例化的Flask对象app,environ表示wsgi环境
return RequestContext(self, environ)
2.1.1、RequestContext类中进行的操作:
def __init__(
self,
app: Flask,
environ: WSGIEnvironment,
request: Request | None = None,
session: SessionMixin | None = None,
) -> None:
self.app = app
'''
判断如果request是空的话,将其赋值为app.request_class(environ)
即: request_class: type[Request] = Request
request=Request
'''
if request is None:
request = app.request_class(environ)
request.json_module = app.json
self.request: Request = request
self.url_adapter = None
try:
self.url_adapter = app.create_url_adapter(self.request)
except HTTPException as e:
self.request.routing_exception = e
self.flashes: list[tuple[str, str]] | None = None
'''
设置session值
'''
self.session: SessionMixin | None = session
self._after_request_functions: list[ft.AfterRequestCallable[t.Any]] = []
......
2.2、调用push方法,对session进行赋值:
def push(self) -> None:
app_ctx = _cv_app.get(None)
if app_ctx is None or app_ctx.app is not self.app:
app_ctx = self.app.app_context()
app_ctx.push()
else:
app_ctx = None
self._cv_tokens.append((_cv_request.set(self), app_ctx))
'''
对session进行赋值
'''
if self.session is None:
# session_interface: SessionInterface = SecureCookieSessionInterface()
session_interface = self.app.session_interface
# SecureCookieSessionInterface.open_session(self.app, self.request)
self.session = session_interface.open_session(self.app, self.request)
if self.session is None:
self.session = session_interface.make_null_session(self.app)
if self.url_adapter is not None:
self.match_request()
2.2.1、SecureCookieSessionInterface.open_session(self.app, self.request)方法:由此所知,session是一个特殊的字典。
def open_session(self, app: Flask, request: Request) -> SecureCookieSession | None:
s = self.get_signing_serializer(app)
if s is None:
return None
'''
如果拿不到值就返回session_class()
session_class() :class SecureCookieSession(CallbackDict, SessionMixin)
继承class CallbackDict(UpdateDictMixin, dict):
CallbackDict继承dict(字典)
'''
val = request.cookies.get(self.get_cookie_name(app))
if not val:
return self.session_class()
max_age = int(app.permanent_session_lifetime.total_seconds())
try:
data = s.loads(val, max_age=max_age)
return self.session_class(data)
except BadSignature:
return self.session_class()
初步学习,继续补充完善......