Flask 各模块分析表
| 模块 / 包 | 说明 |
|---|---|
| json/ | 提供 JSON 支持 |
| init.py | 构造文件，导入了所有其他模块中开放的类和函数 |
| _compat.py | 定义 Python2 与 Python3 版本兼容代码 |
| app.py | 主脚本，实现了 WSGI 程序对象，包含 Flask 类 |
| blueprint.py | 蓝本支持，包含 Blueprint 类定义 |
| cli.py | 提供命令行支持，包含内置的几个命令 |
| config.py | 实现配置相关的对象 |
| ctx.py | 实现上下文对象，比如请求上下文 RequestContext |
| debughelpers.py | 一些辅助开发的函数 / 类 |
| globals.py | 定义全局对象，比如 request、session 等 |
| helpers.py | 包含一些常用的辅助函数，比如 flash()、url_for() |
| logging.py | 提供日志支持 |
| sessions.py | 实现 session 功能 |
| signals.py | 实现信号支持，定义了内置的信号 |
| templating.py | 模板渲染功能 |
| testing.py | 提供用于测试的辅助函数 |
| views.py | 提供了类似 Django 中的类视图，用于编写 Web API 的 MethodView |
| wrappers.py | 实现 WSGI 封装对象，比如代表请求和响应的 Request 对象和 Response 对象 |
Flask 的运行方式
一、运行流程
- 完成对
app对象的初始化，包括路由表的建立 [rule –> endpoint –> view_func] - 运行
app.run()，调用werkzeug的run_simple函数Start a WSGI application，因为调用app.run()无法满足生产环境的security and performance requirements，所以生产环境需要使用gunicorn等Python WSGI HTTP Server - 当有请求发生时，同时将
RequestContext和AppContext推入对应的栈
二、具体代码分析
代码：
from flask import Flask, jsonify
app = Flask(__name__)
@app.route('/')
def index():
return jsonify("test")
1. 初始化 app 对象
Code：
app = Flask(__name__)
Detail：
"""
flask/app.py
初始化 Flask 对象
"""
def __init__(self, import_name, static_path=None, static_url_path=None,
static_folder='static', template_folder='templates',
instance_path=None, instance_relative_config=False,
root_path=None):
_PackageBoundObject.__init__(self, import_name,
template_folder=template_folder,
root_path=root_path)
if static_path is not None:
from warnings import warn
warn(DeprecationWarning('static_path is now called '
'static_url_path'), stacklevel=2)
static_url_path = static_path
if static_url_path is not None:
self.static_url_path = static_url_path
if static_folder is not None:
self.static_folder = static_folder
if instance_path is None:
instance_path = self.auto_find_instance_path()
elif not os.path.isabs(instance_path):
raise ValueError('If an instance path is provided it must be '
'absolute. A relative path was given instead.')
self.instance_path = instance_path
self.config = self.make_config(instance_relative_config)
self._logger = None
self.logger_name = self.import_name
self.view_functions = {}
self._error_handlers = {}
self.error_handler_spec = {None: self._error_handlers}
self.url_build_error_handlers = []
self.before_request_funcs = {}
self.before_first_request_funcs = []
self.after_request_funcs = {}
self.teardown_request_funcs = {}
self.teardown_appcontext_funcs = []
self.url_value_preprocessors = {}
self.url_default_functions = {}
self.template_context_processors = {
None: [_default_template_ctx_processor]
}
self.shell_context_processors = []
self.blueprints = {}
self._blueprint_order = []
self.extensions = {}
self.url_map = Map()
self._got_first_request = False
self._before_request_lock = Lock()
if self.has_static_folder:
self.add_url_rule(self.static_url_path + '/<path:filename>',
endpoint='static',
view_func=self.send_static_file)
self.cli = cli.AppGroup(self.name)
2. 建立路由表
Code：
@app.route('/')
def index():
return jsonify("test")
Detail：
"""
flask/app.py
route 装饰器，仅完成注册功能，所以没有调用 from functools import wraps
"""
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.py
add_url_rule 函数，添加路由表信息
url_rule_class 绑定 url --> endpoint
view_functions 绑定 endpoint --> view_func
"""
@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
# ---------------------- RUN APPLICATION -----------------------------------------
"""
1. 如果使用命令行行的方式启动服务，调用 cli.py 中的 run_command 函数
flask/cli.py
"""
def run_command(info, host, port, reload, debugger, eager_loading, with_threads, cert):
debug = get_debug_flag()
if reload is None:
reload = debug
if debugger is None:
debugger = debug
if eager_loading is None:
eager_loading = not reload
show_server_banner(get_env(), debug, info.app_import_path, eager_loading)
app = DispatchingApp(info.load_app, use_eager_loading=eager_loading)
from werkzeug.serving import run_simple
run_simple(host, port, app, use_reloader=reload, use_debugger=debugger,
threaded=with_threads, ssl_context=cert)
"""
2. 如果使用 app.run() 的方式启动服务，调用 app.py 中的 run 函数
flask/app.py
"""
def run(self, host=None, port=None, debug=None, load_dotenv=True, **options):
if os.environ.get('FLASK_RUN_FROM_CLI') == 'true':
from .debughelpers import explain_ignored_app_run
explain_ignored_app_run()
return
if get_load_dotenv(load_dotenv):
cli.load_dotenv()
if 'FLASK_ENV' in os.environ:
self.env = get_env()
self.debug = get_debug_flag()
elif 'FLASK_DEBUG' in os.environ:
self.debug = get_debug_flag()
if debug is not None:
self.debug = bool(debug)
_host = '127.0.0.1'
_port = 5000
server_name = self.config.get('SERVER_NAME')
sn_host, sn_port = None, None
if server_name:
sn_host, _, sn_port = server_name.partition(':')
host = host or sn_host or _host
port = int(port or sn_port or _port)
options.setdefault('use_reloader', self.debug)
options.setdefault('use_debugger', self.debug)
options.setdefault('threaded', True)
cli.show_server_banner(self.env, self.debug, self.name, False)
from werkzeug.serving import run_simple
try:
run_simple(host, port, self, **options)
finally:
self._got_first_request = False
3. Request & Response
"""
flask/app.py
"""
# environ 是一个字典对象，包括从 wsgi 服务器返回的所有信息
# environ = <class 'dict'>: {'wsgi.version': (1, 0), 'wsgi.url_scheme': 'http', 'wsgi.input': <_io.BufferedReader name=7>, 'wsgi.errors': <_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'>, 'wsgi.multithread': True, 'wsgi.multiprocess': False, 'wsgi.run_once': False, 'werkzeug.server.shutdown': <function WSGIRequestHandler.make_environ.<locals>.shutdown_server at 0x10c407d08>, 'SERVER_SOFTWARE': 'Werkzeug/0.15.1', 'REQUEST_METHOD': 'GET', 'SCRIPT_NAME': '', 'PATH_INFO': '/', 'QUERY_STRING': '', 'REQUEST_URI': '/', 'RAW_URI': '/', 'REMOTE_ADDR': '127.0.0.1', 'REMOTE_PORT': 51127, 'SERVER_NAME': '127.0.0.1', 'SERVER_PORT': '5000', 'SERVER_PROTOCOL': 'HTTP/1.1', 'HTTP_HOST': '127.0.0.1:5000', 'HTTP_CONNECTION': 'keep-alive', 'HTTP_UPGRADE_INSECURE_REQUESTS': '1', 'HTTP_USER_AGENT': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.86 Safari/537.36', 'HTTP_ACCEPT': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3', 'HTTP_ACCEPT_ENCODING': 'gzip, deflate, br', 'HTTP_ACCEPT_LANGUAGE': 'zh-CN,zh;q=0.9,en;q=0.8', 'HTTP_COOKIE': '_ga=GA1.1.1219788582.1544602907', 'werkzeug.request': <BaseRequest 'http://127.0.0.1:5000/' [GET]>}
def __call__(self, environ, start_response):
return self.wsgi_app(environ, start_response)
def wsgi_app(self, environ, start_response):
ctx = self.request_context(environ)
error = None
try:
try:
# 查看栈顶是否有元素（请求上下文），如果有的话将其 pop 掉
# 查看应用上下文栈当中是否有应用上下文，如果其中没有应用上下文，或者有应用上下文但是不是当前应用的上下文，那么久将当前应用的上下文压栈
# 将当前这个请求上下文（self)压如请求上下文栈
# 同时开启一个 session
ctx.push()
# 1. 触发 before_first_request 钩子函数
# 2. 找到 url 对应的函数 dispatch_request 函数中完成
# 3. 执行函数，返回 response 对象
# 4. 触发 after_request 钩子函数
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)
Client & WSGI & APPLICATION 的关系图：
