flask笔记(五):了解app

2019-09-25  本文已影响0人  warmsirius

flask学习笔记(四):一个完整的flask程序 中,我们看见了app这个变量,现在我们来深入学习一下到底什么是app?

一、app的产生

1. 创建一个app

从上一节中,不难看出,app是由Flask类的对象。

# 实例化Flask对象
app = Flask(__name__)

2. Flask类部分源码

下面来学习一下Flask类的源码,探究一下什么是Flask 类。

# Flask类的初始化函数部分源码:
def __init__(
        self,
        import_name, 
        static_url_path=None,
        static_folder="static",
        static_host=None,
        host_matching=False,
        subdomain_matching=False,
        template_folder="templates",
        instance_path=None,
        instance_relative_config=False,
        root_path=None,
    ):
    ...

二、 app.run()

1. 启动app

app.run(host="0.0.0.0", port=5000)

这是启动app的方法,如果不指定host和port,host默认为127.0.0.1,port默认为5000

2. app.run源码

下面看看run方法的源码

# 这是Flask类的对象方法,然后app调用这个方法了
    def run(self, host=None, port=None, debug=None, load_dotenv=True, **options):
        # Change this into a no-op if the server is invoked from the
        # command line. Have a look at cli.py for more information.
        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 set, let env vars override previous values
            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()

        # debug passed to method overrides all other sources
        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
        # pick the first value that's not None (0 is allowed)
        port = int(next((p for p in (port, sn_port) if p is not None), _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:
            # reset the first request information if the development server
            # reset normally.  This makes it possible to restart the server
            # without reloader and that stuff from an interactive shell.
            self._got_first_request = False

三、 app.url_map查看所有路由

1. url_map应用举例

from flask import Flask, current_app

app = Flask(__name__)


@app.route('/')
def hello_world():
    return 'Hello World!'


if __name__ == '__main__':
    print(app.url_map)
    app.run()

输出

/Users/yuanjun/.virtualenvs/flask_py3/bin/python /Users/yuanjun/PycharmProject/code/app.py
Map([<Rule '/' (OPTIONS, GET, HEAD) -> hello_world>,
 <Rule '/static/<filename>' (OPTIONS, GET, HEAD) -> static>])
 * Serving Flask app "app" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: on
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
 * Restarting with stat
Map([<Rule '/' (HEAD, OPTIONS, GET) -> hello_world>,
 <Rule '/static/<filename>' (HEAD, OPTIONS, GET) -> static>])
 * Debugger is active!
 * Debugger PIN: 251-856-449

2. 说明

Map([<Rule '/' (HEAD, OPTIONS, GET) -> hello_world>,
 <Rule '/static/<filename>' (HEAD, OPTIONS, GET) -> static>])

3. url_map的源码

self.url_map = self.url_map_class()

self.url_map.host_matching = host_matching
self.subdomain_matching = subdomain_matching

url_map_class = Map


class Map(object):

    #: A dict of default converters to be used.
    default_converters = ImmutableDict(DEFAULT_CONVERTERS)

    def __init__(
        self,
        rules=None,
        default_subdomain="",
        charset="utf-8",
        strict_slashes=True,
        redirect_defaults=True,
        converters=None,
        sort_parameters=False,
        sort_key=None,
        encoding_errors="replace",
        host_matching=False,
    ):
        self._rules = []
        self._rules_by_endpoint = {}
        self._remap = True
        self._remap_lock = Lock()

        self.default_subdomain = default_subdomain
        self.charset = charset
        self.encoding_errors = encoding_errors
        self.strict_slashes = strict_slashes
        self.redirect_defaults = redirect_defaults
        self.host_matching = host_matching

        self.converters = self.default_converters.copy()
        if converters:
            self.converters.update(converters)

        self.sort_parameters = sort_parameters
        self.sort_key = sort_key

        for rulefactory in rules or ():
            self.add(rulefactory)

    def is_endpoint_expecting(self, endpoint, *arguments):
        self.update()
        arguments = set(arguments)
        for rule in self._rules_by_endpoint[endpoint]:
            if arguments.issubset(rule.arguments):
                return True
        return False

    def iter_rules(self, endpoint=None):
        self.update()
        if endpoint is not None:
            return iter(self._rules_by_endpoint[endpoint])
        return iter(self._rules)

    def add(self, rulefactory):
        for rule in rulefactory.get_rules(self):
            rule.bind(self)
            self._rules.append(rule)
            self._rules_by_endpoint.setdefault(rule.endpoint, []).append(rule)
        self._remap = True

    def bind(
        self,
        server_name,
        script_name=None,
        subdomain=None,
        url_scheme="http",
        default_method="GET",
        path_info=None,
        query_args=None,
    ):
        server_name = server_name.lower()
        if self.host_matching:
            if subdomain is not None:
                raise RuntimeError("host matching enabled and a subdomain was provided")
        elif subdomain is None:
            subdomain = self.default_subdomain
        if script_name is None:
            script_name = "/"
        if path_info is None:
            path_info = "/"
        try:
            server_name = _encode_idna(server_name)
        except UnicodeError:
            raise BadHost()
        return MapAdapter(
            self,
            server_name,
            script_name,
            subdomain,
            url_scheme,
            path_info,
            default_method,
            query_args,
        )

    def bind_to_environ(self, environ, server_name=None, subdomain=None):
        environ = _get_environ(environ)

        wsgi_server_name = get_host(environ).lower()

        if server_name is None:
            server_name = wsgi_server_name
        else:
            server_name = server_name.lower()

        if subdomain is None and not self.host_matching:
            cur_server_name = wsgi_server_name.split(".")
            real_server_name = server_name.split(".")
            offset = -len(real_server_name)
            if cur_server_name[offset:] != real_server_name:
                # This can happen even with valid configs if the server was
                # accesssed directly by IP address under some situations.
                # Instead of raising an exception like in Werkzeug 0.7 or
                # earlier we go by an invalid subdomain which will result
                # in a 404 error on matching.
                subdomain = "<invalid>"
            else:
                subdomain = ".".join(filter(None, cur_server_name[:offset]))

        def _get_wsgi_string(name):
            val = environ.get(name)
            if val is not None:
                return wsgi_decoding_dance(val, self.charset)

        script_name = _get_wsgi_string("SCRIPT_NAME")
        path_info = _get_wsgi_string("PATH_INFO")
        query_args = _get_wsgi_string("QUERY_STRING")
        return Map.bind(
            self,
            server_name,
            script_name,
            subdomain,
            environ["wsgi.url_scheme"],
            environ["REQUEST_METHOD"],
            path_info,
            query_args=query_args,
        )

    def update(self):
        """Called before matching and building to keep the compiled rules
        in the correct order after things changed.
        """
        if not self._remap:
            return

        with self._remap_lock:
            if not self._remap:
                return

            self._rules.sort(key=lambda x: x.match_compare_key())
            for rules in itervalues(self._rules_by_endpoint):
                rules.sort(key=lambda x: x.build_compare_key())
            self._remap = False

    def __repr__(self):
        rules = self.iter_rules()
        return "%s(%s)" % (self.__class__.__name__, pformat(list(rules)))
上一篇下一篇

猜你喜欢

热点阅读