从django的rest framwork过渡到flask框架的时候，经常会想flask的路由部分是怎么走的，这篇博客将一步步展示从启动程序到请求来路径和函数是怎么去匹配的。
1.首先是启动flask程序，python解释器就会从上到下加载我们的app
@app.route(‘/home‘，endpoint=‘index‘)
def home():
return render_template(‘dist/index.html‘)
2.在函数上的装饰器即会运行，我们进入装饰器中查看运行内容
def route(self, rule, **options)：
def decorator(f):
endpoint = options.pop("endpoint", None)
self.add_url_rule(rule, endpoint, f, **options)
return f
return decorator
3.运行装饰器，我们将路径rule和参数传入了装饰器中，在装饰器闭包中我们实际是执行了self.add_url_rule(rule, endpoint, f, **options)这个方法，其实映射规则和保存rule对象也是在这个方法中，然后我们进入add_url_rule方法中查看其中做了什么操作
def add_url_rule(
self,
rule,
endpoint=None,
view_func=None,
provide_automatic_options=None,
**options
):
if endpoint is None:
endpoint = _endpoint_from_view_func(view_func)
options["endpoint"] = endpoint
methods = options.pop("methods", None)
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
methods |= required_methods
# 这步时创建rule对象，里面包含路径方法，endpoint，可接受的方法等
rule = self.url_rule_class(rule, methods=methods, **options)
rule.provide_automatic_options = provide_automatic_options
#前面都是关于其他参数是怎么修改和操作的，接下来的才是路由映射关系，下面这步是把rule对象放进一个特殊的字典中的_rules为键的列表中，这里可以进入方法中查看
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
)# 这步就是把endpoint和函数放入一个字典中做映射如{“index”:home}# 那其实到这里我们就可以大概猜想出寻找路由的过程了
self.view_functions[endpoint] = view_func
4.其实到这步我们就可以大概猜想出寻找路由的过程了，肯定当请求到来，我们会先从请求中获取其路径比如：/home，然后我们根据路径去之前特殊字典self.url_map的_rule列表中去匹配rule对象，当匹配到rule对象后我们取出其endpoint的值，再然后根据这个endpoint的值去self.view_functions这个字典中取出函数home，然后传入请求的参数进函数中运行函数。根据这个猜想我们去源码中验证一下。
5.我么找到当请求到时走app的__call__方法我们一步一步看怎么走的
def __call__(self, environ, start_response):
#直接返回wsgi_app我们去看看这个方法
return self.wsgi_app(environ, start_response)
def wsgi_app(self, environ, start_response):
#请求上下文
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)
6.我们进入request_context中这里返回的是一个RequestContext对象，里面封装了request和session，还有其中还用了设计模式的适配器模式，返回了一个self.url_adapter对象底层即MapAdapter对象这个对象封装之前哦我们哪个特殊的字典对象url_map，我们看代码是怎么走的
class RequestContext(object):
#这是刚开进入的请求对象
def __init__(self, app, environ, request=None, session=None):
self.app = app
if request is None:
request = app.request_class(environ)
self.request = request
self.url_adapter = None
try:
#在请求对象中用适配器模式，返回得url对象封装了之前得url_map
self.url_adapter = app.create_url_adapter(self.request)
except HTTPException as e:
self.request.routing_exception = e
self.flashes = None
self.session = session
self._implicit_app_ctx_stack = []
self.preserved = False
self._preserved_exc = None
self._after_request_functions = []
7.然后我们回到wsgi_app函数中，查看ctx.push()中对路由信息做了怎样的操作
def push(self):
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop(top._preserved_exc)
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)
if self.session is None:
session_interface = self.app.session_interface
# 对session中的值解密打开并放入一个特殊的字典中，如果没有值就生成一个空字典
#SecureCookieSessionInterface是session_interface
self.session = session_interface.open_session(self.app, self.request)
if self.session is None:
self.session = session_interface.make_null_session(self.app)
# 这一步就是我们对路由信息进行匹配，看里面发生什么
if self.url_adapter is not None:
self.match_request()
def match_request(self):
try:
#这一步匹配路由规则和参数信息我么进入这个函数看看是怎么匹配的
result = self.url_adapter.match(return_rule=True)#这一步记住这个request,url_rule这一步的赋值就是把rule对象赋给它，和把请求的参数给了view_args
self.request.url_rule, self.request.view_args = result
except HTTPException as e:
self.request.routing_exception = e
def match(self, path_info=None, method=None, return_rule=False, query_args=None):
self.map.update()
if path_info is None:
path_info = self.path_info
else:
path_info = to_unicode(path_info, self.map.charset)
if query_args is None:
query_args = self.query_args
method = (method or self.default_method).upper()
path = u"%s|%s" % (
self.map.host_matching and self.server_name or self.subdomain,
path_info and "/%s" % path_info.lstrip("/"),
)
have_match_for = set()
# 找出匹配的rule，请求的rule
for rule in self.map._rules:
try:
#这里是通过正则匹配，然后返回rv即参数，看是否匹配到。如果匹配到，后续返回rule对象
rv = rule.match(path, method)
except RequestSlash:
raise RequestRedirect(
self.make_redirect_url(
url_quote(path_info, self.map.charset, safe="/:|+") + "/",
query_args,
)
)
except RequestAliasRedirect as e:
raise RequestRedirect(
self.make_alias_redirect_url(
path, rule.endpoint, e.matched_values, method, query_args
)
)
if rv is None:
continue
if rule.methods is not None and method not in rule.methods:
have_match_for.update(rule.methods)
continue
if self.map.redirect_defaults:
redirect_url = self.get_default_redirect(rule, method, rv, query_args)
if redirect_url is not None:
raise RequestRedirect(redirect_url)
if rule.redirect_to is not None:
if isinstance(rule.redirect_to, string_types):
def _handle_match(match):
value = rv[match.group(1)]
return rule._converters[match.group(1)].to_url(value)
redirect_url = _simple_rule_re.sub(_handle_match, rule.redirect_to)
else:
redirect_url = rule.redirect_to(self, **rv)
raise RequestRedirect(
str(
url_join(
"%s://%s%s%s"
% (
self.url_scheme or "http",
self.subdomain + "." if self.subdomain else "",
self.server_name,
self.script_name,
),
redirect_url,
)
)
)
#这个参数默认是为True的所以这里返回和rule对象和rv即参数，接下来这里和我们猜想的越来越接近了，肯定会通过这个rule对象取它的endpoint然后去匹配对应的函数然后执行
if return_rule:
return rule, rv
else:
return rule.endpoint, rv
if have_match_for:
raise MethodNotAllowed(valid_methods=list(have_match_for))
raise NotFound()
8.然后我们回到wsgi_app函数中，我们进入到执行请求的步骤即response = self.full_dispatch_request()这个函数中，我们查看里面发生了什么
def full_dispatch_request(self):
# 第一次请求的钩子函数执行
self.try_trigger_before_first_request_functions()
try:
# flask信号
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)
def dispatch_request(self):
#首先这一步从请求上下文中取出requestcontext对象，这就是我们前面提到的里面有request对象，session对象，有url_apdate对象，然后这里取是request对象
req = _request_ctx_stack.top.request
if req.routing_exception is not None:
self.raise_routing_exception(req)# 果然这一步即取出了request对象的url_rule属性，即前面我们赋值了的rule对象，这个对象就可以取出rule的各种信息属性
rule = req.url_rule
if (
getattr(rule, "provide_automatic_options", False)
and req.method == "OPTIONS"
):
return self.make_default_options_response()
# 这里view_functions自带你就是程序刚启动时的endpoint和函数对应的大字典，然后字典操作，根据请求的rule.endpoint得到函数并且加了括号和参数就运行了这个函数，然后返回结果，至此整个流程就结束了
return self.view_functions[rule.endpoint](**req.view_args)
9.至此，所有得流程即结束了，从路由规则得建立，到请求到来如何取进行匹配，然后执行函数操作。参考和转载麻烦加个链接，码字不易。
原文地址：https://www.cnblogs.com/lifei01/p/12580420.html
时间： 2024-08-30 05:59:21