Flask 源码阅读：开胃菜

flask项目大名鼎鼎，应该不需要多做介绍了吧。我把它称之为python服务开发的TOP2项目，另外一个就是django了，不需要比较孰优孰劣，我的观点是各有千秋，各自应用于不同的场景，都需要深入理解，熟练掌握。本次源码选择的版本是 1.1.2，我会采用慢读法，尽自己最大努力把它讲透。本篇是开胃菜，主要分析flask的命令行工具的实现。

在正式开篇之前，和大家唠叨一点题外话，不喜欢废话的朋友可以直接跳过本段。网上，码农圈经常有各种对比讨论，最经典的梗是: “php是最好的语言！”。我没有嘲笑php的意思，我也用过一段时间的php，做个几个项目。对于这类问题， “小孩才做选择题，成年人都是我都要”  这是开玩笑了，我们重点应该是放在精通上，不应该关闭自己的视野。学习好flask，对精通django有帮助；精通了django，对flask也能够举一反三。如果把自己限定到某一个框架上，有些可惜，毕竟没有一个职业叫做Django/flask-web开发。精力允许的情况下，还是都掌握更好。

flask 示例

示例 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实现的。

flask 命令行工具

模块的启动入口是 __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 decorator

app_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

• run_command 启动flask应用程序
• shell_command 使用shell环境方式调试flask应用程序
• routes_command 查看flask应用程序的路由注册信息

在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方法。

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 app

routes-command

routes命令最简单，使用方法如下:

$ 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())

• 确保执行with_appcontext，在前文有介绍with_appcontext。可见appcontext的重要性。
• 代码整体就做了一件事，收集app的url_map

shell-command

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-command

最复杂和重要的就是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,
    )

• show_server_banner展示启动服务的banner信息
• 使用DispatchingApp包装业务app
• 使用werkzeug的run_simple启动app，证实了前面我们通过banner的猜测

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)

• 使用线程方式加载app或者直接加载app，这个功能会和reload配合使用，一种是饥渴加载一种是懒加载。
• call函数是flask-app的wsgi接口，接收environ和start_response两个核心参数，是请求响应的入口。

小结

本文我们主要理解了flask-cli的实现逻辑，当然要完全弄懂flask-cli，还需要深入理解click和dotenv的实现，不过掌握到这个程度，我个人认为已经够用了。我们简单小结一下flask-cli的功能:

1. 提供shell，routes和run三个命令帮助开发flask应用程序
2. 用户应用程序使用动态加载的方式导入
3. flask推荐创建flask-app或者提供create_app方式提供应用程序入口

参考链接

• https://dormousehole.readthedocs.io/en/latest/
• python-main https://stackoverflow.com/questions/4042905/what-is-main-py
• click https://click.palletsprojects.com/en/8.0.x/
• dotenv https://pypi.org/project/python-dotenv/

ps:本文为该系列文章第一篇，关注【Python开发者】公众号，锁定后续更新。

- EOF -

推荐阅读  点击标题可跳转

觉得本文对你有帮助？请分享给更多人

推荐关注「Python开发者」，提升Python技能

点赞和在看就是最大的支持❤️