ICode9

精准搜索请尝试: 精确搜索
首页 > 编程语言> 文章详细

flask 源码梗概

2022-07-26 20:01:28  阅读:202  来源: 互联网

标签:None endpoint 梗概 flask self 源码 view config methods


flask 源码梗概

flask 中的线程主要基于LocalStack进行使用,在global中维护这个类的两个对象。

# context locals
_request_ctx_stack = LocalStack() # 请求上下文:主要有 request 和 session 两个对象
_app_ctx_stack = LocalStack() # 应用上下文 :主要有 app 和 g 两个值。

梗概.drawio

额外补充:点击下图框选的图标,可以直接定位到源码所在的文件以及相关的文件路径。

image-20220724183136487

flask 源码启动阶段

说明:启动阶段即为,flask 初始化创建网站开始,请求还未到来之时的阶段。

启动阶段的代码主要如下所示:

from flask import Flask

app = Flask(__name__) # 创建app 对象

    app.config.from_object("config.settings") # 加载配置文件

@app.route("/index") # 注册视图函数
def index():
    return "Hello World"
if __name__ == '__main__':
    app.run() #程序启动

程序启动的步骤主要是有werkzug部分执行app__call__()方法,上述步骤在前面已经剖析过此处不在进行剖析。

  • 实例化对象

    app = Flask(__name__) # 创建app 对象
    

    该步骤会去执行Flask的初始化函数__init__()方法。该方法的剖析如下

    class Flask(Scaffold):# 继承与内部实现的类
        config_class = Config
        secret_key = ConfigAttribute("SECRET_KEY")
        url_map_class = Map
        '''
        	存储了rule 对象
        '''
        url_rule_class = Rule
        '''
        	存储了 路由,以及相关的请求方式,
        	'/index',['GET','Post',],endpoint
        '''
        
        config_class = Config# 此处与配置文件有关
        default_config = ImmutableDict( # ImmutableDict是实现一个不可变的字典
            {
                "ENV": None,
                "DEBUG": None,
                "SECRET_KEY": None,
                "SESSION_COOKIE_NAME": "session",
                "PREFERRED_URL_SCHEME": "http",
                "JSONIFY_PRETTYPRINT_REGULAR": False,
                "JSONIFY_MIMETYPE": "application/json",
    			# 更多的默认配置...
            }
        )
    
        '''... 更多其他的静态变量,不在展示'''
        def __init__(
            self,
            import_name: str,
            static_url_path: t.Optional[str] = None,
            static_folder: t.Optional[t.Union[str, os.PathLike]] = "static",
            static_host: t.Optional[str] = None,
            host_matching: bool = False,
            subdomain_matching: bool = False,
            template_folder: t.Optional[str] = "templates",
            instance_path: t.Optional[str] = None,
            instance_relative_config: bool = False,
            root_path: t.Optional[str] = None,
        ):
            # 实例属性
            self.instance_path = instance_path
            self.url_map = self.url_map_class() # 加(),此处实例化了一个 Map()对象,静态属性url_map_class = Map
            self.config = self.make_config(instance_relative_config)
            self.view_functions: t.Dict[str, t.Callable] = {} # view_functions 为一个字典
        	# ... 更多的实例化属性以及注释,此处不在展示。
            
            if self.has_static_folder:
                assert (
                    bool(static_host) == host_matching
                ), "Invalid static_host/host_matching combination"
                # Use a weakref to avoid creating a reference cycle between the app
                # and the view function (see #3761).
                self_ref = weakref.ref(self)
                self.add_url_rule(
                    # 此阶段的语句表示,即便是没有注册过任何视图函数以及路由,也会有相关的静态路由直接存储在对象中。
                    f"{self.static_url_path}/<path:filename>",
                    endpoint="static",
                    host=static_host,
                    view_func=lambda **kw: self_ref().send_static_file(**kw),
                )
    
  • 加载配置文件

    app.config.from_object("config.settings") # 加载配置文件
    '''
    此步骤便于理解可以写为以下两步
    v = app.config
    v.from_object("config.settings")
    '''
    

    from_object()源码如下图

    def from_object(self, obj: t.Union[object, str]) -> None:
        if isinstance(obj, str): # 判断参数类型
            obj = import_string(obj)
        for key in dir(obj): # 循环对象中的所有键值对
            if key.isupper():# 检查对否都为大写字母
                self[key] = getattr(obj, key) # 通过反射的方式为(self)当前对象进行赋值
    

    当前self是谁的对象需要继续看是谁在调用它。

    # app.config 是因为实例属性中包含了相关的变量
    self.config = self.make_config(instance_relative_config) # 执行的是make_config方法
    

    make_config 的源码如下:

    def make_config(self, instance_relative: bool = False) -> Config: #此处表明返回的是一个Config对象
        root_path = self.root_path
        if instance_relative:
            root_path = self.instance_path 
        defaults = dict(self.default_config) #默认配置,导入的是Flask内部的默认配置
        defaults["ENV"] = get_env()
        defaults["DEBUG"] = get_debug_flag()
        return self.config_class(root_path, defaults) 
    

    由上述源码可知,当我们不去加载自己的配置文件的时候,Flask 会默认加载相关的配置。

    此处的返回值为config_class(),在flask源码中,静态变量有config_class = Config,表明该方法是该类的一个赋值,但是并未实例化,此处加上(),表示实例化该对象,相关的源码如下所示:

    class Config(dict):
    	# 执行初始胡方法
        def __init__(self, root_path: str, defaults: t.Optional[dict] = None) -> None:
            super().__init__(defaults or {}) # 执行父类的传参,默认配置,或者是一个空字典
            self.root_path = root_path # 将路径设置到相关的属性上
    

    由于该类继承的是空字典或者是flask内部的不可变字典ImmutableDict,所以回到本质上讲,配置文件的加载是将相关的配置,以键值的形式加载到对应的字典中,配置文件的对象是一个字典

  • 注册路由及视图函数

    # @app.route("/index",methods=['GET','POST'])
    @app.route("/index") # 注册视图函数
    def index():
        return "Hello World"
    

    上述可知Flask类继承与Scaffold类,而装饰器route则属于该父类多定义的方法

    class Scaffold:
        # 该方法符合装饰器的定义
        def route(self, rule: str, **options: t.Any) -> t.Callable[[F], F]:
            def decorator(f: F) -> F: # 定义函数
                endpoint = options.pop("endpoint", None) # 检查参数中是否存在路由的反向解释符。
                self.add_url_rule(rule, endpoint, f, **options) # 执行相关的路由添加函数。
                # !!! 特别注意此处执行的`add_url_rule`并不一定是父类中的该方法,flask中重写了该方法,此时的self,代指的是当前的 app 对象
                return f
    
            return decorator # 返回内部所执行的函数。
    

    add_url_rule相关源码如下

    补充:字典的pop()方法。

    pop() 方法从字典中删除指定的项目。被删除的项目的是这个 pop() 方法的返回值,第二个参数可以设置默认返回值。请看下面的例子。

    dic = {1:"aa","methods":['GET','POST']}
    print(dic.pop("methods",None)) # ['GET', 'POST']
    print(dic) # {1: 'aa'}
    print(dic.pop("methodsasss",None)) # None
    

    flask 源码中的请求方法的限定正式基于此方法实现获取传入的值。

    @setupmethod
    def add_url_rule(
        self,
        rule: str,# 路由字符串
        endpoint: t.Optional[str] = None,# '路由的名字'
        view_func: t.Optional[t.Callable] = None,# 视图函数
        provide_automatic_options: t.Optional[bool] = None,
        **options: t.Any,
    ) -> None:
        if endpoint is None:
            endpoint = _endpoint_from_view_func(view_func)  # endpoint 如果为空,则能与函数的名
        options["endpoint"] = endpoint # 将 endpoint 值写入到字典中去,
        # 通过此方法获取相应的 methods的列表值,没有传入则返回值为None
        methods = options.pop("methods", None) # 此处因为route装饰器可以传入参数 methods=['GET','POST']
    
        # 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: # 使用反射,为view_func 设置相关的值,没有传入方法的时候,或者将methods使用GET 作为默认年至进行赋值给
            methods = getattr(view_func, "methods", None) or ("GET",)
            # view_func 为装饰器传入的函数,使用 getattr 获取相关的值
        if isinstance(methods, str): # 检查为字符串类型
            raise TypeError(
                "Allowed methods must be a list of strings, for"
                ' example: @app.route(..., methods=["POST"])'
            )
        methods = {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
            )
    	# 是否禁用http,OPTIONS 自动响应的实现。
        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 # 更新集合,添加所有其他元素,Python2.6 版本更新的语法。	
    	
        # url_rule_class这个变量在Flask的类属性中进行了声明,    url_rule_class = Rule
        rule = self.url_rule_class(rule, methods=methods,**options) # 将路由规则和请求方式传入,返回的是Rule对象
        # 此时rule对象中存储了路由和请求方式,如果存在endpoint,options中也一并写入了endpoint。Rule的参数会接收这些参数
       
        # 将自动响应的方式,赋值到rule对象中去。
        rule.provide_automatic_options = provide_automatic_options  # type: ignore
    
        self.url_map.add(rule) # 将rule对象添加到对应的map对象中,
        if view_func is not None: # 如果视图函数不为空
            old_func = self.view_functions.get(endpoint) # 字典中根据endpoint 获取相关的函数
            if old_func is not None and old_func != view_func: # endponit 重复,抛出相关的异常
                raise AssertionError(
                    "View function mapping is overwriting an existing"
                    f" endpoint function: {endpoint}"
                )
            self.view_functions[endpoint] = view_func # 将试图函数写入对应的字典中,使用endpoint作为字典的键
    

    相关联的源码

    def _endpoint_from_view_func(view_func: t.Callable) -> str:
        assert view_func is not None, "expected view func if endpoint is not provided."
        return view_func.__name__ # 返回函数的名称。
    

    总结:

    1.将url='/index' 和 methods= ['GET','Post'] 和 endpoint ='index'封装到Rule对象,注:endpoint 默认是函数名
    
    2.将Rule 对象添加到 url_map 中去
    
    3.把 endpoint 和函数的对应关系放到了 view_functions 字典中去。
    
  • 目前总结

    # 源码中使用的主要属性为一下步骤。
    app.config
    app.url_map
    app.view_functions
    
  • 运行

    app.run(),本步骤与上述的启动函数相同。

标签:None,endpoint,梗概,flask,self,源码,view,config,methods
来源: https://www.cnblogs.com/Blogwj123/p/16522441.html

本站声明: 1. iCode9 技术分享网(下文简称本站)提供的所有内容,仅供技术学习、探讨和分享;
2. 关于本站的所有留言、评论、转载及引用,纯属内容发起人的个人观点,与本站观点和立场无关;
3. 关于本站的所有言论和文字,纯属内容发起人的个人观点,与本站观点和立场无关;
4. 本站文章均是网友提供,不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属;如您发现该文章侵犯了您的权益,可联系我们第一时间进行删除;
5. 本站为非盈利性的个人网站,所有内容不会用来进行牟利,也不会利用任何形式的广告来间接获益,纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

专注分享技术,共同学习,共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有