一:session源码开始
1,创建起始连接
from flask import Flask from index import app app=Flask(__name__) if __name__ == ‘__main__‘: app.__call__ app.run()
2,开始请求app.__call__
def __call__(self, environ, start_response): """Shortcut for :attr:`wsgi_app`.""" return self.wsgi_app(environ, start_response)
3,执行wsgi_app方法
def wsgi_app(self, environ, start_response): ctx = self.request_context(environ) ctx.push() error = None try: try: 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)
==> ctx = self.request_context(environ)
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) self.request = request
==> ctx.push()
def push(self): self.session = self.app.open_session(self.request) #开始运行session if self.session is None: #如果没有session self.session = self.app.make_null_session()
==> def open_session(self, app, request)
开始执行session
def open_session(self, app, request): sid = request.cookies.get(app.session_cookie_name) #如果能从cookie拿到session的话 if not sid: sid = self._generate_sid() return self.session_class(sid=sid, permanent=self.permanent) #如果没有session,则返回一个空字典 if self.use_signer: signer = self._get_signer(app) if signer is None: return None try: sid_as_bytes = signer.unsign(sid) sid = sid_as_bytes.decode() except BadSignature: sid = self._generate_sid() return self.session_class(sid=sid, permanent=self.permanent) if not PY2 and not isinstance(sid, text_type): sid = sid.decode(‘utf-8‘, ‘strict‘) val = self.redis.get(self.key_prefix + sid) if val is not None: try: data = self.serializer.loads(val) #加密val return self.session_class(data, sid=sid) except: return self.session_class(sid=sid, permanent=self.permanent) return self.session_class(sid=sid, permanent=self.permanent) #把session加到字典里
==> def full_dispatch_request(self):
当视图函数执行完成,要把session保存到session里面
self.try_trigger_before_first_request_functions() try: #视图函数开始,发送信号 print(self) 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) return self.finalize_request(rv)
==> def finalize_request(self, rv, from_error_handler=False):
上面是视图函数全部执行完成,开始保存数据,以及session,在process_responce里面存有所有数据
response = self.make_response(rv) try: 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
==> def process_response(self, response):
这里处理保存session,调用
save_session
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): self.save_session(ctx.session, response) return response
==> def save_session(self, app, session, response):
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 not self.should_set_cookie(app, session): return httponly = self.get_cookie_httponly(app) secure = self.get_cookie_secure(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, #最后保存到cookie里面 expires=expires, httponly=httponly, domain=domain, path=path, secure=secure)
