Previously I took a look at the source code of Bottle, so this time I would like to take a look at Flask.
According to Flask's tutorial, a minimal application can be created like this.
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello_world():
return "Hello World!"
if __name__ == '__main__':
app.run()
It's almost the same as Bottle.
The definition of the run method looks like this.
src/flask/app.py
class Flask(_PackageBoundObject):
def run(self, host=None, port=None, debug=None, load_dotenv=True, **options):
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
In the case of Bottle, you used the wsgiref library.
When starting a server with wsgiref.simple_server.make_server () and creating a response, a fishy process like start_response () was implemented.
Flask, on the other hand, uses a convenient library called werkzeug.
According to Documentation, the sample looks like this.
from werkzeug.wrappers import Request, Response
@Request.application
def application(request):
return Response('Hello, World!')
if __name__ == '__main__':
from werkzeug.serving import run_simple
run_simple('localhost', 4000, application)
You can easily return a response like return Response ('Hello, World!').
Same as Bottle in that the Flask class of Honmaru is registered as an application function withrun_simple (host, port, self, ** options).
That means that the response content should be obtained by calling the __call__ method in the formFlask () ().
src/flask/app.py
class Flask(_PackageBoundObject):
def route(self, rule, **options):
"""A decorator that is used to register a view function for a
given URL rule. This does the same thing as :meth:`add_url_rule`
but is intended for decorator usage::
@app.route('/')
def index():
return 'Hello World'
For more information refer to :ref:`url-route-registrations`.
(Omitted below)
"""
def decorator(f):
endpoint = options.pop("endpoint", None)
self.add_url_rule(rule, endpoint, f, **options)
return f
return decorator
@setupmethod
def add_url_rule(
self,
rule,
endpoint=None,
view_func=None,
provide_automatic_options=None,
**options
):
"""
Basically this example::
@app.route('/')
def index():
pass
Is equivalent to the following::
def index():
pass
app.add_url_rule('/', 'index', index)
(Omitted below)
"""
if endpoint is None:
endpoint = _endpoint_from_view_func(view_func)
options["endpoint"] = endpoint
methods = options.pop("methods", None)
# if the methods are not given and the view_func object knows its
# methods we can use that instead. If neither exists, we go with
# a tuple of only ``GET`` as default.
if methods is None:
methods = getattr(view_func, "methods", None) or ("GET",)
if isinstance(methods, string_types):
raise TypeError(
"Allowed methods have to be iterables of strings, "
'for example: @app.route(..., methods=["POST"])'
)
methods = set(item.upper() for item in methods)
# Methods that should always be added
required_methods = set(getattr(view_func, "required_methods", ()))
# starting with Flask 0.8 the view_func object can disable and
# force-enable the automatic options handling.
if provide_automatic_options is None:
provide_automatic_options = getattr(
view_func, "provide_automatic_options", None
)
if provide_automatic_options is None:
if "OPTIONS" not in methods:
provide_automatic_options = True
required_methods.add("OPTIONS")
else:
provide_automatic_options = False
# Add the required methods now.
methods |= required_methods
rule = self.url_rule_class(rule, methods=methods, **options)
rule.provide_automatic_options = provide_automatic_options
self.url_map.add(rule)
if view_func is not None:
old_func = self.view_functions.get(endpoint)
if old_func is not None and old_func != view_func:
raise AssertionError(
"View function mapping is overwriting an "
"existing endpoint function: %s" % endpoint
)
self.view_functions[endpoint] = view_func
after all
self.view_functions[endpoint] = view_func
In the part of, the view function defined by the user is registered as a dictionary. The basics are the same as Bottle.
I was told that I would get the response content from the __call__ method of the Flask instance.
src/flask/app.py
class Flask(_PackageBoundObject):
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)
def wsgi_app(self, environ, start_response):
"""The actual WSGI application. This is not implemented in
:meth:`__call__` so that middlewares can be applied without
losing a reference to the app object. Instead of doing this::
app = MyMiddleware(app)
It's a better idea to do this instead::
app.wsgi_app = MyMiddleware(app.wsgi_app)
Then you still have the original application object around and
can continue to call methods on it.
"""
ctx = self.request_context(environ)
error = None
try:
try:
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)
def full_dispatch_request(self):
"""Dispatches the request and on top of that performs request
pre and postprocessing as well as HTTP exception catching and
error handling.
"""
self.try_trigger_before_first_request_functions()
try:
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
I'm also worried about the part of request_started.send (self).
I wonder if this manages the process.
But I'll leave this for a while.
So what I want to chase now is
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
In the part of, basically, data is received from preprocess_request () or self.dispatch_request ().
Is rv an abbreviation for response body?
So let's take a look at these two methods.
１、 process_response()
src/flask/app.py
class Flask(_PackageBoundObject):
def process_response(self, response):
"""Can be overridden in order to modify the response object
before it's sent to the WSGI server. By default this will
call all the :meth:`after_request` decorated functions.
"""
ctx = _request_ctx_stack.top
bp = ctx.request.blueprint
funcs = ctx._after_request_functions
if bp is not None and bp in self.after_request_funcs:
funcs = chain(funcs, reversed(self.after_request_funcs[bp]))
if None in self.after_request_funcs:
funcs = chain(funcs, reversed(self.after_request_funcs[None]))
for handler in funcs:
response = handler(response)
if not self.session_interface.is_null_session(ctx.session):
self.session_interface.save_session(self, ctx.session, response)
return response
ctx means "context", right?
The handlers registered in context are made to execute the process.
Then when was ctx stacked on _request_ctx_stack?
Actually, it passed through, but the context was created and pushed in the wsgi_app () method of the Flask class I saw earlier (please look back).
By the way, _request_ctx_stack is
src/flask/globals.py
_request_ctx_stack = LocalStack()
Created in. Pushes are defined below.
src/flask/ctx.py
class RequestContext(object):
def push(self):
"""Binds the request context to the current context."""
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop(top._preserved_exc)
# Before we push the request context we have to ensure that there
# is an application context.
app_ctx = _app_ctx_stack.top
if app_ctx is None or app_ctx.app != self.app:
app_ctx = self.app.app_context()
app_ctx.push()
self._implicit_app_ctx_stack.append(app_ctx)
else:
self._implicit_app_ctx_stack.append(None)
if hasattr(sys, "exc_clear"):
sys.exc_clear()
_request_ctx_stack.push(self)
２、 dispatch_request()
src/flask/app.py
class Flask(_PackageBoundObject):
def dispatch_request(self):
"""Does the request dispatching. Matches the URL and returns the
return value of the view or error handler. This does not have to
be a response object. In order to convert the return value to a
proper response object, call :func:`make_response`.
"""
req = _request_ctx_stack.top.request
if req.routing_exception is not None:
self.raise_routing_exception(req)
rule = req.url_rule
# if we provide automatic options for this URL and the
# request came with the OPTIONS method, reply automatically
if (
getattr(rule, "provide_automatic_options", False)
and req.method == "OPTIONS"
):
return self.make_default_options_response()
# otherwise dispatch to the handler for that endpoint
return self.view_functions[rule.endpoint](**req.view_args)
Next, here we are executing the function obtained by self.view_functions [rule.endpoint], but when will it be registered here?
I found two scenes.
The first is towards the end of the Flask.add_url_rule () method we saw earlier (look back at this too).
The second is
class Flask():
@setupmethod
def endpoint(self, endpoint):
def decorator(f):
self.view_functions[endpoint] = f
return f
return decorator
Part of.
But these are not different usage situations, for example, because endpoint did when ʻadd_url_rule ()` was done, hmm. I'm getting tired, so it's getting harder to check.
Well, I feel like I saw ʻadd_url_rule (), and after all, the endpoint function defined by the user in dispatch_request ()` is being executed.
Then, is process_response () a plug-in or peripheral processing?
I'm tired, so this time around.
Recommended Posts