Flask 源码解读
1 环境
- Python 3.8.5
- 依赖包:
- click==8.1.2
- colorama==0.4.4
- Flask==2.1.1
- importlib-metadata==4.11.3
- itsdangerous==2.1.2
- Jinja2==3.1.1
- MarkupSafe==2.1.1
- Werkzeug==2.1.0
- zipp==3.7.0
2 WSGI
WSGI（Python Web Server Gateway Interface ）Python 定义的web服务器与web应用程序之间通讯的接口，WSGI的核心意义在于web应用程序与web服务器解耦合，web应用程序专注于处理业务逻辑，web服务器专注于处理客户端的请求。
flask使用werkzueg库实现wsgi，werkzueg是一个优秀的工具库，使用werkzueg实现一个符合WSGI协议的web 应用程序：
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", 5000, application)
run_simple:开启监听localhost的5000端口，application必须是一个可调用对象，这些都是WSGI协议做的规定。
3 flask 启动流程
- 最小的flask应用框架
# 导入Flask
from flask import Flask
# 实例化Flask类
app = Flask(__name__)
# 注册路由
@app.route('/')
def index():
return "hello world"
if __name__ == '__main__':
# 启动测试服务器
app.run()
- app.run()本质上是执行了werkzeug.serving里面的run_simple函数
venv\lib\site-packages\flask\app.py
class Flask:
"""省略"""
def run(
self,
host: t.Optional[str] = None,
port: t.Optional[int] = None,
debug: t.Optional[bool] = None,
load_dotenv: bool = True,
**options: t.Any,
) -> None:
"""前面主要是检测环境变量和判断用户输入参数等"""
try:
run_simple(t.cast(str, host), port, self, **options)
finally:
self._got_first_request = False
"""省略"""
run_simple函数接收了host，port，这两个参数决定监听的端口，第三个参数是一个Flask实例对象，self在Flask类内部，所以self指的是一个实例。
flask启动起来后，监听端口，当有请求进来时，根据wsgi协议，会执行self(),实例对象加括号会调用类对象的_ _ call _ _方法
面向对象知识点补充：实例后面加括号调用的是类的_ _ call _ _ 方法
class Foo:
def __init__(self):
self.name = 'foo'
def __call__(self, *args, **kwargs):
print(args)
print(kwargs)
print(self.name)
f = Foo()
f(1, 2, 3, a=4, b=5)
"""
(1, 2, 3)
{'a': 4, 'b': 5}
foo
"""
4 flask 开始处理请求
flask启动之后，监听指定端口，默认端口5000，当有请求进来后，执行Flask类的 _ _ call _ _方法：
venv\lib\site-packages\flask\app.py
分析源码过程中不涉及的方法暂时省略掉
class Flask:
"""省略"""
def wsgi_app(self, environ: dict, start_response: t.Callable) -> t.Any:
"""
:param environ: A WSGI environment.
:param start_response: A callable accepting a status code,
a list of headers, and an optional exception context to
start the response.
"""
ctx = self.request_context(environ)
error: t.Optional[BaseException] = 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 __call__(self, environ: dict, start_response: t.Callable) -> t.Any:
"""
param1:environ存储请求键值对
param2:start_response:是一个可调用对象
"""
# 继续调用Flask类中的wsgi_app方法
return self.wsgi_app(environ, start_response)
"""省略"""
5 请求上下文
1 ctx = self.request_context(environ)
class Flask:
def request_context(self, environ: dict) -> RequestContext:
"""
创建一个RequestContext实例，封装environ
请求上下文在处理请求时自动推送
可以使用test_request_context手动创建一个请求上下文实例
"""
return RequestContext(self, environ)
venv\lib\site-packages\flask\ctx.py
class RequestContext:
def __init__(
self,
app: "Flask",
environ: dict,
request: t.Optional["Request"] = None,
session: t.Optional["SessionMixin"] = None,
) -> None:
"""
1.app = Flask()
2.ctx = RequestContext()
3.ctx.app = app
"""
self.app = app
if request is None:
# 将Flask类中定义的request_class实例化后赋值给request
request = app.request_class(environ)
self.request = request
self.url_adapter = None
try:
self.url_adapter = app.create_url_adapter(self.request)
except HTTPException as e:
self.request.routing_exception = e
self.flashes = None
self.session = session
venv\lib\site-packages\flask\wrappers.py
class Request(RequestBase):
pass
2 ctx.push()
venv\lib\site-packages\flask\ctx.py
class RequestContext:
def push(self) -> None:
"""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)
# 推送请求上下文之前检测有没有应用上下文
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)
# 将请求上下文入栈
_request_ctx_stack.push(self)
# session后面说
if self.session is None:
session_interface = self.app.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()
3 response = self.full_dispatch_request()
# step1:venv\lib\site-packages\flask\app.py
def wsgi_app(self, environ: dict, start_response: t.Callable) -> t.Any:
"""省略"""
response = self.full_dispatch_request()
"""省略"""
return response(environ, start_response)
# step2：venv\lib\site-packages\flask\app.py
def full_dispatch_request(self) -> Response:
"""省略"""
return self.finalize_request(rv)
"""省略"""
# step3:venv\lib\site-packages\flask\app.py
def finalize_request(
self,
rv: t.Union[ResponseReturnValue, HTTPException],
from_error_handler: bool = False,
) -> Response:
response = self.make_response(rv)
"""省略"""
return response
# step4:venv\lib\site-packages\flask\app.py
def make_response(self, rv: ResponseReturnValue) -> Response:
return rv
# step5:venv\lib\site-packages\flask\response.py
# step1中的return response(environ, start_response)会调用Response类的call方法，实例加括号调用类的call方法在前面也说过了，执行wsgi server的start_response方法返回响应。
class Response:
def __call__(
self, environ: "WSGIEnvironment", start_response: "StartResponse"
) -> t.Iterable[bytes]:
app_iter, status, headers = self.get_wsgi_response(environ)
start_response(status, headers)
return app_iter
6 local、localstack、localproxy
venv\lib\site-packages\flask\globals.py
# context locals
_request_ctx_stack = LocalStack()
_app_ctx_stack = LocalStack()
请求上下文栈、应用上下文栈是flask本地上下文的核心，flask使用本地上下文技术，使得request等对象可以全局使用，而不需要django那样参数传递。
大家可能会有疑问，多个请求同时进来，flask怎么把请求与响应一一对应，下面就开始探索吧
1 预备知识
想要了解local、localstack这两个类的预备知识：
-
预备知识一：Python创建实例的过程
# Python创建实例需要两步： # 1-调用_ _new_ _方法，创建一个实例 # 2-调用_ _init_ _方法，对新创建的实例进行实例化 class Foo: def __new__(cls, *args, **kwargs): print("类型：" + str(cls)) # 类型：<class '__main__.Foo'> self = object.__new__(cls) print("在内存中创建实例" + str(self)) # 在内存中创建实例<__main__.Foo object at 0x000001A0A5968940> return self def __init__(self): print(f"将{self}进行初始化") # 将<__main__.Foo object at 0x000001A0A5968940>进行初始化 self.storage = {}顺便扩展一下可以通过python创建实例这个机制，实现单例模式：
class Foo: _instance = False def __new__(cls, *args, **kwargs): if not cls._instance: cls._instance = object.__new__(cls) return cls._instance else: return cls._instance def __init__(self): self.storage = {} if __name__ == '__main__': foo = Foo() foo2 = Foo() print(foo) # <__main__.Foo object at 0x0000025F73FF8070> print(foo2) # <__main__.Foo object at 0x0000025F73FF8070> # foo和foo2指向的是一个实例 -
预备知识二：特殊方法_ _ getattr _ _ 、_ _ setattr _ _ 、_ _ delattr _ _
class Bar: def __init__(self): self.storage = {} def __getattr__(self, item): return self.storage[item] def __setattr__(self, key, value): self.storage[key] = value def __delattr__(self, item): del self.storage[item] if __name__ == '__main__': b = Bar() # RecursionError: maximum recursion depth exceeded报错原因如下:
class Bar:
def __init__(self):
object.__setattr__(self, 'storage', {})
def __getattr__(self, item):
return self.storage[item]
def __setattr__(self, key, value):
self.storage[key] = value
def __delattr__(self, item):
del self.storage[item]
if __name__ == "__main__":
b = Bar()
b.name = 'zhangsan' # 调用__setattr__
print(b.name) # 调用__getattr__
del b.name # 调用__delattr__
-
预备知识三：_ _ slots _ _ 类属性
Python实例的属性以键值对的形式存储在一个字典对象中
class Bar: def __init__(self): object.__setattr__(self, 'storage', {}) def __getattr__(self, item): return self.storage[item] def __setattr__(self, key, value): self.storage[key] = value def __delattr__(self, item): del self.storage[item] if __name__ == '__main__': b = Bar() print(b.__dict__) # {'storage': {}}设置 _ _ slots _ _属性
class Person: __slots__ = ('name', 'age') def __init__(self, name, age): self.name = name self.age = age if __name__ == '__main__': p1 = Person('zhangsan', 18) p1.gender = 'male' print(p1.name) #将slots属性设置为一个元组，元组内写类实例中定义好的属性，不能在动态添加属性，这样做的目的是节省内存空间，因为字典占用的内存空间远大于元组。
2 Local类
venv\lib\site-packages\flask\local.py
class Local:
__slots__ = ("_storage",) # Local类的实例属性中只能有_storage，不能再设置其他属性
def __init__(self) -> None:
# 设置一个_storage属性，属性值为ContextVar对象
object.__setattr__(self, "_storage", ContextVar("local_storage"))
def __iter__(self) -> t.Iterator[t.Tuple[int, t.Any]]:
# self._storage.get({}) 返回的是一个{} 空字典
# 返回一个迭代器
return iter(self._storage.get({}).items())
def __call__(self, proxy: str) -> "LocalProxy":
"""Create a proxy for a name."""
# 调用self(),返回一个代理对象
return LocalProxy(self, proxy)
def __release_local__(self) -> None:
# 将self._storage设置为一个空字典
self._storage.set({})
def __getattr__(self, name: str) -> t.Any:
# self.xxx 会调用这个方案 xxx将做为name的实际参数
# values最初是一个空字典
values = self._storage.get({})
try:
# 说到根上，这句话执行了dict类的__getitem__方法，不说那么深入，
# 如果字典有name属性则返回，否则抛出KeyError异常
return values[name]
except KeyError:
# 抛出异常
raise AttributeError(name) from None
def __setattr__(self, name: str, value: t.Any) -> None:
# self.xxx = yyy 调用这个方法 xxx是name的实参，yyy是value的实参
# copy当前字典
values = self._storage.get({}).copy()
# 给字典设置值
values[name] = value
# 给self._storage的值设置成新字典
self._storage.set(values)
def __delattr__(self, name: str) -> None:
# del self.xxx 执行这个方法， xxx做为name的实参
# 同上个方法，copy一份
values = self._storage.get({}).copy()
try:
del values[name]
self._storage.set(values)
except KeyError:
raise AttributeError(name) from None
3 LocalStack类
补充一个函数:
在内建函数中有一个getattr函数
def getattr(object, name, default=None): # known special case of getattr
"""
getattr(object, name[, default]) -> value
Get a named attribute from an object; getattr(x, 'y') is equivalent to x.y.
When a default argument is given, it is returned when the attribute doesn't
exist; without it, an exception is raised in that case.
"""
pass
通过getattr可以获取某个对象的某个属性，并且可以设置默认值
venv\lib\site-packages\flask\local.py
class LocalStack:
def __init__(self) -> None:
# 设置_local属性
self._local = Local()
def __release_local__(self) -> None:
# 调用Local类的__release_local__方法
self._local.__release_local__()
def __call__(self) -> "LocalProxy":
# 返回一个LocalProxy对象
def _lookup() -> t.Any:
rv = self.top
if rv is None:
raise RuntimeError("object unbound")
return rv
return LocalProxy(_lookup)
def push(self, obj: t.Any) -> t.List[t.Any]:
"""Pushes a new item to the stack"""
# 入栈操作
# 执行Local类中的__getattr__方法，如果没有stack属性那么设置默认值为一个空列表
rv = getattr(self._local, "stack", []).copy()
# 将对象append到列表尾部
rv.append(obj)
# 调用Local对象中的__setattr__方法，_local._storage = {'stack': [obj]}
self._local.stack = rv
# 返回这个列表
return rv
def pop(self) -> t.Any:
"""弹出栈顶对象，并将弹出的对象返回"""
# 执行Local类中的__getattr__方法
stack = getattr(self._local, "stack", None)
if stack is None:
return None
# 如果stack列表中只有一个元素，将local清空，并返回stack中的这个元素
elif len(stack) == 1:
release_local(self._local)
return stack[-1]
else:
# 如果stack中有多个元素，返回最后一个元素
return stack.pop()
@property
def top(self) -> t.Any:
"""获取栈顶对象，不弹出，只是查看"""
try:
return self._local.stack[-1]
except (AttributeError, IndexError):
return None
4 LocalProxy类
我们在使用flask进行开发时候肯定会用到下面这种结构
from flask import Flask, request
app = Flask()
@app.route('/', methods=['GET', 'POST'])
def index():
if request.method == 'POST':
request.form.get('name')
request.form.get('age')
else:
....
类似django这种框架是这样的
def view(request, *args, **kwargs):
request.form
request.method
request封装的请求相关的参数做为形参传入到视图函数中,这样视图函数就知道为哪个请求服务了
了解了这个场景后，继续回过头看flask源码
venv\lib\site-packages\flask\globals.py
current_app: "Flask" = LocalProxy(_find_app) # type: ignore
request: "Request" = LocalProxy(partial(_lookup_req_object, "request")) # type: ignore
session: "SessionMixin" = LocalProxy( # type: ignore
partial(_lookup_req_object, "session")
)
g: "_AppCtxGlobals" = LocalProxy(partial(_lookup_app_object, "g")) # type: ignore
简化一下：
current_app: = LocalProxy(_find_app)
request = LocalProxy(partial(_lookup_req_object, "request"))
session = LocalProxy( partial(_lookup_req_object, "session"))
g = LocalProxy(partial(_lookup_app_object, "g"))
这四个全局变量都是LocalProxy类的实例：
还是老样子，在看LocalProxy源码之前先做一些知识储备：
-
偏函数的使用：particial
from functools import partial def testfunc(name): print(f'hello {name}') res = partial(testfunc, 'zhangsan') print(res) # functools.partial(<function testfunc at 0x0000020CF6356160>, 'zhangsan') # 偏函数返回一个固定参数的函数引用 res() # hello zhangsan print(callable(res))# True 偏函数返回值是一个可调用对象 -
Python类定义私有属性规则
python是面向”规范“编程的，不管实现什么操作，都有着其固有的规范，相比较java使用关键字定义私有属性，python使用的依然是规范
class Foo: def __init__(self, name): self.__private = name def get_private(self): return self.__private f = Foo('fa') print(f._Foo__private) # ‘fa’ print(f.__dict__) # {'_Foo__private': 'fa'}
拿request为例：
request = LocalProxy(partial(_lookup_req_object, “request”))
def _lookup_req_object(name):
top = _request_ctx_stack.top
if top is None:
raise RuntimeError(_request_ctx_err_msg)
return getattr(top, name)
等价于：request = LocalProxy(_lookup_req_object(‘request’)) 注意这里只是一个函数引用并不会执行函数
下面重点看一下LocalProxy类是怎么定义的
venv\lib\site-packages\flask\local.py
class LocalProxy:
__slots__ = ("__local", "__name", "__wrapped__")
def __init__(self,
local: t.Union["Local", t.Callable[[], t.Any]],
name: t.Optional[str] = None,
) -> None:
object.__setattr__(self, "_LocalProxy__local", local)
object.__setattr__(self, "_LocalProxy__name", name)
if callable(local) and not hasattr(local, "__release_local__"):
# "local" is a callable that is not an instance of Local or
# LocalManager: mark it as a wrapped function.
object.__setattr__(self, "__wrapped__", local)
我把__init__函数简化一下：
class LocalProxy:
# 不能设置动态属性，前面说过slots属性的作用,注意这里的属性都是私有属性，只有在类内可以访问，实例不能直接访问
__slots__ = ("__local", "__name", "__wrapped__")
def __init__(self,local,name=None) -> None:
# 使用object类创建属性，避免递归，前面也提到过
object.__setattr__(self, "_LocalProxy__local", local)
object.__setattr__(self, "_LocalProxy__name", name)
# 如果local是可调用的并且local中不含有__release_local__属性或者方法
if callable(local) and not hasattr(local, "__release_local__"):
# 设置__wrapped__属性，值为local
object.__setattr__(self, "__wrapped__", local)
class LocalProxy:
def _get_current_object(self):
"""返回代理对象背后的真实对象"""
# __local传入的是一个偏函数
if not hasattr(self.__local, '__release_local__'):
# 在这里我们修改一下源码
print("这里被执行了")
# 看一下这个对象里面保存的信息
print(self.__local().__dict__)
return self.__local()
try:
return getattr(self.__local, self.__name__)
except AttributeError:
raise RuntimeError('no object bound to %s' % self.__name__)
启动下面这个测试函数：
# 导入Flask
from flask import Flask, request
# 实例化Flask类
app = Flask(__name__)
# 注册路由
@app.route('/')
def index():
if request.method == 'GET':
return "hello world"
if __name__ == '__main__':
# 启动测试服务器
app.run()
浏览器访问根路径：
这里被执行了
{'method': 'GET', 'scheme': 'http', 'server': ('127.0.0.1', 5000), 'root_path': '', 'path': '/', 'query_string': b'', 'headers': EnvironHeaders([('Host', '127.0.0.1:5000'), ('User-Agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/100.0.4896.60 Safari/537.36'), ('Accept', 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9'), ('Accept-Encoding', 'gzip, deflate, br'), ('Accept-Language', 'en-US,en;q=0.9'), ('Cache-Control', 'max-age=0'), ('Connection', 'keep-alive'), ('Sec-Ch-Ua', '" Not A;Brand";v="99", "Chromium";v="100", "Google Chrome";v="100"'), ('Sec-Ch-Ua-Mobile', '?0'), ('Sec-Ch-Ua-Platform', '"Windows"'), ('Sec-Fetch-Dest', 'document'), ('Sec-Fetch-Mode', 'navigate'), ('Sec-Fetch-Site', 'none'), ('Sec-Fetch-User', '?1'), ('Upgrade-Insecure-Requests', '1')]), 'remote_addr': '127.0.0.1', 'environ': {'wsgi.version': (1, 0), 'wsgi.url_scheme': 'http', 'wsgi.input': <_io.BufferedReader name=652>, 'wsgi.errors': <_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>, 'wsgi.multithread': True, 'wsgi.multiprocess': False, 'wsgi.run_once': False, 'werkzeug.socket': <socket.socket fd=652, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 5000), raddr=('127.0.0.1', 58598)>, 'SERVER_SOFTWARE': 'Werkzeug/2.1.0', 'REQUEST_METHOD': 'GET', 'SCRIPT_NAME': '', 'PATH_INFO': '/', 'QUERY_STRING': '', 'REQUEST_URI': '/', 'RAW_URI': '/', 'REMOTE_ADDR': '127.0.0.1', 'REMOTE_PORT': 58598, 'SERVER_NAME': '127.0.0.1', 'SERVER_PORT': '5000', 'SERVER_PROTOCOL': 'HTTP/1.1', 'HTTP_HOST': '127.0.0.1:5000', 'HTTP_USER_AGENT': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/100.0.4896.60 Safari/537.36', 'HTTP_ACCEPT': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9', 'HTTP_ACCEPT_ENCODING': 'gzip, deflate, br', 'HTTP_ACCEPT_LANGUAGE': 'en-US,en;q=0.9', 'HTTP_CACHE_CONTROL': 'max-age=0', 'HTTP_CONNECTION': 'keep-alive', 'HTTP_SEC_CH_UA': '" Not A;Brand";v="99", "Chromium";v="100", "Google Chrome";v="100"', 'HTTP_SEC_CH_UA_MOBILE': '?0', 'HTTP_SEC_CH_UA_PLATFORM': '"Windows"', 'HTTP_SEC_FETCH_DEST': 'document', 'HTTP_SEC_FETCH_MODE': 'navigate', 'HTTP_SEC_FETCH_SITE': 'none', 'HTTP_SEC_FETCH_USER': '?1', 'HTTP_UPGRADE_INSECURE_REQUESTS': '1', 'werkzeug.request': <Request 'http://127.0.0.1:5000/' [GET]>}, 'shallow': False, 'url_rule': <Rule '/' (GET, OPTIONS, HEAD) -> index>, 'view_args': {}, 'host': '127.0.0.1:5000', 'url': 'http://127.0.0.1:5000/'}
返回了这么多信息，通过这些信息应该也能够猜到，这就是当前的Request对象
request.xxx 实际上会调用LocalProxy的_ _ getattr _ _方法：
再看LocalProxy中的_ _ getattr _ _方法：
class LocalProxy:
def __getattr__(self, name):
if name == '__members__':
return dir(self._get_current_object())
# 获取被代理对象的属性
return getattr(self._get_current_object(), name)
下面简单总结一下Flask的本地上下文：
7 路由规则
装饰器在什么时候执行？是在被装饰函数被调用时候执行还是程序编译后运行？flask的路由是通过装饰器实现的。
def decorator(f):
print('装饰器函数执行')
def inner():
f()
return inner
@decorator
def bar():
print('bar')
不调用任何函数，直接运行py文件，控制台输出“装饰器函数执行”，说明装饰器函数不用等待调用bar就会执行。
了解了装饰器的执行机制后，转过头来看app.route内部实现了什么
venv\lib\site-packages\flask\app.py
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)
return f
return decorator
通过装饰器函数发现，app.route()，在被装饰函数没有被调用的情况下，就执行了add_url_rule函数：
add_url_rule函数的作用是把路由字符串与视图函数一一对应。
8 session实现原理
未完待续