Flask入门教程(二十七):Session Interface API——自定义Session存储后端

发布时间:2026/9/5 10:26:00
Flask入门教程(二十七):Session Interface API——自定义Session存储后端 1. SessionInterface抽象基类SessionInterface定义了Session存储后端需要实现的核心接口。属性/方法说明null_session_class当Session不可用时使用的类默认为NullSessionpickle_basedSession是否基于pickle序列化默认为Falseopen_session(app, request)从请求中读取Session数据返回SessionMixin实例或Nonesave_session(app, session, response)将Session数据保存到响应中make_null_session(app)创建空Sessionopen_session返回None时自动调用is_null_session(obj)判断是否为NullSession实例2. Cookie选项方法以下方法用于获取Session Cookie的各项属性可被子类重写方法说明get_cookie_name(app)获取Cookie名称默认读取SESSION_COOKIE_NAME配置get_cookie_domain(app)获取Cookie Domain默认读取SESSION_COOKIE_DOMAINget_cookie_path(app)获取Cookie Path默认读取SESSION_COOKIE_PATH或APPLICATION_ROOTget_cookie_httponly(app)获取HttpOnly标志默认读取SESSION_COOKIE_HTTPONLYget_cookie_secure(app)获取Secure标志默认读取SESSION_COOKIE_SECUREget_cookie_samesite(app)获取SameSite值默认读取SESSION_COOKIE_SAMESITEget_cookie_partitioned(app)获取Partitioned标志默认读取SESSION_COOKIE_PARTITIONEDv3.1get_expiration_time(app, session)获取Session过期时间永久Session返回now lifetimeshould_set_cookie(app, session)判断是否需要设置Cookie3. 内置实现类说明SecureCookieSessionInterface默认实现。使用itsdangerous签名Cookie存储Session数据SecureCookieSession基于CallbackDict的Session类检测顶层键的修改自动设置modifiedTrueNullSession未设置SECRET_KEY时使用的占位Session读取正常但修改时报错SessionMixinSession类的Mixin提供permanent、new、modified、accessed属性4. 替换默认Session实现自定义内存Session示例from flask import Flask from flask.sessions import SessionInterface, SessionMixin from uuid import uuid4 # 自定义Session类存储用户数据 class MySession(dict, SessionMixin): def __init__(self, session_idNone, dataNone): super().__init__() self.session_id session_id or str(uuid4()) if data: self.update(data) self.permanent False # 自定义Session接口内存存储 class MySessionInterface(SessionInterface): # 模拟数据库存储实际生产环境应使用Redis等 _store {} def open_session(self, app, request): # 从Cookie中获取session_id session_id request.cookies.get(my_session_id) if session_id and session_id in self._store: data self._store[session_id] return MySession(session_id, data) # 创建新Session return MySession() def save_session(self, app, session, response): if not session: return # 如果Session为空删除Cookie if not session: response.delete_cookie(my_session_id) return # 保存数据到存储 self._store[session.session_id] dict(session) # 设置Cookie response.set_cookie( my_session_id, session.session_id, max_ageapp.permanent_session_lifetime.total_seconds() if session.permanent else None, httponlyTrue, secureapp.config.get(SESSION_COOKIE_SECURE, False), samesiteapp.config.get(SESSION_COOKIE_SAMESITE, Lax) ) # 应用配置 app Flask(__name__) app.secret_key dev-secret app.session_interface MySessionInterface() app.route(/) def index(): session[count] session.get(count, 0) 1 return f访问次数: {session[count]}5. 使用Redis存储Session生产级示例import redis import pickle from flask import Flask from flask.sessions import SessionInterface, SessionMixin from uuid import uuid4 from datetime import timedelta class RedisSession(dict, SessionMixin): Redis存储的Session类 def __init__(self, session_idNone, dataNone, expiryNone): super().__init__() self.session_id session_id or str(uuid4()) self.permanent False self.expiry expiry if data: self.update(data) class RedisSessionInterface(SessionInterface): 使用Redis存储Session def __init__(self, redis_client, key_prefixsession:): self.redis redis_client self.key_prefix key_prefix def _get_key(self, session_id): return f{self.key_prefix}{session_id} def open_session(self, app, request): session_id request.cookies.get(app.config.get(SESSION_COOKIE_NAME, session)) if session_id: key self._get_key(session_id) data self.redis.get(key) if data: try: data_dict pickle.loads(data) return RedisSession(session_id, data_dict) except: pass return RedisSession() def save_session(self, app, session, response): if not session: return key self._get_key(session.session_id) # 如果Session为空删除Redis数据并清除Cookie if not session: self.redis.delete(key) response.delete_cookie(app.config[SESSION_COOKIE_NAME]) return # 计算过期时间 if session.permanent: lifetime app.config.get(PERMANENT_SESSION_LIFETIME, timedelta(days31)) else: lifetime timedelta(days1) # 序列化并保存到Redis self.redis.setex(key, int(lifetime.total_seconds()), pickle.dumps(dict(session))) # 设置Cookie response.set_cookie( app.config[SESSION_COOKIE_NAME], session.session_id, max_ageint(lifetime.total_seconds()) if session.permanent else None, httponlyapp.config.get(SESSION_COOKIE_HTTPONLY, True), secureapp.config.get(SESSION_COOKIE_SECURE, False), samesiteapp.config.get(SESSION_COOKIE_SAMESITE, Lax) ) # 使用 import redis r redis.Redis(hostlocalhost, port6379, db0) app Flask(__name__) app.secret_key dev-secret app.session_interface RedisSessionInterface(r) app.route(/) def index(): session[views] session.get(views, 0) 1 return f访问次数: {session[views]}6. 什么时候需要自定义SessionInterface场景说明多服务器/负载均衡默认签名Cookie在单服务器可用多服务器间需共享Session存储Session数据量较大Cookie有4KB限制外部存储无此限制需要服务端Session控制如强制下线、Session黑名单等与现有系统集成需对接已有的Session存储体系审计和安全要求需要完整的服务端Session日志7. 不同Session方案对比存储方式优点缺点适用场景签名Cookie默认无状态、无需服务端存储、部署简单4KB限制、数据可见、无法主动失效单机应用、轻量SessionRedis高性能、支持过期、多服务器共享需要额外服务、网络延迟分布式应用、高并发数据库持久化、支持复杂查询性能较低对持久性要求高的场景Memcached高性能不支持持久化缓存型Session8. SessionInterface API速查表类别方法说明核心open_session(app, request)从请求加载Session数据核心save_session(app, session, response)保存Session到响应Cookieget_cookie_name(app)Cookie名称Cookieget_cookie_secure(app)Secure标志Cookieget_cookie_httponly(app)HttpOnly标志Cookieget_cookie_samesite(app)SameSite值Cookieget_expiration_time(app, session)过期时间工具should_set_cookie(app, session)是否设置Cookie工具make_null_session(app)创建空Session小结本章全面讲解了Flask Session Interface的完整API。SessionInterface是Session存储的底层抽象通过open_session(app, request)从请求加载Session数据通过save_session(app, session, response)将Session保存到响应Cookie相关方法get_cookie_name、get_cookie_secure、get_cookie_httponly、get_cookie_samesite等可被子类重写以自定义Cookie属性Flask默认使用SecureCookieSessionInterface签名Cookie存储通过替换app.session_interface可切换为Redis、数据库等外部存储解决多服务器共享、数据量限制等问题。自定义SessionInterface是实现分布式Session管理的核心手段。