本文共 24400 字，大约阅读时间需要 81 分钟。

Flask源码分析

本文环境python3.5.2，flask-1.0.2。

Flask的初探

首先，在项目文件夹下建立flask_run.py文件，然后写入如下，

from flask   import Flaskapp = Flask(__name__)@app.route('/')def hello_world():    return 'Hello, World!'

在该文件目录下的终端中，输入如下，

(venv) wuzideMacBook-Air:flask wuzi$ export FLASK_APP=flask_run.py(venv) wuzideMacBook-Air:flask wuzi$ flask run

此时flask就已经启动，然后打开浏览器输入127.0.0.1:5000就在浏览器中看到，

'Hello, World!'

此时，flask最简单的helloworld接口已经完成。

Flask的启动源码分析

首先从终端分析开始，当在终端中输入flask run的时候，此时找到bin/flask文件，

# -*- coding: utf-8 -*-import reimport sysfrom flask.cli import mainif __name__ == '__main__':    sys.argv[0] = re.sub(r'(-script\.pyw?|\.exe)?$', '', sys.argv[0])    sys.exit(main())

此时调用了flask.cli中的main方法，

def main(as_module=False):

args = sys.argv[1:] # 获取输入参宿

if as_module:                                           # 是否是python 脚本启动    this_module = 'flask'    if sys.version_info < (2, 7):                       # 检查版本        this_module += '.cli'    name = 'python -m ' + this_module    # Python rewrites "python -m flask" to the path to the file in argv.    # Restore the original command so that the reloader works.    sys.argv = ['-m', this_module] + args              # 重置输入参数else:    name = Nonecli.main(args=args, prog_name=name)                    # 调用cli的main函数

此时的cli就是,

cli = FlaskGroup(help="""\A general utility script for Flask applications.Provides commands from Flask, extensions, and the application. Loads theapplication defined in the FLASK_APP environment variable, or from a wsgi.pyfile. Setting the FLASK_ENV environment variable to 'development' will enabledebug 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 '> '))

cli就是FlaskGroup的一个实例，其中该类是继承自AppGroup，该类在初始化的时候，会将flask默认的类的参数传入，

def __init__(self, add_default_commands=True, create_app=None,             add_version_option=True, load_dotenv=True, **extra):        params = list(extra.pop('params', None) or ())        if add_version_option:            params.append(version_option)        AppGroup.__init__(self, params=params, **extra)        self.create_app = create_app        self.load_dotenv = load_dotenv        if add_default_commands:            self.add_command(run_command)            self.add_command(shell_command)            self.add_command(routes_command)        self._loaded_plugin_commands = False

其中通过self.add_command(run_command)就将run命令传入其中，run_command的函数为，

@click.command('run', short_help='Runs a development server.')@click.option('--host', '-h', default='127.0.0.1',              help='The interface to bind to.')@click.option('--port', '-p', default=5000,              help='The port to bind to.')@click.option('--cert', type=CertParamType(),              help='Specify a certificate file to use HTTPS.')@click.option('--key',              type=click.Path(exists=True, dir_okay=False, resolve_path=True),              callback=_validate_key, expose_value=False,              help='The key file to use when specifying a certificate.')@click.option('--reload/--no-reload', default=None,              help='Enable or disable the reloader. By default the reloader '              'is active if debug is enabled.')@click.option('--debugger/--no-debugger', default=None,              help='Enable or disable the debugger. By default the debugger '              'is active if debug is enabled.')@click.option('--eager-loading/--lazy-loader', default=None,              help='Enable or disable eager loading. By default eager '              'loading is enabled if the reloader is disabled.')@click.option('--with-threads/--without-threads', default=True,              help='Enable or disable multithreading.')@pass_script_infodef run_command(info, host, port, reload, debugger, eager_loading,                with_threads, cert):    """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.    """    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)

其中就将run命令行对应的参数传入其中并设置了相关的默认值。

此时调用的就是FlaskGroup的main方法，

def main(self, *args, **kwargs):    # Set a global flag that indicates that we were invoked from the    # command line interface. This is detected by Flask.run to make the    # call into a no-op. This is necessary to avoid ugly errors when the    # script that is loaded here also attempts to start a server.    os.environ['FLASK_RUN_FROM_CLI'] = 'true'                   # 设置环境变量    if get_load_dotenv(self.load_dotenv):                       # 检查是否加载        load_dotenv()    obj = kwargs.get('obj')                                     # 检查参数中是否传入obj    if obj is None:                                             # 如果没有传入        obj = ScriptInfo(create_app=self.create_app)            # 初始化ScriptInfo实例    kwargs['obj'] = obj                                         # 设置obj    kwargs.setdefault('auto_envvar_prefix', 'FLASK')                return super(FlaskGroup, self).main(*args, **kwargs)        # 调用父类的main方法

其中ScriptInfo类如下，

class ScriptInfo(object):    """Help object to deal with Flask applications.  This is usually not    necessary to interface with as it's used internally in the dispatching    to click.  In future versions of Flask this object will most likely play    a bigger role.  Typically it's created automatically by the    :class:`FlaskGroup` but you can also manually create it and pass it    onwards as click object.    """    def __init__(self, app_import_path=None, create_app=None):        #: Optionally the import path for the Flask application.        self.app_import_path = app_import_path or os.environ.get('FLASK_APP')   # 是否传入app导入路径，如果没有则使用环境变量中的FLASK_APP设置的路径        #: Optionally a function that is passed the script info to create        #: the instance of the application.        self.create_app = create_app                                            # 创建app的方法        #: A dictionary with arbitrary data that can be associated with        #: this script info.        self.data = {}        self._loaded_app = None    def load_app(self):        """Loads the Flask app (if not yet loaded) and returns it.  Calling        this multiple times will just result in the already loaded app to        be returned.        """        __traceback_hide__ = True        if self._loaded_app is not None:                                        # 检查是否已经创建_loader_app            return self._loaded_app                                             # 如果已经创建则返回        app = None                                                                      if self.create_app is not None:                                         # 初始化传入的创建方法是否为空            app = call_factory(self, self.create_app)                           # 创建app实例        else:             if self.app_import_path:                                            # 如果传入文件路径                path, name = (self.app_import_path.split(':', 1) + [None])[:2]  # 获取路径和名称                import_name = prepare_import(path)                                              app = locate_app(self, import_name, name)                       # 导入app            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)                  # 导入app                    if app:                                                     # 如果导入成功则停止循环                        break        if not app:                                                             # 如果app为空则报错            raise NoAppException(                'Could not locate a Flask application. You did not provide '                'the "FLASK_APP" environment variable, and a "wsgi.py" or '                '"app.py" module was not found in the current directory.'            )        debug = get_debug_flag()                                                # 获取是否是调试模式        # Update the app's debug flag through the descriptor so that other        # values repopulate as well.        if debug is not None:                                                   # 如果获取的值不为空            app.debug = debug                                                   # 设置app的模式        self._loaded_app = app                                                  # 将加载的app设置到_loaded_app属性中        return app                                                              # 返回app

执行到最后就调用了父类的main方法执行，此时调用的就是包click/core.py中的BaseCommand类中的main函数，

def main(self, args=None, prog_name=None, complete_var=None,         standalone_mode=True, **extra):    """This is the way to invoke a script with all the bells and    whistles as a command line application.  This will always terminate    the application after a call.  If this is not wanted, ``SystemExit``    needs to be caught.    This method is also available by directly calling the instance of    a :class:`Command`.    .. versionadded:: 3.0       Added the `standalone_mode` flag to control the standalone mode.    :param args: the arguments that should be used for parsing.  If not                 provided, ``sys.argv[1:]`` is used.    :param prog_name: the program name that should be used.  By default                      the program name is constructed by taking the file                      name from ``sys.argv[0]``.    :param complete_var: the environment variable that controls the                         bash completion support.  The default is                         ``"_
   
    _COMPLETE"`` with prog name in                         uppercase.    :param standalone_mode: the default behavior is to invoke the script                            in standalone mode.  Click will then                            handle exceptions and convert them into                            error messages and the function will never                            return but shut down the interpreter.  If                            this is set to `False` they will be                            propagated to the caller and the return                            value of this function is the return value                            of :meth:`invoke`.    :param extra: extra keyword arguments are forwarded to the context                  constructor.  See :class:`Context` for more information.    """    # If we are in Python 3, we will verify that the environment is    # sane at this point of reject further execution to avoid a    # broken script.    if not PY2:        _verify_python3_env()    else:        _check_for_unicode_literals()    if args is None:        args = get_os_args()    else:        args = list(args)    if prog_name is None:        prog_name = make_str(os.path.basename(            sys.argv and sys.argv[0] or __file__))    # Hook for the Bash completion.  This only activates if the Bash    # completion is actually enabled, otherwise this is quite a fast    # noop.    _bashcomplete(self, prog_name, complete_var)    try:        try:            with self.make_context(prog_name, args, **extra) as ctx:                rv = self.invoke(ctx)                if not standalone_mode:                    return rv                ctx.exit()        except (EOFError, KeyboardInterrupt):            echo(file=sys.stderr)            raise Abort()        except ClickException as e:            if not standalone_mode:                raise            e.show()            sys.exit(e.exit_code)    except Abort:        if not standalone_mode:            raise        echo('Aborted!', file=sys.stderr)        sys.exit(1)
   

此时调用了self.invoke(ctx)，由于是在FlaskGroup调用该方法，该类的继承关系是FlaskGrooup的父类AppGroup的父类click.Group的父类MultiCommand的父类Command的父类BaseCommand,此时调用的inovke是调用了MultiCommand的inovke方法，此时执行流程会执行到，

cmd_name, cmd, args = self.resolve_command(ctx, args)            ctx.invoked_subcommand = cmd_name            Command.invoke(self, ctx)

然后调用了Command的inovke方法，此时调用的方法是，

def invoke(self, ctx):    """Given a context, this invokes the attached callback (if it exists)    in the right way.    """    if self.callback is not None:        return ctx.invoke(self.callback, **ctx.params)

此时就调用了Context类的实例的inovke方法，

def invoke(*args, **kwargs):    """Invokes a command callback in exactly the way it expects.  There    are two ways to invoke this method:    1.  the first argument can be a callback and all other arguments and        keyword arguments are forwarded directly to the function.    2.  the first argument is a click command object.  In that case all        arguments are forwarded as well but proper click parameters        (options and click arguments) must be keyword arguments and Click        will fill in defaults.    Note that before Click 3.2 keyword arguments were not properly filled    in against the intention of this code and no context was created.  For    more information about this change and why it was done in a bugfix    release see :ref:`upgrade-to-3.2`.    """    self, callback = args[:2]    ctx = self    # It's also possible to invoke another command which might or    # might not have a callback.  In that case we also fill    # in defaults and make a new context for this command.    if isinstance(callback, Command):        other_cmd = callback        callback = other_cmd.callback        ctx = Context(other_cmd, info_name=other_cmd.name, parent=self)        if callback is None:            raise TypeError('The given command does not have a '                            'callback that can be invoked.')        for param in other_cmd.params:            if param.name not in kwargs and param.expose_value:                kwargs[param.name] = param.get_default(ctx)    args = args[2:]    with augment_usage_errors(self):        with ctx:            return callback(*args, **kwargs)

在重新设置好上下文后， 调用了com的callback方法，此时当前输入的命令是run，则执行了run对应的Command的callback即run_command方法，该方法如下，

def run_command(info, host, port, reload, debugger, eager_loading,                with_threads, cert):    """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.    """    debug = get_debug_flag()                                                        # 获取调试标志位    if reload is None:                                                              # 如果reload为空则将该值设置成调试模式        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)            # 初始化DispatchingApp类    from werkzeug.serving import run_simple    run_simple(host, port, app, use_reloader=reload, use_debugger=debugger,               threaded=with_threads, ssl_context=cert)                             # 启动服务

其中对应的DispatchingApp类的大致信息如下，

class DispatchingApp(object):    """Special application that dispatches to a Flask application which    is imported by name in a background thread.  If an error happens    it is recorded and shown as part of the WSGI handling which in case    of the Werkzeug debugger means that it shows up in the browser.    """    def __init__(self, loader, use_eager_loading=False):        self.loader = loader                                            # app类的加载方法        self._app = None                                                # 保存app实例        self._lock = Lock()        self._bg_loading_exc_info = None            if use_eager_loading:                                                   self._load_unlocked()                                       # 直接加载生成app        else:            self._load_in_background()                                  # 通过线程生成app    def _load_in_background(self):        def _load_app():            __traceback_hide__ = True            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 _flush_bg_loading_exception(self):        __traceback_hide__ = True        exc_info = self._bg_loading_exc_info        if exc_info is not None:            self._bg_loading_exc_info = None            reraise(*exc_info)    def _load_unlocked(self):        __traceback_hide__ = True        self._app = rv = self.loader()        self._bg_loading_exc_info = None        return rv    def __call__(self, environ, start_response):                        #        __traceback_hide__ = True        if self._app is not None:                                       # 如果_app已经加载            return self._app(environ, start_response)                   # 调用加载的app去处理请求        self._flush_bg_loading_exception()        with self._lock:                                                # 加锁并加载_app            if self._app is not None:                rv = self._app            else:                rv = self._load_unlocked()            return rv(environ, start_response)

其中的info.loader就是ScriptInfo类实例的load_app方法，即就是脚本文件中通过Flask实例的app，此时继续分析，接下来就调用了run_simple函数，

def run_simple(hostname, port, application, use_reloader=False,

use_debugger=False, use_evalex=True, extra_files=None, reloader_interval=1, reloader_type=’auto’, threaded=False, processes=1, request_handler=None, static_files=None, passthrough_errors=False, ssl_context=None): “”“Start a WSGI application. Optional features include a reloader, multithreading and fork support.

This function has a command-line interface too::    python -m werkzeug.serving --help.. versionadded:: 0.5   `static_files` was added to simplify serving of static files as well   as `passthrough_errors`... versionadded:: 0.6   support for SSL was added... versionadded:: 0.8   Added support for automatically loading a SSL context from certificate   file and private key... versionadded:: 0.9   Added command-line interface... versionadded:: 0.10   Improved the reloader and added support for changing the backend   through the `reloader_type` parameter.  See :ref:`reloader`   for more information.:param hostname: The host for the application.  eg: ``'localhost'``:param port: The port for the server.  eg: ``8080``:param application: the WSGI application to execute:param use_reloader: should the server automatically restart the python                     process if modules were changed?:param use_debugger: should the werkzeug debugging system be used?:param use_evalex: should the exception evaluation feature be enabled?:param extra_files: a list of files the reloader should watch                    additionally to the modules.  For example configuration                    files.:param reloader_interval: the interval for the reloader in seconds.:param reloader_type: the type of reloader to use.  The default is                      auto detection.  Valid values are ``'stat'`` and                      ``'watchdog'``. See :ref:`reloader` for more                      information.:param threaded: should the process handle each request in a separate                 thread?:param processes: if greater than 1 then handle each request in a new process                  up to this maximum number of concurrent processes.:param request_handler: optional parameter that can be used to replace                        the default one.  You can use this to replace it                        with a different                        :class:`~BaseHTTPServer.BaseHTTPRequestHandler`                        subclass.:param static_files: a list or dict of paths for static files.  This works                     exactly like :class:`SharedDataMiddleware`, it's actually                     just wrapping the application in that middleware before                     serving.:param passthrough_errors: set this to `True` to disable the error catching.                           This means that the server will die on errors but                           it can be useful to hook debuggers in (pdb etc.):param ssl_context: an SSL context for the connection. Either an                    :class:`ssl.SSLContext`, a tuple in the form                    ``(cert_file, pkey_file)``, the string ``'adhoc'`` if                    the server should automatically create one, or ``None``                    to disable SSL (which is the default)."""if not isinstance(port, int):    raise TypeError('port must be an integer')if use_debugger:    from werkzeug.debug import DebuggedApplication    application = DebuggedApplication(application, use_evalex)if static_files:    from werkzeug.wsgi import SharedDataMiddleware    application = SharedDataMiddleware(application, static_files)def log_startup(sock):    display_hostname = hostname not in ('', '*') and hostname or 'localhost'    if ':' in display_hostname:        display_hostname = '[%s]' % display_hostname    quit_msg = '(Press CTRL+C to quit)'    port = sock.getsockname()[1]    _log('info', ' * Running on %s://%s:%d/ %s',         ssl_context is None and 'http' or 'https',         display_hostname, port, quit_msg)def inner():    try:        fd = int(os.environ['WERKZEUG_SERVER_FD'])    except (LookupError, ValueError):        fd = None    srv = make_server(hostname, port, application, threaded,                      processes, request_handler,                      passthrough_errors, ssl_context,                      fd=fd)    if fd is None:        log_startup(srv.socket)    srv.serve_forever()if use_reloader:    # If we're not running already in the subprocess that is the    # reloader we want to open up a socket early to make sure the    # port is actually available.    if os.environ.get('WERKZEUG_RUN_MAIN') != 'true':        if port == 0 and not can_open_by_fd:            raise ValueError('Cannot bind to a random port with enabled '                             'reloader if the Python interpreter does '                             'not support socket opening by fd.')        # Create and destroy a socket so that any exceptions are        # raised before we spawn a separate Python interpreter and        # lose this ability.        address_family = select_ip_version(hostname, port)        s = socket.socket(address_family, socket.SOCK_STREAM)        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)        s.bind(get_sockaddr(hostname, port, address_family))        if hasattr(s, 'set_inheritable'):            s.set_inheritable(True)        # If we can open the socket by file descriptor, then we can just        # reuse this one and our socket will survive the restarts.        if can_open_by_fd:            os.environ['WERKZEUG_SERVER_FD'] = str(s.fileno())            s.listen(LISTEN_QUEUE)            log_startup(s)        else:            s.close()    # Do not use relative imports, otherwise "python -m werkzeug.serving"    # breaks.    from werkzeug._reloader import run_with_reloader    run_with_reloader(inner, extra_files, reloader_interval,                      reloader_type)else:    inner()

该函数的大致流程就是再进行相关判断后，检查是否需要修改自动加载功能，如果没有设置则直接调用inner()方法，而该方法就是默认就是启动多线程的server，至此flask run命令的大致执行流程分析完成。

Flask的处理请求

此时当有请求进来时，此时就会通过调用app

application_iter = app(environ, start_response)

此时调用的就是Flask实例的call方法，

def __call__(self, environ, start_response):    """The WSGI server calls the Flask application object as the    WSGI application. This calls :meth:`wsgi_app` which can be    wrapped to applying middleware."""    return self.wsgi_app(environ, start_response)

调用了self.wsgi_app函数处理传入的environ,start_response方法，

此时就开始处理url请求，请求的具体分析留待后文分析。

本文总结

本文主要讲述了flask框架的helloworld程序的编写和启动过程，并分析了在终端启动flask项目的大致执行流程。

转载地址：https://blog.csdn.net/qq_33339479/article/details/81029467 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！