flask基础知识深入——会话管理：Flask Session从原生到扩展源码分析及使用
会话管理：Flask Session从原生到扩展源码分析及使用目录
[*]会话管理：Flask Session从原生到扩展源码分析及使用
[*]一、Flask 原生Session机制之会话的创建与恢复源码分析
[*]二、原生Session机制之会话的保存与延长会话有效期源码分析及依赖配置
[*]三、flask-session扩展使用
一、Flask 原生Session机制之会话的创建与恢复源码分析
[*]会话管理简述：Flask会话管理的核心是Session机制，它将用户状态信息加密签名后通过Cookie存储在客户端。即：基于Cookie的客户端加密存储
[*]签名的实现：依赖SECRET_KEY完成签名防止篡改
[*]过程描述：当请求到来，会从cookie中获取session会话信息，如果没有会话信息则新创建一个会话，如果有的话就将这个加密签名的session进行签名验证然后进行解密和反序列化还原出Session对象。请求结束时将session转换为一个字典进行序列化进行签名加密，最终将处理后的会话数据通过 Cookie 保存在客户端
[*]源码分析：
[*]请求到来
def wsgi_app(
self, environ: WSGIEnvironment, start_response: StartResponse
) -> cabc.Iterable:
"""
1.1 ctx = self.request_context(environ)
创建请求上下文，且将请求环境数据environ传入
即 ctx =RequestContext(environ)
通过1.2/1.3处理 ctx有了request/session等属性
"""
ctx = self.request_context(environ)
try:
try:
"""
2.ctx.push()
使用RequestContext类的push()方法
通过2.1/2.2的系列操作完成了将当前上下文设置为活跃的上下文和将ctx中的session最终设置为
session_class()即SecureCookieSession类的实例对象，使其不在为None而是像字典一样进行会话数据的设置操作
"""
ctx.push()
"""
进行全局钩子执行，路由匹配，视图函数执行，保存Session会话信息等操作
"""
response = self.full_dispatch_request()
...
ctx.pop(error)
[*]RequestContext类
class RequestContext:
def __init__(
self,
app: Flask,
environ: WSGIEnvironment,
request: Request | None = None,
session: SessionMixin | None = None,
) -> None:
self.app = app
"""
1.2 通过1.1操作可见中未传入request则默认request = None,执行request = app.request_class(environ)
调用app.request_class()方法并将environ传入
即 request = Request(environ)
相当于实例化Request类将WSGI服务器中给WSGI应用程序的environ封装到RequestContext类实例对象ctx的 request属性中
则ctx的request属性中有了请求相关的数据
"""
if request is None:
request = app.request_class(environ)
self.request: Request = request
"""
1.3 同理可见实例化时未设置session的值，此时session为None，则ctx有了session属性只不过现在为None
"""
self.session: SessionMixin | None = session
...
def push(self) -> None:
"""
2.1 设置当前上下文为活跃的上下文操作
"""
...
"""
2.2 上述省略操作完成后：
if self.session is None
通过1.3操作 可见session为None
session_interface = self.app.session_interface
让session_interface等于app的session_interface属性，其属性为一个类的实例对象
即session_interface 为 SecureCookieSessionInterface类的实例对象
self.session = session_interface.open_session(self.app, self.request)
此时给session进行从Cookie中获取相关信息
"""
if self.session is None:
session_interface = self.app.session_interface
"""
通过2.2.2操作 根据cookie中是否会话信息有来进行判断重新创建一个会话还是用之前的会话，
若之前cookie中有会话信息，2.2.2操作可见序列化器s对之前序列化的session数据进行了反序列化，以保证数据可以被使 用
但无论什么情况session最终是SecureCookieSession类的实例对象，而这个类最终通过层层继承，继承了字典使session
可以像字典一样进行使用
所以不论cookie中是否有会话信息来判断是否创建新会话还是用之前的会话，2.2.2操作最终保证了session能像字典一样
进行数据的操作设置。而且session是SecureCookieSession类的实例对象
"""
self.session = session_interface.open_session(self.app, self.request)
"""
由2.2.1决定是否进入，如果进入则表示2.2.1操作时
调用session_interface即SecureCookieSessionInterface类的get_signing_serializer()方法获取s时检查到
app.secret_key没有进行设置，导致session为None，从而进入下面的if条件快
然后继续调用session_interface的make_null_session，实际上是SecureCookieSessionInterface类继承了
SessionInterface类的make_null_session方法，而这个方法返回一个null_session_class()对象，而
null_session_class = NullSession为NullSession类
所以如果检查到没有配置secret_key则，此时session为NullSession类的实例对象（会话不可用，但允许只读空会话，可 是对会话进行设置则报错）
而这个对象，一旦对session进行操作则进行报错
"""
if self.session is None:
self.session = session_interface.make_null_session(self.app)
[*]SecureCookieSessionInterface类
class SecureCookieSessionInterface(SessionInterface):
salt = "cookie-session"
digest_method = staticmethod(_lazy_sha1)
key_derivation = "hmac"
serializer = session_json_serializer
session_class = SecureCookieSession
def get_signing_serializer(self, app: Flask) -> URLSafeTimedSerializer | None:
if not app.secret_key:
"""
和open_session方法中
s = self.get_signing_serializer(app)
if s is None:
return None
配合
"""
return None
keys: list = []
if fallbacks := app.config["SECRET_KEY_FALLBACKS"]:
keys.extend(fallbacks)
keys.append(app.secret_key)
return URLSafeTimedSerializer(
keys,
salt=self.salt,
serializer=self.serializer,
signer_kwargs={
"key_derivation": self.key_derivation,
"digest_method": self.digest_method,
},
)
def open_session(self, app: Flask, request: Request) -> SecureCookieSession | None:
"""
2.2.1 通过s = self.get_signing_serializer(app)判断是否要进入2.2操作中的分支，即:
if self.session is None:
self.session = session_interface.make_null_session(self.app)
即此操作是检查序列化器s能否被正常使用，即检查有没有配置secret_key，以保证后续操作可以正常进行
"""
s = self.get_signing_serializer(app)
if s is None:
return None
"""
2.2.2
如果上面操作中if分支为进入,即表示调用get_signing_serializer()方法得到了URLSafeTimedSerializer
类的实例对象，即s
val = request.cookies.get(self.get_cookie_name(app))
表示通过1.2操作中完成了的request封装了请求相关的数据中获取cookie的信息从中查找会话数据
if not val:
return self.session_class()
表示如果从cookie中获取的会话数据为空则返回一个新会话给2.2中的session，即表示为第一次访问
则2.2中session为session_class()，而session_class = SecureCookieSession即表示session为
SecureCookieSession类的实例对象
而这个类继承CallbackDict类，然而这个个类又继承字典，则表示session可以像字典一样使用来设置信息
"""
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())
"""
data = s.loads(val, max_age=max_age)
如果从获取到了会话信息，即表示不是首次访问，则使用序列化器s的loads方法将会话信息进行反序列化，即将之前保存的序 列化后的session的数据进行反序列化，并且验证会话是否过期，成功然后则实例化一个SecureCookieSession对象且将
反序列化后的数据赋值给其属性，最后将这个SecureCookieSession实例化对象返回给到2.2中的session
失败则创建一个新会话
"""
try:
data = s.loads(val, max_age=max_age)
return self.session_class(data)
except BadSignature:
return self.session_class()
def save_session(
self, app: Flask, session: SessionMixin, response: Response
) -> None:
name = self.get_cookie_name(app)
domain = self.get_cookie_domain(app)
path = self.get_cookie_path(app)
secure = self.get_cookie_secure(app)
partitioned = self.get_cookie_partitioned(app)
samesite = self.get_cookie_samesite(app)
httponly = self.get_cookie_httponly(app)
if session.accessed:
response.vary.add("Cookie")
if not session:
if session.modified:
response.delete_cookie(
name,
domain=domain,
path=path,
secure=secure,
partitioned=partitioned,
samesite=samesite,
httponly=httponly,
)
response.vary.add("Cookie")
return
if not self.should_set_cookie(app, session):
"""
判断会话是否需要保存
"""
return
"""
expires = self.get_expiration_time(app, session)
延长会话的有效期
val = self.get_signing_serializer(app).dumps(dict(session))
获取序列化器并且将session转换为真正的字典进行序列化
response.set_cookie
将response中的cookie设置会话信息，会话有效期等一些相关信息
"""
expires = self.get_expiration_time(app, session)
val = self.get_signing_serializer(app).dumps(dict(session))# type: ignore
response.set_cookie(
name,
val,
expires=expires,
httponly=httponly,
domain=domain,
path=path,
secure=secure,
partitioned=partitioned,
samesite=samesite,
)
response.vary.add("Cookie")
[*]NullSession类
#类用于在会话不可用时生成更漂亮的错误信息。仍然允许只读空会话，但设置失败。
class NullSession(SecureCookieSession):
"""
2.2.1 步骤成立即没有配置secret_key，则2.2操作步骤中session为None进入嵌套的if条件快
然后调用SecureCookieSessionInterface类继承的SessionInterface类的make_null_session()方法，返回一个
null_session_class()实例对象而这个对象属于 NullSession类
即：当没有配置secret_key时session为NullSession类的实例对象，而不是SecureCookieSession类的实例对象
即这个类表示不可用会话，没有配置secret_key时session就是这个类的实例话对象
"""
def _fail(self, *args: t.Any, **kwargs: t.Any) -> t.NoReturn:
raise RuntimeError(
"The session is unavailable because no secret "
"key was set.Set the secret_key on the "
"application to something unique and secret."
)
"""
由于这个也继承SecureCookieSession，然后层层导致这个个也可以像字典一样使用
而下面代码表示当这个类的实例化对象，像字典一样使用后则调用上面的函数进行报错
则表示这个NullSession类实例化对象session为只读会话，而当其进行数据操作时进行报错
"""
__setitem__ = __delitem__ = clear = pop = popitem = update = setdefault = _fail# noqa: B950
del _fail
二、原生Session机制之会话的保存与延长会话有效期源码分析及依赖配置
[*]预备知识：
[*]Vary头：
[*]是HTTP响应头中的重要字段，用于告知缓存服务器（如浏览器），同一URL的响应内容可能应请求头的不同而不同。确保不同请求能拿到相应的响应数据，避免数据的混乱
[*]Vary: Cookie头，告知缓存服务器，同一URL的响应内容会因请求头里面的Cookie不同而变化
[*]modified : 保存
[*]permanent：永久
[*]PERMANENT_SESSION_LIFETIME = timedelta(days=31)默认有效期为31天
[*]SESSION_REFRESH_EACH_REQUEST = True与会话能否保存和有效期刷新相关
[*]源码分析
[*]SecureCookieSession类
#如果session不是 不可用会话，即不是NullSession类实例化对象。则sessin即是该类的实例化对象
class SecureCookieSession(CallbackDict, SessionMixin):
#默认为Fasle，其用来决定会话数据是否需要被序列化并存储到Cookie中
modified = False
#默认为Fasle，其用来决定是否需要添加Vary: Cookie头以确保缓存安全
accessed = False
def __init__(
self,
initial: c.Mapping | c.Iterable] | None = None,
) -> None:
"""
**
如果会话字典（伪字典）中，包含嵌套字典或者其他可变类型，直接修改这些嵌套对象不会触发该方法
即：修改会话字典中嵌套的复杂数据结构（如字典、列表、自定义对象等）的内部内容，而不是直接替换整个嵌套对象
不会触发该方法
**
只读操作，只读取会话数据（（如session.get('key')或session['key']））也不会触发该方法，只会设置
accessed = True
"""
def on_update(self: te.Self) -> None:
self.modified = True
self.accessed = True
super().__init__(initial, on_update)
#下面表示session如果使用了这些方法进行操作时，即表示会话被访问
def __getitem__(self, key: str) -> t.Any:
self.accessed = True
return super().__getitem__(key)
def get(self, key: str, default: t.Any = None) -> t.Any:
self.accessed = True
return super().get(key, default)
def setdefault(self, key: str, default: t.Any = None) -> t.Any:
self.accessed = True
return super().setdefault(key, default)
[*]process_response类
def process_response(self, response: Response) -> Response:
#再此之前视图函数执行完成且返回了响应对象
ctx = request_ctx._get_current_object()
"""
执行所有after_request 钩子
"""
for func in ctx._after_request_functions:
response = self.ensure_sync(func)(response)
for name in chain(request.blueprints, (None,)):
if name in self.after_request_funcs:
for func in reversed(self.after_request_funcs):
response = self.ensure_sync(func)(response)
"""
钩子执行完后
1.
if not self.session_interface.is_null_session(ctx.session)
调用session_interface的is_null_session方法，实际是SecureCookieSessionInterface类
继承的SessionInterface类的is_null_session方法
即判断session会话是否不是只读会话，即判断sesion是不是NullSession类的实例对象
"""
if not self.session_interface.is_null_session(ctx.session):
"""
2.
如果session不是NullSession类的实例对象，即session不是只渎空会话
即其是SecureCookieSession类的实例对象，然后执行session_interface实例的save_session方法，
进行session的保存
"""
self.session_interface.save_session(self, ctx.session, response)
"""
通过上述操作将response对象中的cookie得到有最新的会话相关数据，
后续则将进行把响应发送给客户端
"""
return response
[*]SessionInterface类
class SessionInterface:
null_session_class = NullSession
pickle_based = False
def make_null_session(self, app: Flask) -> NullSession:
"""
创建或恢复会话时，检查到没有配置secret_key后，调用此方法将session设置为只读空会话，即NullSession的实例化对象
"""
return self.null_session_class()
def is_null_session(self, obj: object) -> bool:
"""
1.1
1. 中调用了该方法，其用来判断session是不是NullSession的实例化对象，最终决定1.中是否要进行if块内部代码是否
执行
"""
return isinstance(obj, self.null_session_class)
def get_expiration_time(self, app: Flask, session: SessionMixin) -> datetime | None:
#既然来到这个方法则表明重新设置cookie是肯定的，而有效期是否刷新则要看里面的if session.permanent是否成立
"""
if session.permanent
如果设置了永久会话的配置，则有效期要刷新
现在的时间+有效期时间
return datetime.now(timezone.utc) + app.permanent_session_lifetime
"""
if session.permanent:
return datetime.now(timezone.utc) + app.permanent_session_lifetime
"""
session.permanent = False 表示为临时会话，浏览器关闭则下次访问得进行验证。没有有效期一说
"""
return None
def should_set_cookie(self, app: Flask, session: SessionMixin) -> bool:
#决定是否重新设置Cookie以刷新有效期
"""
2.3.1
情况一：
如果session.modified为True了，不用看后面条件，表示会话相关数据进行了重新赋值，则2.3中不直接return
继续后面操作进行，要进行保存
情况二：
如果会话数据没有进行重新赋值，且开了永久会话和会话要进行刷新的配置，则2.3中不直接return
继续后面操作进行，要进行保存
"""
return session.modified or (
session.permanent and app.config["SESSION_REFRESH_EACH_REQUEST"]
)
[*]SecureCookieSessionInterface类
class SecureCookieSessionInterface(SessionInterface):
def save_session(
self, app: Flask, session: SessionMixin, response: Response
) -> None:
#获取Cookie配置
name = self.get_cookie_name(app)
domain = self.get_cookie_domain(app)
path = self.get_cookie_path(app)
secure = self.get_cookie_secure(app)
partitioned = self.get_cookie_partitioned(app)
samesite = self.get_cookie_samesite(app)
httponly = self.get_cookie_httponly(app)
"""
2.1
如果会话被访问，添加vary头：vary：Cookie，告诉缓存服务器，响应内容会因请求头里面的cookie不同而变化
"""
if session.accessed:
response.vary.add("Cookie")
"""
2.2
if not session
if session.modified
如果会话数据为空，而且这个空会话是因为进行操作造成的，
表明这个空会话是将原有会话数据进行清空了，
response.delete_cookie()
将客户端中cookie中保存的数据进行删除，
因为原来session有数据，而此时没有数据，表示操作者不想要客户端原有的数据了且不想再存新数据
response.vary.add("Cookie")
添加vary头，告诉缓存服务器，响应内容会因请求头里面的cookie不同而变化，确保不同请求拿到相应的
数据
return
表明，如果会话数据为空，而且这个空会话不是因为进行操作造成的，直接return 不进行无意义的保存。
这针对于，之前创建的新会话而会话没有进行数据的设置，依然为空则不用保存在cookie中直接return
"""
if not session:
if session.modified:
response.delete_cookie(
name,
domain=domain,
path=path,
secure=secure,
partitioned=partitioned,
samesite=samesite,
httponly=httponly,
)
response.vary.add("Cookie")
return
"""
2.3
if not self.should_set_cookie(app, session)
判断是否重新设置Cookie以刷新有效期
如果2.3.1中是返回return，不论是情况一还是情况二成立均表明重新设置cookie是肯定的，
而是否要刷新有效期则看下面代码调用的get_expiration_time方法
"""
if not self.should_set_cookie(app, session):
return
"""
2.4
expires = self.get_expiration_time(app, session)
进行会话是否能够延长有效期判断并延长有效期计算
"""
expires = self.get_expiration_time(app, session)
"""
2.5
将session转换成真正的字典，用签名序列化器进行序列化
"""
val = self.get_signing_serializer(app).dumps(dict(session))
"""
2.6
将cookie中获取的配置、session中保存的数据、会话有效期时间等放到cookie中，
添加响应头vary字段设置为cookie，告诉缓存服务器，响应的内容随请求头中cookie的不同而不同，确保同一URL下的不同请求
能获取对应的响应，确保客户端的cookie能够得到最新的数据以便下一次的会话正常进行
"""
response.set_cookie(
name,
val,
expires=expires,
httponly=httponly,
domain=domain,
path=path,
secure=secure,
partitioned=partitioned,
samesite=samesite,
)
response.vary.add("Cookie")
[*]会话保存与有效期延期依赖配置
通过上面源码可知：
[*]会话是否要保存取决于SessionInterface类的should_set_cookie方法
[*]如果session.modified = True 则会话数据会进行保存
[*]由于session是SecuresCookieSession的实例化对象而其中类属性session.modified默认为False
[*]session.modified = True触发条件
[*]会话字典中嵌套的复杂数据结构（如字典、列表、自定义对象等）修改其复杂数据结构的内部内容，而不是直接替换整个嵌套对象不会触发SessionInterface类的on_update方法，即不会触发将其改为True
[*]只访问会话字典的数据时，也不会触上述该方法。只会将accessed改为True
[*]如果不为True则须看session.permanent = True和app.config["SESSION_REFRESH_EACH_REQUEST"] = True是否同时成立，一般源码中默认app.config["SESSION_REFRESH_EACH_REQUEST"] = True，而默认session.permanent = False
[*]会话有效长能否延期取决于会话是否要保存和SecureCookieSessionInterface类继承的SessionInterface类的get_expiration_time方法
[*]如果session.permanent = True则刷新会话有效期
[*]总结：
通过上述可知，会话有效期能否进行刷新首先取决于会话保存与否，再取决于session.permanent
[*]会话保存与否依赖（两种情况能够进行保存）：
[*]session.modified
[*]session.permanent and app.config["SESSION_REFRESH_EACH_REQUEST"]
[*]会话能否进行有效期刷新依赖（两种情况可进行会话有效期刷新）：
[*]session.permanent = True and session.modified = True
app.config["SESSION_REFRESH_EACH_REQUEST"]任意值
[*]session.permanent = True and app.config["SESSION_REFRESH_EACH_REQUEST"] = True
session.modified任意值
[*]即：session.permanent = True andsession.modified = Trueor app.config["SESSION_REFRESH_EACH_REQUEST"] = True
三、flask-session扩展使用
[*]用途：将原生保存在Cookie中的session，保存在redis或memcached等指定的地方
[*]使用分析：由上述源码可见，原生session保存在Cookie中一切源于:
session_interface = self.app.session_interface即：session_interface = SecureCookieSessionInterface
[*]使用方法（Redis）：与保存在Cookie不同，redis中默认session.permanent = True，其他大体一致
[*]源码内容：
class Session:
def __init__(self, app=None):
self.app = app
if app is not None:
self.init_app(app)
def init_app(self, app):
app.session_interface = self._get_interface(app)
def _get_interface(self, app):
config = app.config
SESSION_TYPE = config.get("SESSION_TYPE", Defaults.SESSION_TYPE)
...
# Redis settings
SESSION_REDIS = config.get("SESSION_REDIS", Defaults.SESSION_REDIS)
if SESSION_TYPE == "redis":
from .redis import RedisSessionInterface
session_interface = RedisSessionInterface(
**common_params,
client=SESSION_REDIS,
)
elif:
...
...
return session_interface
[*]方法：
from flask import Flask, session
from flask_session import Session
import redis
app = Flask(__name__)
# 1. 配置Redis连接
app.config['SESSION_TYPE'] = 'redis'# 指定Session存储类型为Redis
app.config['SESSION_REDIS'] = redis.Redis(
host='127.0.0.1',# Redis服务器地址
port=6379, # Redis端口
# password='xxx',# 若Redis有密码则添加
db=0 # 使用第0个数据库
)
# 2. 可选配置（按需加）
app.config['SECRET_KEY'] = 'your_secret_key_here'# 必须设置，用于Session加密
app.config['SESSION_PERMANENT'] = True# Session是否持久化（默认True，关闭则设为False）
app.config['PERMANENT_SESSION_LIFETIME'] = 3600# 持久化Session的有效期（秒，默认31天）
app.config['SESSION_USE_SIGNER'] = True# 是否对SessionID签名（防止篡改，建议开启）
# 3. 初始化Session
Session(app)
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！ 前排留名，哈哈哈
页:
[1]