路由和视图
这一波主要是通过看源码加深对 Flask 中路由和视图的了解，可以先回顾一下装饰器的知识：【装饰器函数与进阶】
路由设置的两种方式
# 示例代码
from flask import Flask app = Flask(__name__) @app.route('/index')
def index():
return 'index' if __name__ == '__main__':
app.run()
直接看上面代码，在 index 方法上通过装饰器 @app.route('/index') 就建立路由 '/index' 和方法 index 的对应关系。
查看 app.route 的源码：
def route(self, rule, **options):
def decorator(f):
endpoint = options.pop('endpoint', None)
self.add_url_rule(rule, endpoint, f, **options)
return f return decorator
flask.app.Flask.route
在上述示例中， rule 就是我们自定义的路由参数 '/index' ； endpoint 就是终结点参数（用来反向生成 url），这里我们没传；而 f 实际上就是该装饰器所装饰的函数，在这里也就是 index 函数。其实到这里就可以断定，该装饰器实际上就是通过第 4 行的 add_url_rule 函数来建立路由和视图的对应关系。我们可以测试：
from flask import Flask app = Flask(__name__) def index():
return 'index' app.add_url_rule('/index', view_func=index) if __name__ == '__main__':
app.run()
在这里我去掉了 index 函数的装饰器，而直接通过 app.add_url_rule 函数建立路由 '/index' 和 index 函数的对应关系，正常访问。
endpoint
接着看 app.add_url_rule 函数做了什么：
@setupmethod
def add_url_rule(self, rule, endpoint=None, view_func=None, provide_automatic_options=None, **options):
if endpoint is None:
endpoint = _endpoint_from_view_func(view_func)
options['endpoint'] = endpoint
methods = options.pop('methods', None) if methods is None:
methods = getattr(view_func, 'methods', None) or ('GET',)
if isinstance(methods, string_types):
raise TypeError('Allowed methods have to be iterables of strings, '
'for example: @app.route(..., methods=["POST"])')
methods = set(item.upper() for item in methods) required_methods = set(getattr(view_func, 'required_methods', ())) if provide_automatic_options is None:
provide_automatic_options = getattr(view_func,
'provide_automatic_options', None) if provide_automatic_options is None:
if 'OPTIONS' not in methods:
provide_automatic_options = True
required_methods.add('OPTIONS')
else:
provide_automatic_options = False methods |= required_methods rule = self.url_rule_class(rule, methods=methods, **options)
rule.provide_automatic_options = provide_automatic_options self.url_map.add(rule)
if view_func is not None:
old_func = self.view_functions.get(endpoint)
if old_func is not None and old_func != view_func:
raise AssertionError('View function mapping is overwriting an '
'existing endpoint function: %s' % endpoint)
self.view_functions[endpoint] = view_func
flask.app.Flask.add_url_rule
看 3、4、5 行，在示例中我们并没有传入 endpoint 参数，所以 endpoint 在第 3 行肯定是 None。接着执行第 4 行，查看 _endpoint_from_view_func 方法：
def _endpoint_from_view_func(view_func):
assert view_func is not None, 'expected view func if endpoint ' \
'is not provided.'
return view_func.__name__
flask.helpers._endpoint_from_view_func
看第 4 行的返回值是视图函数的函数名称，所以当不传 endpoint 参数时， endpoint 的值就是视图函数的函数名称。
继续看 flask.app.Flask.add_url_rule 函数的 34-39 行， 39 行做的就是每次装饰器执行时，就会把装饰器当前装饰的函数当做值， endpoint 当做 key ，放入 view_functions 这个字典中。而从 35-38 行可以看到，如果一个新的视图函数的 endpoint 已经存在 view_functions 中，但这个函数又与 endpoint 之前对应的视图函数不是同一个函数，就会产生 37 行错误。所以我们要保证每个视图函数对应的 endpoint 不重复。
app.route的参数
除了我们已经熟悉的 rule 和 view_func ，它还可传如下参数：
-
defaults
默认值，当URL中无参数，但函数需要参数时，可以使用 defaults={'k':'v'} 为函数提供参数。
-
endpoint
名称，用于反向生成 URL，即： url_for('endpoint') 。
-
methods
允许的请求方式，如： methods=["GET","POST"] 。
-
strict_slashes
对 URL 最后的 '/' 符号是否严格要求。
@app.route('/index',strict_slashes=False) # 访问 http://www.xx.com/index/ 或 http://www.xx.com/index均可
@app.route('/index',strict_slashes=True) # 仅访问 http://www.xx.com/index -
redirect
重定向到指定地址。
@app.route('/index/<int:nid>', redirect_to='/home/<nid>')
def index():
return 'index' -
subdomain
子域名访问。
from flask import Flask app = Flask(import_name=__name__)
app.config['SERVER_NAME'] = 'zze.com:5000' @app.route("/", subdomain="admin") # admin.zze.com:5000
def admin_index():
return "admin" @app.route("/", subdomain="guest") # guest.zze.com:5000
def guest_index():
return "guest" @app.route("/dynamic", subdomain="<username>") # http://test.zze.com:5000/dynamic
def dynamic_index(username):
return username if __name__ == '__main__':
app.run()
CBV
from flask import Flask, views app = Flask(__name__) class TestView(views.MethodView):
methods = ['GET'] # 只支持 GET 请求
decorators = [] # 批量加上装饰器 def get(self, *args, **kwargs):
return 'GET' def post(self, *args, **kwargs):
return 'POST' app.add_url_rule('/test', None, TestView.as_view('test')) # as_view 的参数就是 endpoint if __name__ == '__main__':
app.run()
它的实现其实和 Django 中的 CBV 实现很相似，源码就不细说了。
正则匹配URL
from flask import Flask, views, url_for
from werkzeug.routing import BaseConverter app = Flask(import_name=__name__) # 自定制类
class RegexConverter(BaseConverter):
"""
自定义URL匹配正则表达式
""" def __init__(self, map, regex):
super(RegexConverter, self).__init__(map)
self.regex = regex def to_python(self, value):
"""
路由匹配时，匹配成功后传递给视图函数中参数的值
:param value:
:return:
"""
return int(value) def to_url(self, value):
"""
使用url_for反向生成URL时，传递的参数经过该方法处理，返回的值用于生成URL中的参数
:param value:
:return:
"""
val = super(RegexConverter, self).to_url(value)
return val # 注册到 flask 的转换器中
app.url_map.converters['regex'] = RegexConverter # 使用
@app.route('/index/<regex("\d+"):nid>')
def index(nid):
print(url_for('index', nid=''))
return 'Index' if __name__ == '__main__':
app.run()
Session
源码
首先我们要知道 Flask 初执行是会经过 flask.app.Flask.__call__ 方法的，可以参考【Flask 的入口】。
def __call__(self, environ, start_response):
# environ ：请求相关所有数据
# start_response ：用于设置响应相关数据
return self.wsgi_app(environ, start_response)
再查看 wsgi_app 方法：
def wsgi_app(self, environ, start_response):
'''
获取environ并对其进行封装
从environ中获取名为session的cookie，解密并反序列化
放入请求上下文
'''
ctx = self.request_context(environ)
error = None
try:
try:
ctx.push()
'''
执行视图函数
'''
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
'''
获取session，解密并序列化，写入cookie
清空请求上下文
'''
ctx.auto_pop(error)
flask.app.Flask.wsgi_app
environ 是请求相关信息，第 7 行将 environ 传入 request_context 方法，看一下：
def request_context(self, environ):
return RequestContext(self, environ)
flask.app.Flask.request_context
可以看到它的返回值就是以 environ 为构造参数传入 RequestContext 类中的一个实例，看一下它初始化时做了什么：
def __init__(self, app, environ, request=None):
self.app = app
if request is None:
request = app.request_class(environ)
self.request = request
self.url_adapter = app.create_url_adapter(self.request)
self.flashes = None
self.session = None self._implicit_app_ctx_stack = [] self.preserved = False self._preserved_exc = None self._after_request_functions = [] self.match_request()
flask.ctx.RequestContext.__init__
看 3-8 行， environ 传入 request_class 方法中返回一个 request 实例，赋值给 self ，并在第 8 行给 self 新增一个 session 属性并赋值为 None 。而最终这个 self 在 flask.app.Flask.wsgi_app 的第 7 行赋值给 ctx 。总结一下就是在执行完 flask.app.Flask.wsgi_app 的第 7 行后， ctx 被赋值为 RequestContext 的一个实例，且这个实例中存在了将 environ 再次封装的属性 request 和一个为 None 的属性 session 。
接着看到 flask.app.Flask.wsgi_app 中的第 11 行，查看 push 方法：
def push(self):
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop(top._preserved_exc) 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) 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)
flask.ctx.RequestContext.push
直接看 19-26 行。如果 session 为空，就将传入 app 和 request 参数执行 session_interface 的 open_session 方法的返回值赋给 session ，而此时这个 session_interface 默认就是 flask.sessions.SecureCookieSessionInterface 类的实例，查看其 open_session 方法：
def open_session(self, app, request):
s = self.get_signing_serializer(app)
if s is None:
return None
val = request.cookies.get(app.session_cookie_name)
if not val:
return self.session_class()
max_age = total_seconds(app.permanent_session_lifetime)
try:
data = s.loads(val, max_age=max_age)
return self.session_class(data)
except BadSignature:
return self.session_class()
flask.sessions.SecureCookieSessionInterface.open_session
看第 5 行是从 cookie 中取一个键为 app.session_cookie_name 的值，而这个键的默认值就是 'session' ，可在配置文件中配置（点击查看配置文件默认配置参数）。紧接着就将这个值反序列化传入 session_class 并返回 session_class 的实例，而 session_class 对应的是类 flask.sessions.SecureCookieSession 。所以在上面 flask.ctx.RequestContext.push 方法中 21 行 self.session 的值就是 flask.sessions.SecureCookieSession 的实例。也就是在上面 flask.app.Flask.wsgi_app 的 11 行执行之后， ctx 的 session 就有值了。
接着看 flask.app.Flask.wsgi_app 的第 行，这行就是通过 full_dispatch_request 方法来完成执行视图函数和部分请求的收尾操作：
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)
return self.finalize_request(rv)
flask.app.Flask.full_dispatch_request
看第 10 行，在完成了上面视图函数相关操作后，通过 finalize_request 方法完成请求收尾操作：
def finalize_request(self, rv, from_error_handler=False):
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
flask.app.Flask.finalize_request
再看到第 4 行的 process_response 方法，这个方法如其名，就是用来处理响应相关信息：
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):
self.session_interface.save_session(self, ctx.session, response)
return response
flask.app.process_response
直接看 11、12 行，当 session 不为空时，调用 session_interface.save_session 方法，而 session_interface 就是上面执行 open_session 方法的 flask.sessions.SecureCookieSessionInterface 类实例。
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 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
)
flask.sessions.SecureCookieSessionInterface.save_session
看 25-35 行，会发现最后又将 session 序列化，再次写入 cookie 。
得出结论：当请求刚到来时，flask 读取 cookie 中 session 对应的值，并将该值解密并反序列化成字典，放入内存以便视图函数使用；当请求结束时，flask 会读取内存中字典的值，进行序列化和加密，再写入到 cookie 中。
第三方session
-
使用
from flask import Flask, request, session, redirect
from flask.sessions import SecureCookieSessionInterface
from flask_session import Session
from redis import Redis app = Flask(__name__)
app.debug = True app.config['SESSION_REDIS'] = Redis(host='127.0.0.1', port='', password='')
# 设置 session 类型
app.config['SESSION_TYPE'] = 'redis'
# 设置 根据 session 类型设置 app.session_interface
Session(app) @app.route('/login')
def login():
session['username'] = 'zze'
return 'success' app.run() -
源码
通过上面的源码部分已经知道了，flask 中的 session 存取就是通过 app.session_interface 来完成的，默认 app.session_interface = SecureCookieSessionInterface() ，而我们只要修改这一部分，让其存取是通过 redis 就 ok 了。查看 14 行 Session 类：
class Session(object):
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.copy()
config.setdefault('SESSION_TYPE', 'null')
config.setdefault('SESSION_PERMANENT', True)
config.setdefault('SESSION_USE_SIGNER', False)
config.setdefault('SESSION_KEY_PREFIX', 'session:')
config.setdefault('SESSION_REDIS', None)
config.setdefault('SESSION_MEMCACHED', None)
config.setdefault('SESSION_FILE_DIR',
os.path.join(os.getcwd(), 'flask_session'))
config.setdefault('SESSION_FILE_THRESHOLD', 500)
config.setdefault('SESSION_FILE_MODE', 384)
config.setdefault('SESSION_MONGODB', None)
config.setdefault('SESSION_MONGODB_DB', 'flask_session')
config.setdefault('SESSION_MONGODB_COLLECT', 'sessions')
config.setdefault('SESSION_SQLALCHEMY', None)
config.setdefault('SESSION_SQLALCHEMY_TABLE', 'sessions') if config['SESSION_TYPE'] == 'redis':
session_interface = RedisSessionInterface(
config['SESSION_REDIS'], config['SESSION_KEY_PREFIX'],
config['SESSION_USE_SIGNER'], config['SESSION_PERMANENT'])
elif config['SESSION_TYPE'] == 'memcached':
session_interface = MemcachedSessionInterface(
config['SESSION_MEMCACHED'], config['SESSION_KEY_PREFIX'],
config['SESSION_USE_SIGNER'], config['SESSION_PERMANENT'])
elif config['SESSION_TYPE'] == 'filesystem':
session_interface = FileSystemSessionInterface(
config['SESSION_FILE_DIR'], config['SESSION_FILE_THRESHOLD'],
config['SESSION_FILE_MODE'], config['SESSION_KEY_PREFIX'],
config['SESSION_USE_SIGNER'], config['SESSION_PERMANENT'])
elif config['SESSION_TYPE'] == 'mongodb':
session_interface = MongoDBSessionInterface(
config['SESSION_MONGODB'], config['SESSION_MONGODB_DB'],
config['SESSION_MONGODB_COLLECT'],
config['SESSION_KEY_PREFIX'], config['SESSION_USE_SIGNER'],
config['SESSION_PERMANENT'])
elif config['SESSION_TYPE'] == 'sqlalchemy':
session_interface = SqlAlchemySessionInterface(
app, config['SESSION_SQLALCHEMY'],
config['SESSION_SQLALCHEMY_TABLE'],
config['SESSION_KEY_PREFIX'], config['SESSION_USE_SIGNER'],
config['SESSION_PERMANENT'])
else:
session_interface = NullSessionInterface() return session_interfaceflask_session.Session
执行到第 8 行，可以看到它就是给 app.session_interface 赋值为 self._get_interface(app) ，而这个方法的返回值是根据在上面使用中第 12 行配置的 'SESSION_TYPE' 字段决定的，这里设置的是 'redis' ，所以 self._get_interface(app) 的返回值就为第 82 行的 RedisSessionInterface 实例。