flask项目大名鼎鼎，应该不需要多做介绍了吧。我把它称之为python服务开发的TOP2项目，另外一个就是django了，不需要比较孰优孰劣，我的观点是各有千秋，各自应用于不同的场景，都需要深入理解，熟练掌握。本次源码选择的版本是 1.1.2，我会采用慢读法，尽自己最大努力把它讲透。本篇是开胃菜，主要分析flask的命令行工具的实现。
在正式开篇之前，和大家唠叨一点题外话，不喜欢废话的朋友可以直接跳过本段。网上，码农圈经常有各种对比讨论，最经典的梗是: “php是最好的语言！”。我没有嘲笑php的意思，我也用过一段时间的php，做个几个项目。对于这类问题， “小孩才做选择题，成年人都是我都要” 这是开玩笑了，我们重点应该是放在精通上，不应该关闭自己的视野。学习好flask，对精通django有帮助；精通了django，对flask也能够举一反三。如果把自己限定到某一个框架上，有些可惜，毕竟没有一个职业叫做Django/flask-web开发。精力允许的情况下，还是都掌握更好。
示例 hello.py 如下:
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello():
return 'Hello, World!'flask模块可以自启动，启动前需要设置FLASK_APP和FLASK_ENV两个环境变量，然后使用 run 命令:
$ export FLASK_APP=hello.py
$ export FLASK_ENV=development
$ python -m flask run控制台会显示启动成功信息:
* Serving Flask app "hello.py" (lazy loading)
* Environment: development
* Debug mode: on
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
* Restarting with stat
* Debugger is active!
* Debugger PIN: 986-558-971眼尖的朋友会发现，这和werkzeug的run_simple一样。没错，flask的cli就是使用werkzegu的run_simple实现的。
模块的启动入口是 __main__.py ，关于 __main__.py 的介绍可以查看附录的参考链接。在__main__.py主要是主要调用了 cli.py 的main函数:
# cli.py
cli = FlaskGroup(
help="""\
A general utility script for Flask applications.
Provides commands from Flask, extensions, and the application. Loads the
application defined in the FLASK_APP environment variable, or from a wsgi.py
file. Setting the FLASK_ENV environment variable to 'development' will enable
debug mode.
\b
{prefix}{cmd} FLASK_APP=hello.py
{prefix}{cmd} FLASK_ENV=development
{prefix}flask run
""".format(
cmd="export" if os.name == "posix" else "set",
prefix="$ " if os.name == "posix" else "> ",
)
)
def main(as_module=False):
# TODO omit sys.argv once https://github.com/pallets/click/issues/536 is fixed
cli.main(args=sys.argv[1:], prog_name="python -m flask" if as_module else None)cli工具的继承关系是: FlaskGroup->AppGroup->click.Group, 使用 click 模块构建的命令行工具。
Click 是一个 Python 包，用于以可组合的方式使用尽可能少的代码创建漂亮的命令行界面。它是“命令行界面创建工具包”，开箱即用同时高度可配置的。
基类AppGroup主要扩展了command方法，使用app_context来装饰所有命令:
# AppGroup
def with_appcontext(f):
@click.pass_context
def decorator(__ctx, *args, **kwargs):
with __ctx.ensure_object(ScriptInfo).load_app().app_context():
return __ctx.invoke(f, *args, **kwargs)
return update_wrapper(decorator, f)
class AppGroup(click.Group):
def command(self, *args, **kwargs):
wrap_for_ctx = kwargs.pop("with_appcontext", True)
def decorator(f):
if wrap_for_ctx:
f = with_appcontext(f)
return click.Group.command(self, *args, **kwargs)(f)
return decoratorapp_context是flask非常重要的特性，我们以后再详细分析，现在先记住所有的cmd都在app_context的上下文环境中。
FlaskGroup主要收集扩展命令，代码可以看到总共收集了3个命令:
class FlaskGroup(AppGroup):
def __init__(
self,
add_default_commands=True,
create_app=None,
add_version_option=True,
load_dotenv=True,
set_debug_flag=True,
**extra
):
...
AppGroup.__init__(self, params=params, **extra)
...
if add_default_commands:
self.add_command(run_command)
self.add_command(shell_command)
self.add_command(routes_command)
self._loaded_plugin_commands = False在FlaskGroup中，如果安装了 dotenv 可以使用 .env 文件来控制开发环境/正式环境, 比如数据库配置之类，会非常方便。
Python-dotenv 从.env文件中读取键值对，并可以将它们设置为环境变量。它有助于遵循12要素原则开发应用程序 。
FlaskGroup的main函数中会加载flask-app，主要使用ScriptInfo实现:
class ScriptInfo(object):
def __init__(self, app_import_path=None, create_app=None, set_debug_flag=True):
#: Optionally the import path for the Flask application.
self.app_import_path = app_import_path or os.environ.get("FLASK_APP")
...
def load_app(self):
...
if self.create_app is not None:
app = call_factory(self, self.create_app)
else:
if self.app_import_path:
path, name = (
re.split(r":(?![\\/])", self.app_import_path, 1) + [None]
)[:2]
import_name = prepare_import(path)
app = locate_app(self, import_name, name)
else:
for path in ("wsgi.py", "app.py"):
import_name = prepare_import(path)
app = locate_app(self, import_name, None, raise_if_not_found=False)
if app:
break
...
self._loaded_app = app
return app在ScriptInfo中会读取FLASK_APP的环境变量，然后会导入类文件，定位app, 然后加载app:
def locate_app(script_info, module_name, app_name, raise_if_not_found=True):
__traceback_hide__ = True # noqa: F841
try:
__import__(module_name)
except ImportError:
...
module = sys.modules[module_name]
if app_name is None:
return find_best_app(script_info, module)
else:
return find_app_by_string(script_info, module, app_name)__import__ 动态加载模块sys.modules[module_name] 获取加载后的模块app加载有find_best_app和find_app_by_string两种方法, 默认使用的是find_best_app方法。
由于app的script模块已经导入，优先在这个模块中查找名字为 app 或者 application 的变量，这是一种约定:
def find_best_app(script_info, module):
from . import Flask
# Search for the most common names first.
for attr_name in ("app", "application"):
app = getattr(module, attr_name, None)
if isinstance(app, Flask):
return app
...
# Otherwise find the only object that is a Flask instance.
matches = [v for v in itervalues(module.__dict__) if isinstance(v, Flask)]
if len(matches) == 1:
return matches[0]
...
for attr_name in ("create_app", "make_app"):
app_factory = getattr(module, attr_name, None)
if inspect.isfunction(app_factory):
try:
app = call_factory(script_info, app_factory)
if isinstance(app, Flask):
return app如果没有app或者application的变量定义，则转而寻找其它是flask对象的变量。
如果上述两种方法都找不到app变量，最后查找名字为create_app和make_app的工厂函数, 使用工厂函数创建flask-app。
比如flask示例的flaskr项目中，就提供的是create_app的工厂函数:
# flaskr/__init__.py
def create_app(test_config=None):
app = Flask(__name__, instance_relative_config=True)
....
return approutes命令最简单，使用方法如下:
$ flask routes --all-methods
Endpoint Methods Rule
-------- ------------------ -----------------------
hello GET, HEAD, OPTIONS /
static GET, HEAD, OPTIONS /static/<path:filename>routes_command函数实现routers指令:
@with_appcontext
def routes_command(sort, all_methods):
"""Show all registered routes with endpoints and methods."""
rules = list(current_app.url_map.iter_rules())
ignored_methods = set(() if all_methods else ("HEAD", "OPTIONS"))
if sort in ("endpoint", "rule"):
rules = sorted(rules, key=attrgetter(sort))
elif sort == "methods":
rules = sorted(rules, key=lambda rule: sorted(rule.methods))
rule_methods = [", ".join(sorted(rule.methods - ignored_methods)) for rule in rules]
headers = ("Endpoint", "Methods", "Rule")
widths = (
max(len(rule.endpoint) for rule in rules),
max(len(methods) for methods in rule_methods),
max(len(rule.rule) for rule in rules),
)
widths = [max(len(h), w) for h, w in zip(headers, widths)]
row = "{{0:<{0}}} {{1:<{1}}} {{2:<{2}}}".format(*widths)
click.echo(row.format(*headers).strip())
click.echo(row.format(*("-" * width for width in widths)))
for rule, methods in zip(rules, rule_methods):
click.echo(row.format(rule.endpoint, methods, rule.rule).rstrip())shell指令提供一个命令行方式运行app:
$ flask shell
Python 3.8.5 (v3.8.5:580fbb018f, Jul 20 2020, 12:11:27)
[Clang 6.0 (clang-600.0.57)] on darwin
App: ch21-flask.hello [development]
Instance: /Users/yoo/work/yuanmahui/python/instance
>>> app
<Flask 'ch21-flask.hello'>
>>> dir(app)
['__call__', '__class__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', ... ,'update_template_context', 'url_build_error_handlers', 'url_default_functions', 'url_defaults', 'url_map', 'url_map_class', 'url_rule_class', 'url_value_preprocessor', 'url_value_preprocessors', 'use_x_sendfile', 'view_functions', 'wsgi_app']
>>> app.url_map
Map([<Rule '/' (OPTIONS, HEAD, GET) -> hello>,
<Rule '/static/<filename>' (OPTIONS, HEAD, GET) -> static>])
>>> locals()
{'app': <Flask 'ch21-flask.hello'>, 'g': <flask.g of 'ch21-flask.hello'>,...}可以看到，在交互式终端中可以查看 app， url_map 。locals变量中带有app和g两个属性，协助进行调试程序。
shell_command的实现:
@with_appcontext
def shell_command():
"""Run an interactive Python shell in the context of a given
Flask application. The application will populate the default
namespace of this shell according to it's configuration.
This is useful for executing small snippets of management code
without having to manually configure the application.
"""
import code
from .globals import _app_ctx_stack
app = _app_ctx_stack.top.app
banner = "Python %s on %s\nApp: %s [%s]\nInstance: %s" % (
sys.version,
sys.platform,
app.import_name,
app.env,
app.instance_path,
)
ctx = {}
# Support the regular Python interpreter startup script if someone
# is using it.
startup = os.environ.get("PYTHONSTARTUP")
if startup and os.path.isfile(startup):
with open(startup, "r") as f:
eval(compile(f.read(), startup, "exec"), ctx)
ctx.update(app.make_shell_context())
code.interact(banner=banner, local=ctx)shell主要利用code模块的 interact 实现，在之前介绍德国锤子werkzeug时也有过介绍。
在make_shell_context中可以看到app和g两个属性设置:
def make_shell_context(self):
rv = {"app": self, "g": g}
for processor in self.shell_context_processors:
rv.update(processor())
return rv利用参考shell命令可以实现程序热更功能，以后有机会再详细介绍。
最复杂和重要的就是run命令了, run_command的注释很清晰的介绍了功能:
@pass_script_info
def run_command(
info, host, port, reload, debugger, eager_loading, with_threads, cert, extra_files
):
"""Run a local development server.
This server is for development purposes only. It does not provide
the stability, security, or performance of production WSGI servers.
The reloader and debugger are enabled by default if
FLASK_ENV=development or FLASK_DEBUG=1.
"""
...
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,
extra_files=extra_files,
)DispatchingApp实现也不复杂:
class DispatchingApp(object):
def __init__(self, loader, use_eager_loading=False):
self.loader = loader
self._app = None
self._lock = Lock()
self._bg_loading_exc_info = None
if use_eager_loading:
self._load_unlocked()
else:
self._load_in_background()
def _load_in_background(self):
def _load_app():
__traceback_hide__ = True # noqa: F841
with self._lock:
try:
self._load_unlocked()
except Exception:
self._bg_loading_exc_info = sys.exc_info()
t = Thread(target=_load_app, args=())
t.start()
def _load_unlocked(self):
__traceback_hide__ = True # noqa: F841
self._app = rv = self.loader()
self._bg_loading_exc_info = None
return rv
...
def __call__(self, environ, start_response):
__traceback_hide__ = True # noqa: F841
if self._app is not None:
return self._app(environ, start_response)
self._flush_bg_loading_exception()
with self._lock:
if self._app is not None:
rv = self._app
else:
rv = self._load_unlocked()
return rv(environ, start_response)本文我们主要理解了flask-cli的实现逻辑，当然要完全弄懂flask-cli，还需要深入理解click和dotenv的实现，不过掌握到这个程度，我个人认为已经够用了。我们简单小结一下flask-cli的功能:
Copyright© 2013-2020
All Rights Reserved 京ICP备2023019179号-8