Flask 源码阅读-开胃菜

from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello():
    return 'Hello, World!'
复制代码

$ 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
复制代码

# 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)
复制代码

# 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
复制代码

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
复制代码

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
复制代码

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)
复制代码

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
复制代码

# flaskr/__init__.py
def create_app(test_config=None):
    app = Flask(__name__, instance_relative_config=True)
    ....
    return app
复制代码

$ flask routes --all-methods
Endpoint  Methods             Rule
--------  ------------------  -----------------------
hello     GET, HEAD, OPTIONS  /
static    GET, HEAD, OPTIONS  /static/<path:filename>
复制代码

@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())
复制代码

$ 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'>,...}
复制代码

@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)
复制代码

def make_shell_context(self):
    rv = {"app": self, "g": g}
    for processor in self.shell_context_processors:
        rv.update(processor())
    return rv
复制代码

@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,
    )
复制代码

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)
复制代码