1.session
session存储了特定用户会话所需的属性及配置信息，这样，当用户在应用程序的 Web 页之间跳转时，存储在 session 对象中的变量将不会丢失，而是在整个用户会话中一直存在下去。当用户请求到来时，如果该用户还没有会话，则自动创建一个 session 。当会话过期或被放弃后，服务器将终止该会话。
session在flask已经被封装好了，直接from flask import session即可，这个session是一个<class 'werkzeug.local.LocalProxy'>对象，这个对象是什么，我们在后面介绍上下文的时候会说。
总之我们可以通过这个对象来操作session。
设置：session['xxx'] = 123
获取：session['xxx']
删除：del session['xxx']
这个session本质上是<class 'werkzeug.local.LocalProxy'>对象，这个对象继承了字典，也就是说我们可以按照字典的方法来操作session
关于flask中session原理，当用户第一次发送请求过来的时候，cookie里面实际上是空的，那么flask会为其创建一个session(<class 'werkzeug.local.LocalProxy'>对象)，或者就简单理解为空字典，来存储一些信息，比如用户名、密码等等，这些都是在内存里面操作的，如果需要就拿来用。当用户请求结束时，flask将这个字典进行序列化、加密然后写回用户的cookie里面去,如果不指定名字，那么这个cookie的名字就叫session，存储的值则是加密之后的结果。
当用户再次发送请求时，flask读取cookie中存储的key为session所对应的value，也就是加密的结果，进行解密然后反序列化得到字典，放入内存，以便于视图函数使用。
import secrets
from flask import Flask, session
app = Flask(__name__)
# session是根对字典进行序列化然后加密，所以必须要指定secret_key
app.secret_key = secrets.token_hex()
@app.route(r"/index")
def index():
session["xxx"] = 123
session["yyy"] = 456
del session["xxx"]
return "session"
if __name__ == "__main__":
app.run(port=8888, debug=True)
import secrets
from flask import Flask, session
app = Flask(__name__)
app.secret_key = secrets.token_hex()
# 修改session的名称
app.config["SESSION_COOKIE_NAME"] = "session_session"
@app.route(r"/index")
def index():
session["xxx"] = 123
session["yyy"] = 456
del session["xxx"]
return "session"
if __name__ == "__main__":
app.run(port=8888, debug=True)
模拟flash
import secrets
from flask import Flask, session
app = Flask(__name__)
app.secret_key = secrets.token_hex()
@app.route(r"/set")
def set_session():
session["xxx"] = 123
session["yyy"] = 456
return "session"
@app.route(r"/get")
def get_session():
return f"{session.get('xxx')} {session.get('yyy')}"
if __name__ == "__main__":
app.run(port=8888, debug=True)
import secrets
from flask import Flask, session
app = Flask(__name__)
app.secret_key = secrets.token_hex()
# 也可以设置session的过去时间，通过app.config指定
# app.config["PERMANENT_SESSION_LIFETIME"] = datetime.timedelta(hours=2)，两小时后过期
@app.route(r"/set")
def set_session():
session["xxx"] = 123
session["yyy"] = 456
# 也可以通过session设置
# session.permanent = True, 表示session默认为31天之后过期
return "session"
@app.route(r"/get")
def get_session():
# 当我们获取"xxx"我将其pop掉，这就意味着只能获取一次
xxx = session.pop("xxx")
yyy = session.get("yyy")
return f"{xxx} {yyy}"
if __name__ == "__main__":
app.run(port=8888, debug=True)
实际上flash也是这么做的，就是使用了类似于session.pop的方式
import secrets
from flask import Flask, session, flash, get_flashed_messages
app = Flask(__name__)
app.secret_key = secrets.token_hex()
@app.route(r"/set")
def set_session():
flash("临时数据存储", category="aa") # session['aa'] = "临时数据存储"
return "session"
@app.route(r"/get")
def get_session():
# 由于flash里面可以指定category，所以可以多指定几个
# 如果还有一个flash("", category="bb")
# 那么我们可以指定category_filter=["aa", "bb"]，因此返回的结果是一个列表。这里只有一个"aa"
s = get_flashed_messages(category_filter=["aa"]) # s = [session.pop("aa")]
return f"{s}"
if __name__ == "__main__":
app.run(port=8888, debug=True)
看一下flash和get_flashed_messages的源码
def flash(message, category="message"):
flashes = session.get("_flashes", [])
flashes.append((category, message))
session["_flashes"] = flashes
message_flashed.send(
current_app._get_current_object(), message=message, category=category
)
# 直接使用session['xx'] = xx的形式
def get_flashed_messages(with_categories=False, category_filter=()):
flashes = _request_ctx_stack.top.flashes
if flashes is None:
_request_ctx_stack.top.flashes = flashes = (
session.pop("_flashes") if "_flashes" in session else []
)
if category_filter:
flashes = list(filter(lambda f: f[0] in category_filter, flashes))
if not with_categories:
return [x[1] for x in flashes]
return flashes
# 本质上也是调用了session.pop()，并且session里面可以设置很多个键值对，最后返回一个列表
# 因此闪现实际上是对session的一种封装，在session中存储数据，读取的时候将其移除
2.中间件
我们可以看看flask的执行流程是怎么样的，我们知道flask只是一个简单的web框架，不具备路由映射等功能，这些是通过werkzeug这个wsgi工具包实现的，包括执行app.run()时启动的测试服务器。
我们来看看这个测试服务器
from werkzeug.serving import run_simple
from werkzeug.wrappers import Response, Request
@Request.application
def hello(request):
return Response("hello world")
if __name__ == '__main__':
# 只需要传入ip、端口、函数名即可，会自动对函数进行调用(关键)
run_simple("localhost", 8888, hello)
以上便可以启动这个测试服务器，当然flask内部使用的也是这个测试服务器，我们可以看看源码。首先app是我们Flask对象，run方法是在类Flask里面定义的一个方法
我们来看一下这个run方法
可以看到，run方法还是调用run_simple，启动我们之前说的测试服务器。但是我们注意到第三个参数是self，也就是把Flask类的实例对象app传进去了，刚才我们手动使用run_simple时说过第三个参数会自动进行调用，我们只需要传入函数名即可。而源码中传入的是self，说明self是可以调用的，这就意味着Flask这个类的内部一定实现了__call__方法
可以看到最终调用了wagi_app方法，里面的environ，是请求相关的所有数据。而start_response则是设置响应相关的数据
因此我们可以得出flask的app.run的执行流程，先执行run_simple，开启监听，一旦请求过来,再执行__call__，然后执行wsgi_app，那么我们可以在wsgi_app函数之上做操作。为了不修改源码，我们换种方式操作
from flask import Flask
app = Flask(__name__)
@app.route(r"/index")
def index():
return "index"
class MiddleWare:
def __init__(self, old_func):
self.old_func = old_func
def __call__(self, *args, **kwargs):
print("请求来了")
res = self.old_func(*args, **kwargs)
print("请求走了")
return res
if __name__ == "__main__":
"""
我们让app.wsgi_app方法等于MiddleWare(app.wsgi_app)，
此时app.wsgi_app就相当于MiddleWare的一个实例
那么当调用 app.wsgi_app，实际上调用的是MiddleWare的__call__方法
而原来的app.wsgi_app就是MiddleWare实例的old_func，因此请求依旧会成功执行，但是我们加上了自己的逻辑
"""
app.wsgi_app = MiddleWare(app.wsgi_app)
app.run(port=8888, debug=True)
这种方法基本不用，但是对理解flask的执行很有帮助。
3.特殊装饰器
还记得之前说的before_request和after_request吗？前者不需要参数和返回值，后者需要一个参数且必须有返回值。
但是如果有多个before_request和after_request呢，它们的执行顺序是怎么样的呢？
插一嘴，可以把特殊装饰器理解为之前说的钩子函数
from flask import Flask
app = Flask(__name__)
@app.before_request
def x1():
print("x1")
@app.before_request
def x2():
print("x2")
@app.after_request
def y1(response):
print("y1")
return response
@app.after_request
def y2(response):
print("y2")
return response
@app.route(r"/index")
def index():
return "index"
if __name__ == "__main__":
app.run(port=8888, debug=True)
怎么来理解这个顺序呢？来看一张图
另外如果在x1就返回了那么，x2和视图函数不会执行，但是两个after_request依旧会执行。在Django<=1.9的时候，也是这么玩的。但是在Django1.10之后，就不会执行y2，而是只执行y1然后返回，这是Django内部的一个调整
因此如果想实现类似Django中间件的效果，使用before_request和after_request就可以了
4.什么是wsgi
web服务网关接口，是一个协议。
实现该协议的模块有：
wsgiref：python内置的
werkzeug
实现协议的模块本质上就是socket server，用于接收用户的请求，并处理。这类模块只专注于处理请求，对于业务类的操作会交给web框架。一般web框架基于实现wsgi协议的模块，因此这样就实现了关注点分离。对于werkzeug这种实现了wsgi协议的模块，只专注于请求的处理，而web框架则是根据到来的请求执行业务逻辑
因此我们app下面有一个url_map，还记得吗？自定义转换器的时候，app.url_map.converters的时候，你就可以认为url_map除此之外还保存了一系列的路由与视图函数之间的关系。
[
("/index", index),
("/login", login),
("/signin", signin)
]
werkzeug则是进行路由映射，判断是哪个路由，然后web框架根据请求执行相应的视图
5.路由和视图剖析
@app.route(r"/index")
def index():
return "index"
我们来看看这段代码都做了些什么？首先核心在于app.route(r"/index")，我们看看这个route的源码
app.route
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
# route是一个装饰器
# 当我调用@app.route(r"/index")的时候
# 相当于decorator = app.route(r"/index")，然后@decorator
# 然后调用执行decorator(index),endpoint我们不管，这里显然是None，因为我们在route当中没有传
# 然后调用了self.add_url_rule将我们的url和函数给添加了进去，然后返回函数本身，也就是我们的index函数
# 很明显，核心就是这个self.add_url_rule
app.add_url_rule
def add_url_rule(
self,
rule,
endpoint=None,
view_func=None,
provide_automatic_options=None,
**options
):
# 函数定义如下，很明显装饰器的模式，本质上还是调用了这个函数
from flask import Flask
app = Flask(__name__)
def index():
return "index"
# endpoint如果不指定，默认是函数名，但是endpoint和view_func至少要指定一个
app.add_url_rule(r"/index", view_func=index, endpoint="index")
if __name__ == "__main__":
app.run(port=8888, debug=True)
依旧是可以访问成功的
from flask import Flask
app = Flask(__name__)
def index():
return "index"
app.add_url_rule(r"/index", endpoint="index")
# 如果不指定函数，那么仍然可以通过endpoint添加
app.view_functions["index"] = index
# 因此从这里也能看出endpoint和函数是一对一的
if __name__ == "__main__":
app.run(port=8888, debug=True)
实现类似tornado中的路由
from flask import Flask
app = Flask(__name__)
def index_handler():
return "index"
def login_handler():
return "login"
def fuck_handler():
return "fucker"
application = [
(r"/index", index_handler),
(r"/login", login_handler),
(r"/fuck", fuck_handler)
]
for url_func in application:
app.add_url_rule(url_func[0], view_func=url_func[1], endpoint=url_func[1].__name__)
if __name__ == "__main__":
app.run(port=8888, debug=True)
app.route里面的更多参数
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
# 参数都在这个options里面
# 参数一：rule：url规则，也就是我们的"/index"
# 参数二：view_func：视图函数
# 参数三：endpoint
# 参数四：methods：请求方法
# 以上四个是常用的
# 参数五：defaults，这个是干什么的
@app.route("/index", defaults={"name": "satori", "age": 18})
def index(name, age):
print(name, age) # satori 18
return "index"
# 参数六：strict_slashes：对结尾的/是否严格要求。就拿我们的这个路由为例，如果我们访问localhost:8888/index可以访问，但如果访问localhost:8888/index/就会报错，这在Django中两种方式都是可以的。如果在flask中也想这样的话，至于要加上strict_slashes=False即可，表示对/不严格要求，默认是True
# 参数六：redirect_to：重定向。如果一个url废弃了，但是呢？又怕用户不知道，所以可以采取重定向，当用户访问"/index/1"的时候，会自动跳转到"/new/1",这是有参数的情况，当然也可以没有参数
@app.route("/index/<int:nid>", redirect_to="/new/<nid>")
def index(name, age):
return "index"
@app.route("/new/<nid>")
def new():
return "new"
# 参数七：subdomain：子域名。当子域名为web的时候才会触发。当然这个需要改hosts文件，比如将127.0.0.1映射为satori.com,然后进行配置app.config['SERVER_NAME'] = 'satori.com:8888',不加subdomain的时候，那么访问http://www.satori.com:8888会触发。如果加上了，那么下面只有当你访问http://web.satori.com:8888才会触发
@app.route("/index", subdomain="web")
def index():
return "index"
@app.route("/index", subdomain="<xxx>")
def index():
return xxx + "index"
"""
这个<>表示里面可以传入任何字符串，所以一定要放在最下面
"""
CBV
之前我们定义视图的时候，定义的都是一个函数，至于判断是什么请求，则是通过request.method来判断，而实际上flask也支持视图类
from flask import Flask, views
app = Flask(__name__)
class IndexView(views.MethodView):
"""定义一个类，要继承自views下的MethodView
然后重写get和post方法，当get请求到来执行get，post请求到来执行post
"""
def get(self):
return "这是一个get请求"
def post(self):
return "这是一个post请求"
# 可以看到有点类似tornado
# 同样需要注册路由，并且如果是类视图，那么就不能通过装饰器的方式了
# view_func传入我们的 类.as_view(endpoint)
app.add_url_rule("/index", view_func=IndexView.as_view("index"))
if __name__ == "__main__":
app.run(port=8888, debug=True)
import requests
print(requests.get("http://localhost:8888/index").text) # 这是一个get请求
print(requests.post("http://localhost:8888/index").text) # 这是一个post请求
"""可以看到以上逻辑的实现是没有问题的，但是flask是怎么做到的呢？
熟悉Django的应该知道，有一个dispatch，实际上再flask里面也是这么做的，我们来看一看源码
"""
# 下面就是views.py中把注释去掉之后的全部内容，很短。
from ._compat import with_metaclass
from .globals import request
http_method_funcs = frozenset(
["get", "post", "head", "options", "delete", "put", "trace", "patch"]
)
class View(object):
# 后面说
methods = None
provide_automatic_options = None
# 在CBV中使用装饰器，后面说
decorators = ()
def dispatch_request(self):
raise NotImplementedError()
# 我们的IndexView继承的MethodView实际上还继承了这个view，调用的as_view实际上就是这个View类下的as_view方法。
# 这是一个类方法，所以我们可以直接通过类来调用
# 我们看看它返回了什么，返回了view，没错，as_view里面定义了一个view函数
# 所以我们在CBV中，view_func最终返回的还是一个函数
# as_view里面的name参数就是我们传过去之前传过去的"index"
@classmethod
def as_view(cls, name, *class_args, **class_kwargs):
def view(*args, **kwargs):
self = view.view_class(*class_args, **class_kwargs)
# 这里调用了dispatch_request方法
# 我们看看dispath_request，这里面没有，那么应该在MethodView里面
return self.dispatch_request(*args, **kwargs)
if cls.decorators:
view.__name__ = name
view.__module__ = cls.__module__
for decorator in cls.decorators:
view = decorator(view)
view.view_class = cls
# 这里将函数名设置成了我们传过来的name
# 所以不用想也能猜到，我们传过来的name就是最终的endpoint
# 因为IndexView.as_view("index")返回了这里定义的view函数
# 因此我们调用add_url_rule("/index",view_func=IndexView.as_view("index"))就等价于
# add_url_rule("/index", view_func=view),而如果不指定endpoint，那么endpoint默认是函数名
# 而通过view.__name__ = name,又将函数名设置成了name,也就是我们传的"index"
# 因此，我们在as_view里面传入的"index"为什么是endpoint，就是这个原因
view.__name__ = name
view.__doc__ = cls.__doc__
view.__module__ = cls.__module__
view.methods = cls.methods
view.provide_automatic_options = cls.provide_automatic_options
return view
class MethodViewType(type):
# 这里定义的是一个元类，顾名思义就是生成类的类
# 这里我们先不管它，看看下面的我们继承的MethodView
def __init__(cls, name, bases, d):
super(MethodViewType, cls).__init__(name, bases, d)
if "methods" not in d:
methods = set()
for base in bases:
if getattr(base, "methods", None):
methods.update(base.methods)
for key in http_method_funcs:
if hasattr(cls, key):
methods.add(key.upper())
if methods:
cls.methods = methods
class MethodView(with_metaclass(MethodViewType, View)):
# 果然里面定义了dispatch_request方法
def dispatch_request(self, *args, **kwargs):
# 使用反射获取我们定义的类里面相应的方法，request.method.lower()
# 所以如果是get请求执行get函数，post请求执行post函数
meth = getattr(self, request.method.lower(), None)
if meth is None and request.method == "HEAD":
meth = getattr(self, "get", None)
assert meth is not None, "Unimplemented method %r" % request.method
# 执行逻辑
return meth(*args, **kwargs)
可以看到flask中的CBV就是这么干的，那么刚才的MethodViewType是干什么的呢？举个例子
class IndexView(views.MethodView):
# 如果我定义了
methods = ["GET"] # 代表只能进行get请求，post请求发不过来
def get(self):
return "这是一个get请求"
def post(self):
return "这是一个post请求"
# 因为我们定义的IndexView继承了MethodView，而MethodView是由MethodViewType生成的
# 那么我们自己定义的IndexView也是有MethodViewType这个元类生成的。
class MethodViewType(type):
def __init__(cls, name, bases, d):
super(MethodViewType, cls).__init__(name, bases, d)
# 这里给我们进行了逻辑上的判断，这是在生成IndexView这个类的时候就已经判断了
# 如果d就是我们IndexView的所有属性，如果methods不在d里面的话，默认请求都支持。
# 但如果在呢？我们再回到View这个类
if "methods" not in d:
methods = set()
for base in bases:
if getattr(base, "methods", None):
methods.update(base.methods)
for key in http_method_funcs:
if hasattr(cls, key):
methods.add(key.upper())
if methods:
cls.methods = methods
class View(object):
# methods初始是为None的
# 如果我们methods一开始定义了
methods = None
provide_automatic_options = None
decorators = ()
...
...
...
view.__name__ = name
view.__doc__ = cls.__doc__
view.__module__ = cls.__module__
# 这里的cls，指的是IndexView，如果有的话就就是用自己的methods
# 最终将methods赋值给view.methods
view.methods = cls.methods
view.provide_automatic_options = cls.provide_automatic_options
以上就是CBV的实现逻辑，此外CBV也可以使用装饰器，当然不是那么app.route，而是用户认证之类的装饰器，当然使用是有特殊方法的。还记得View里面的两个属性吗？methods(已经介绍了)和decorators吗？这个decorators就是帮我们使用装饰器的
def deco(func):
@wraps(func)
def inner(*args, **kwargs):
if request.args.get("username") == "satori":
return func(*args, **kwargs)
else:
return "<h1>fuck off</h1>"
return inner
class IndexView(views.MethodView):
# 如果我定义了
methods = ["GET"] # 代表只能进行get请求，post请求发不过来
# 怎么使用装饰器呢,这样使用即可
# 注意名字必须叫decorators，因为从源码我们也能看出来，源码里面的就叫这个名字
# 既然是列表，说明可以放入多个装饰器
decorators = [deco]
def get(self):
return "这是一个get请求"
def post(self):
return "这是一个post请求"
6.session执行流程源码初探
我们之前看过app.run的执行流程，但看到wsgi_app就停下来了，我们下面继续简单剖析一下
def __call__(self, environ, start_response):
# environ：与请求相关的所有数据(做了初步的整理、封装)
"""
有一点必须要清楚，请求来的是最原始的数据，werkzeug则是负责进行一些简单的整理
使得请求不再那么原始了，但仍然还是有那么一点原始
这第一步处理之后的就是也就是werkzeug进行整理、封装之后environ
但我能通environ来获取请求头、方法、参数之类的吗？
显然是可以的，只是比较麻烦。因为werkzeug只做了初步的封装
于是接下面又对其进行了更进一步的整理，封装成了request对象，使得我们可以通过
request.headers:获取请求头
request.method:获取请求方法
request.path:获取请求路径
request.args|form:获取请求|表单内容
request.cookies:获取cookie
。。。
等等
"""
# start_response：用于设置响应相关的数据
return self.wsgi_app(environ, start_response)
def wsgi_app(self, environ, start_response):
# 先不要管这第一句代码具体是做什么的？目前只需要知道这行代码完成了重要的功能
# 这行代码就是对我们传进去的environ做了进一步的封装，怎么封装呢
# 说白了就是把werkzeug初步处理之后environ传入request_context里面进行封装，得到request对象
# 然而完成的并不止这些功能，总结一下就是：
"""
1.获取environ并对其再次封装，得到request
2.从request中获取名称为session的cookie，然后解密，反序列化，session就是在这里获取的
3.然后将request和session存在某一个位置。这里提一下，Django的请求是通过参数传递的，在Django中定义视图的时候会有一个request参数，直接通过这个参数去获取请求相关的数据；而flask的request和session显然是内置的，需要导入，说明请求到来时，flask经过两层封装先获取请求和session，封装为request对象和session对象，保存起来。当使用的时候，直接from flask import request, session去取，然后使用即可。
"""
ctx = self.request_context(environ)
error = None
try:
try:
ctx.push()
# 4.执行视图函数，一旦触发，就可以去取数据了。取到之后，就可以对session做操作
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except:
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
"""
当响应结束时
5.获取session，序列化、加密，写入cookie
6.清空
"""
ctx.auto_pop(error)
# flask源码的执行就是在这个wsgi_app函数里面触发
7.session执行流程源码深入分析
def wsgi_app(self, environ, start_response):
# 可以看到这里的request_context(environ)，实际上返回的是一个RequestContext(self, environ)
# 这个RequestContext显然是一个类(源码在下面)，self就是我们的app，environ就是请求相关的数据
# ctx = RequestContext(app, environ)
ctx = self.request_context(environ)
# 从RequestContext类中，我们可以看到，如果没有传入request，默认是None
# 如果request is None，那么request = app.request_class(environ)
# self.request = request，因此我们是直接可以通过ctx.request获取的。
# 那么app.request_class(environ)是个什么东西嘞？
# 实际上是一个Request对象，用来接收environ，对其进一步封装的
# self.request = Request(environ)，那么此时的self.request则是最终处理之后的request
# 可以进行args、headers、form等等
# 然后我们在RequestContext源码中还可以看到self.session = session
# 只不过这个session一开始是一个空值，按理说不应该为空的。
# 但是别急，只是最开始为空，然后下一步就该获取名称为session的cookie值了。
# 但是此时ctx里面已经封装了两个值了，一个请求相关的数据request，一个是目前为空的session
# 如果现在把ctx存起来，相当于把request和session存起来了
# 如果存起来之后，那么我们获取session实际上是为空的，这就意味着ctx存起来之后，会对session再次加工
error = None
try:
try:
# 执行ctx.push()
# 我们来看看这个push(),源码在下面
# 通过观察push的源码，我们发现
# 我们将ctx保存起来之后，request已经被封装好了，但是session还是空的(或者说为None)
# 然后调用ctx.push()，内部执行session_interface.open_session(self.app, self.request)
# 通过request获取名称为'session'的cookie，如果没有就手动生成一个初始的空字典
# 然后将值重新赋给ctx里面的session
ctx.push()
# 此时再往下走，执行下面的代码的时候，ctx里面的request和session已经全部都有值了
# 我们可以对其执行任意的操作
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except:
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 request_context(self, environ):
return RequestContext(self, environ)
class RequestContext(object):
def __init__(self, app, environ, request=None, session=None):
self.app = app
# 创建request
if request is None:
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
# 创建session
self.session = session
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()
# 上面的不用看，这一步就等于说把ctx给存起来了
_request_ctx_stack.push(self)
# 如果session是空的，此时session确实是空的
if self.session is None:
# 然后执行这一行，这个self.app.session_interface是啥嘞？
# 这是app下面的，里面定义了session_interface = SecureCookieSessionInterface()
# 说明这个session_interface是一个对象，然后执行这个对象下的open_session方法
# 然后将返回值赋值给self.session，相当于是给为None的session重新赋值
# 那么session_interface.open_session这一步做了什么很容易就明白了
# 就是获取名称为session的cookie，然后解密、反序列化然后赋值给self.session
# 那么这一步是如何实现的呢？是怎么获取的呢？
# 显然我们要看一下session_interface.open_session的源码，源码在下面
session_interface = self.app.session_interface
# 因此这个self.session就是通过session_interface.open_session(self.app, self.request)实现的
# 再回到wsgi_app函数的源码中
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()
class SecureCookieSessionInterface(SessionInterface):
...
...
...
# 之前说过，这个request就是对environ进一步封装之后的结果
# 里面存储了请求相关的信息，当然也包括cookie
def open_session(self, app, request):
s = self.get_signing_serializer(app)
if s is None:
return None
# 从request.cookies里面获取名称为app.session_cookie_name的cookie
# session_cookie_name = ConfigAttribute("SESSION_COOKIE_NAME")
# 这个默认值就是"session"，如果我们不指定的话，当然指定了就获取我们所指定的cookie值。
val = request.cookies.get(app.session_cookie_name)
# 当用户第一次请求，也就是第一次获取的时候，肯定是没有值的
if not val:
# 可以把self.session_class()理解为一个空字典，直接就返回了
# 我们拿到的session就是一个空字典，我们可以对session进行设置值
# 比如session['username'] = 'satori',session['password'] = 'aaa123456'
# 当请求结束之后，此时session不再是空字典了
# 然后对session进行序列化、加密，写到用户的名称为'session'的cookie里面去
return self.session_class()
max_age = total_seconds(app.permanent_session_lifetime)
try:
# 走到这里说明val不为None，说明不是第一次获取，意味着用户不是第一次发送请求
# 那么在第一次的时候，我们已经为session(空字典)设置了一些值
# 这里的s.loads是在对val(密串)进行解密、反序列化
data = s.loads(val, max_age=max_age)
# 将data传进去得到了一个有值的字典，{'username': "satori", 'password': 'aaa123456'}
# 我们可以对其进行一些操作，用验证用户是否登陆等等。
# 如果确实正确，并且session还没有过期，就不用再二次登陆了
# 回到push函数的源码中
return self.session_class(data)
except BadSignature:
return self.session_class()
以上就是flask中读取session的执行流程，但是我们注意到这个request和session都是放在ctx里面的，我们想用的话就拿来取。但是怎么获取呢？from flask import request, session即可，为什么能通过这种机制获取，flask中又是怎么实现的呢？我们在后续的flask上下文管理中继续分析源码。
那么flask是如何写入cookie的呢？
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:
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):
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)
def finalize_request(self, rv, from_error_handler=False):
response = self.make_response(rv)
try:
# 执行了这一步，善后工作就是在这里触发的，看源码
response = self.process_response(response)
request_finished.send(self, response=response)
except Exception:
if not from_error_handler:
raise
self.logger.exception(
"Request finalizing failed with an error while handling an error"
)
return response
def process_response(self, response):
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):
# 最终调用了save_session
self.session_interface.save_session(self, ctx.session, response)
return response
def save_session(self, app, session, response):
domain = self.get_cookie_domain(app)
path = self.get_cookie_path(app)
if not session:
if session.modified:
response.delete_cookie(
app.session_cookie_name, domain=domain, path=path
)
return
if session.accessed:
response.vary.add("Cookie")
if not self.should_set_cookie(app, session):
return
httponly = self.get_cookie_httponly(app)
secure = self.get_cookie_secure(app)
samesite = self.get_cookie_samesite(app)
expires = self.get_expiration_time(app, session)
val = self.get_signing_serializer(app).dumps(dict(session))
# 通过set_cookie的方式将session写到用户的名称为app.session_cookie_name(默认为session)的cookie里面去
response.set_cookie(
app.session_cookie_name,
val,
expires=expires,
httponly=httponly,
domain=domain,
path=path,
secure=secure,
samesite=samesite,
)
所以核心就是open_session和save_session,请求进来的时候执行open_session，请求走的时候执行save_session。但这两个方法是SecureCookieSessionInterface类下，我们是不是也可以将其进行修改呢？这样就能执行我们自己定制的open_session和save_session方法。比如默认情况下，session是放到cookie面的，我们可不可以放到redis里面呢？显然是可以的，而且已经有第三方组件实现了，目前不用关心。
以上就是flask中request、session源码的执行流程，对于我们理解flask是很有帮助的
8.蓝图
蓝图，和我们平常说的宏伟蓝图的蓝图有区别，flask的蓝图是为了将解耦，我们之前的所有程序都写在一个py文件里面，然而一旦程序代码多了，就显得很乱，所以需要分层解耦。
book.py
from flask import Blueprint
# 第一个参数是py文件名,这个时候就不能使用Flask了
book_app = Blueprint("book", __name__)
@book_app.route("/book/gone")
def gone():
return "gone"
@book_app.route("/index")
def index():
return "index"
movie.py
from flask import Blueprint
# 如果加上了url_prefix，那么在访问的时候必须要加上url_prefix
movie = Blueprint("movie", __name__, url_prefix="/movie")
@movie.route("/kiminonawa")
def kimi():
return "你的名字"
from flask import Flask
# 将蓝图导入进来
from bl.movie import movie_app
from bl.book import book_app
app = Flask(__name__)
# 然后注册到app里面去
app.register_blueprint(movie_app)
app.register_blueprint(book_app)
@app.route("/index")
def index():
return "app_index"
if __name__ == "__main__":
app.run(port=8888, debug=True)
可以看到，蓝图实现了视图的分离。不同的版块，对应的视图定义在不同的蓝图里。但是我们发现蓝图里面有一个"/index",而我们的主app里面也有"/index"，那么当我访问localhost:8888/index的时候，会执行哪一个视图呢？
答案是会执行蓝图里面的视图
9.蓝图中模板文件寻找规则
首先如果我们不指定蓝图中的模板目录的话
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<p>我是整个项目的templates目录中的模板文件</p>
<p>你猜蓝图中的视图能否渲染到我呢？</p>
</body>
</html>
from flask import Blueprint, render_template
book_app = Blueprint("book", __name__)
@book_app.route("/book/gone")
def gone():
return render_template("模板文件.html")
@book_app.route("/index")
def index():
return "index"
答案是可以找到的
但如果我不想在主项目的templates目录中找呢？只需要单独指定目录即可。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<p>我是666目录下的模板文件</p>
</body>
</html>
from flask import Blueprint, render_template
# 同样可以指定文件夹路径，指定为666，则默认是和当前py文件同级的目录下的666目录
book_app = Blueprint("book", __name__, template_folder="666")
@book_app.route("/book/gone")
def gone():
return render_template("模板文件.html")
@book_app.route("/index")
def index():
return "index"
咦，什么鬼？为什么访问的还是主项目对应templates目录下的模板文件，这里因为对于蓝图来说，即便指定了目录，也还是会先到主项目对应的templates目录下寻找模板文件，找不到的话，再去蓝图中指定的模板目录去找。而我们在666目录下定义的也叫"模板文件.html"，和主项目对应的templates目录下的"模板文件.html"重名了，如果我将其删掉的话。
再次访问的话
就会渲染666目录下的"模板文件.html"了，因此在蓝图中指定的模板目录下的模板文件，一定不要和主项目对应的模板目录下的模板文件重名
10.蓝图中静态文件寻找规则
这个比较简单，因为我们在指定的时候，可以通过蓝图的名字来指定。
from flask import Blueprint, render_template
# 同样可以指定静态文件路径
book_app = Blueprint("book", __name__, template_folder="666", static_folder="static")
@book_app.route("/book/gone")
def gone():
return render_template("模板文件.html")
@book_app.route("/index")
def index():
return "index"
蓝图下的static
h2{
color: red;
}
主项目目录下的static
h2{
color: green;
}
蓝图中的templates
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
<link rel="stylesheet" href='{{ url_for('static',filename="1.css") }}'>
</head>
<body>
<h2>我是666目录下的模板文件,我会被哪里的css渲染呢？</h2>
</body>
</html>
可以看到，被主项目目录中的static下的css渲染了，如果都叫static，那么不会找蓝图下的static，所以名字不能一样。
修改一下
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
<link rel="stylesheet" href='{{ url_for('bl_static',filename="1.css") }}'>
</head>
<body>
<h2>我是666目录下的模板文件,我会被哪里的css渲染呢？</h2>
</body>
</html>
这里我们把static改成了bl_static
from flask import Blueprint, render_template
# 将static改成bl_static
book_app = Blueprint("book", __name__, template_folder="666", static_folder="bl_static")
@book_app.route("/book/gone")
def gone():
return render_template("模板文件.html")
@book_app.route("/index")
def index():
return "index"
结果在渲染的时候，找不到bl_static，可是我们明明指定了。
这是因为bl_static是在蓝图中，我们必须要指定名字。
这个bl_static是在哪一个蓝图中被定义的，是在book.py中，所以要通过book.static去找。
这里又会好奇了，为什么不是book.bl_static呢？因为flask就是这么设计的，book.static的意思就是
通过蓝图为book所指定的静态文件目录，而我们指定的是bl_static，所以就会去bl_static目录去找。
如果这里我写成book.bl_static的话
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
<link rel="stylesheet" href='{{ url_for('book.bl_static',filename="1.css") }}'>
</head>
<body>
<h2>我是666目录下的模板文件,我会被哪里的css渲染呢？</h2>
</body>
</html>
会很智能的提示我们，你是不是想输入book.static呢？
我们改一下
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
<link rel="stylesheet" href='{{ url_for('book.static',filename="1.css") }}'>
</head>
<body>
<h2>我是666目录下的模板文件,我会被哪里的css渲染呢？</h2>
</body>
</html>
渲染成功了
因此可以看到，我们在名为book的蓝图中指定了静态目录为bl_static,那么模板文件中通过book.static就会去寻找book的蓝图中指定的静态目录。book蓝图中指定的静态目录是bl_static，那么渲染的时候就会去找bl_static目录下的静态文件。
这里值得一提的是，不要让蓝图下的静态目录和主项目对应静态目录重名，如果重名了，即便是book.static，依旧回去主项目对应的static目录里面去找。
这里把静态目录指定为绝对路径
import os
BASE_DIR = os.path.dirname(__file__)
book_app = Blueprint("book", __name__, template_folder="666", static_folder=os.path.join(BASE_DIR, "static"))
如果蓝图中的静态文件目录和主项目对应的静态文件目录都叫static的话，即使我这里指定的是绝对路径，明确表明了，不是某个static，而是和蓝图(或者说那个book.py)同级的static，但由于重名，还是会去主项目对应的静态目录里面去找。
from flask import Blueprint, render_template
import os
BASE_DIR = os.path.dirname(os.path.abspath( __file__))
book_app = Blueprint("book", __name__, template_folder="666", static_folder=os.path.join(BASE_DIR, "static"))
@book_app.route("/book/gone")
def gone():
return render_template("模板文件.html")
@book_app.route("/index")
def index():
return "index"
如果我把名字改掉
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
<link rel="stylesheet" href='{{ url_for('book.static',filename="1.css") }}'>
</head>
<body>
<h2>我是666目录下的模板文件,我会被哪里的css渲染呢？</h2>
</body>
</html>
from flask import Blueprint, render_template
import os
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
book_app = Blueprint("book", __name__, template_folder="666", static_folder=os.path.join(BASE_DIR, "static1"))
@book_app.route("/book/gone")
def gone():
return render_template("模板文件.html")
@book_app.route("/index")
def index():
return "index"
因此不要让蓝图下的静态目录和主项目对应的目录重名，然后寻找的时候，通过 蓝图.static来指定
11.url_for翻转蓝图
老规矩，我想项目的主入口，app.py中获取其url，如果两个方法都在主程序中，毫无疑问一个url_for("gone")就搞定了，但是现在其中一个方法在蓝图中，还可以这样吗？我们来试一下
from flask import Flask, url_for
# 将蓝图导入进来
from bl.movie import movie_app
from bl.book import book_app
app = Flask(__name__)
# 然后注册到app里面去
app.register_blueprint(movie_app)
app.register_blueprint(book_app)
@app.route("/hello")
def hello():
return url_for("gone")
if __name__ == "__main__":
app.run(port=8888, debug=True)
提示我们找不到endpoint 'gone'，问我们是不是要找book.gone。改一下，再走一个
from flask import Flask, url_for
# 将蓝图导入进来
from bl.movie import movie_app
from bl.book import book_app
app = Flask(__name__)
# 然后注册到app里面去
app.register_blueprint(movie_app)
app.register_blueprint(book_app)
@app.route("/hello")
def hello():
return url_for("book.gone")
if __name__ == "__main__":
app.run(port=8888, debug=True)
获取成功，其实也不难理解。如果我们多个蓝图都定义gone这个视图函数，或者说endpoint都叫gone的话，那么如果不指定相应蓝图名的话， 那么到底要找哪一个endpoint为gone的路由呢？
那么如果在同一个蓝图中翻转的话呢？
from flask import Blueprint, url_for
import os
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
book_app = Blueprint("book", __name__, template_folder="666", static_folder=os.path.join(BASE_DIR, "static1"))
@book_app.route("/book/gone")
def gone():
return "gone"
@book_app.route("/index")
def index():
return url_for("gone")
可以看到依旧会报错，原因和上面的是一样的。如果想获取其他蓝图里面的gone呢？所以必须要带上蓝图名。但如果是项目的主入口呢？
不用想也知道，肯定不需要指定蓝图名，因为根本就没有蓝图
from flask import Flask, url_for
# 将蓝图导入进来
from bl.movie import movie_app
from bl.book import book_app
app = Flask(__name__)
# 然后注册到app里面去
app.register_blueprint(movie_app)
app.register_blueprint(book_app)
@app.route("/hello")
def hello():
return "hello"
@app.route("/get")
def get():
# 这里如果不指定蓝图····，貌似也没法指定蛤，因为这里根本没有蓝图
return url_for("hello")
if __name__ == "__main__":
app.run(port=8888, debug=True)
是可以访问成功的
12.Local线程隔离对象
还记得python的多线程吗？python中的多线程实际上一个伪多线程，因为它不能同时利用多核，只有在高io的时候，才会有作用。而且多个线程之间是共享全局变量的，所以我们在使用多线程的时候，如果要对同一个全局变量修改的时候，一定要加锁。
import threading
import time
num = 0
def task(i):
global num
num = i
time.sleep(1)
print(num)
for i in range(10):
threading.Thread(target=task, args=(i, )).start()
"""
9
9
9
9
9
9
9
9
9
9
"""
为什么打印的都是9，因为在print(num)之前，num被修改了，而num是全局的，所有的子线程修改的同一份num，打印的也是同一份num。因此由于sleep了一秒，导致在打印之前已经修改完了，所以按照循环顺序，打印的都是9
再来看一个神奇的Local
import threading
import time
num = 0
local = threading.local()
def task(i):
global num
local.num = i
num = i
time.sleep(1)
print(local.num, num)
for i in range(10):
threading.Thread(target=task, args=(i, )).start()
"""
1 9
2 9
0 9
5 9
4 9
6 9
3 9
8 9
7 9
9 9
"""
# 尽管打印的顺序不一样，这是由于操作系统调度的线程的先后顺序决定的，但是至少local上的数据是不重复的
# 因此绑定在local上的变量，尽管都叫num，但他们是相互隔离的。
# 怎么做到的呢？每个线程都会有一个属于自己的id，解释会根据id单独分配一块独立的空间，然后存储数据
# 所以local上的数据是隔离的
那么其内部是如何实现的呢？其实使用了类似于字典的方式
import threading
import time
d = {}
def task(i):
"""
{
"id1": {"num": 1},
"id2": {"num": 2},
}
"""
# 每一个线程都有自己唯一的id
ident = threading.get_ident()
if ident in d:
d[ident]["num"] = i
else:
d[ident] = {}
d[ident]["num"] = i
time.sleep(1)
print(d[ident]["num"], i)
for i in range(10):
threading.Thread(target=task, args=(i, )).start()
"""
0 0
2 2
1 1
3 3
6 6
4 4
8 8
7 7
9 9
5 5
"""
然鹅目前的版本还是有点low，我们也成类的模式
import threading
import time
class Local:
d = {}
def __getattr__(self, item):
ident = threading.get_ident()
return self.d[ident].get(item)
def __setattr__(self, key, value):
ident = threading.get_ident()
if ident in self.d:
self.d[ident][key] = value
else:
self.d[ident] = {}
self.d[ident][key] = value
local = Local()
def task(i):
local.num = i
time.sleep(1)
print(local.num)
for i in range(10):
threading.Thread(target=task, args=(i, )).start()
"""
0
3
1
5
2
4
7
6
9
8
"""
flask里面Local也是类似的原理，from werkzeug.local import Local，和我们自定制的Local一样使用即可
为什么要有Local呢？因为每一个请求过来，我们都会将ctx(里面包含了request和session)放在某一个地方，方便之后去取。多个请求过来的时候，要开启多个线程去响应，如果不做数据隔离的话，那么不就乱套了吗？第二个请求的数据就会把第一个请求的数据给覆盖。所以每个请求过来的时候，都会为其创建一个单独的Local空间，每个空间是隔离的，当视图函数响应时，会根据线程的id去取数据。而Django里面是没有Local对像的，为什么没有？之前说过，Django的数据是通过参数传递的，这样的根据参数去处理请求时不会发生混乱的，但是flask不一样，是先根据请求把相应参数存起来，然后再去取，所以要做数据隔离，不然会发生混乱。
13.上下文管理基本流程
之前我们说过请求到来的时候，会创建一个RequestContext对象，创建出来的对象叫ctx，里面有request和session两个属性，然后会把ctx给存起来，存在某个地方，以后视图函数去取，这就叫上下文管理。
请求到来的时候：
调用wsgi_app,创建RequestContext对象
ctx = RequestContext(self, environ) # self是app，environ请求相关的、经过werkzeug处理的半原始数据
ctx.request = Request(environ)
ctx.session = None
然后对session做进一步处理，通过ctx.push，调用SecureCookieSessionInterface()创建一个实例session_interface,然后调用内部open_session方法，读取request.cookies里面的名称为session的cookie，解密反序列化得到session，如果是第一次请求没有的话，就创建一个空的，最后赋值给session，等于说session就是一个变量，指向一个类似于字典的东西。此时ctx里面的request和session都已经有值了，可以随意操作了
然后将包含了request和session的ctx对象放到某个地方，这个地方就可以理解为一个大字典，每个字典的key就是线程id，因为要做数据隔离
{
1232: {"ctx": ctx},
1325: {"ctx": ctx},
......
......
}
视图函数：
通过from flask import request, session，会根据当前的线程id去取ctx对象，再根据ctx对象去取request和session，至于这一步怎么实现的，之前已经说过了，后续会说。现在只需要知道，请求到来时，flask会做上面说的很多的事情，比如：封装request和session，根据线程创建空间，存储ctx对象，然后我们from flask import request, session，还会根据线程id去取相应的ctx对象，从而得到里面的session和request
请求结束：
将session对象写入名称为session的cookie中，然后根据线程id，从内存中的某个存储线程id和ctx对象的地方(或者直接理解为大字典)移除相应的数据
现在先有一个大概的印象，后面我们就会看源码了。
14.上下文管理前戏：偏函数
不知道大家有没有遇到过这样一种情况，就是在指定回调函数的时候，需要传入函数名，但是这个函数却需要参数。这个时候偏函数就派上用场了，偏函数就是可以让我们先将函数和一部分参数组合起来，变成一个新的函数。然后将剩余的参数传递给新的函数就可以了。
from functools import partial
def add(a1, a2, a3):
print(a1, a2, a3)
# 得到的p就是偏函数
p = partial(add, 1)
# 此时对于p来说，只需要传入a2和a3即可，因为1会自动帮我们传递给a1
p(2, 3) # 1 2 3
"""
注意：对于partial来说，接收的参数是*args和**kwargs
因此构造偏函数所传入的参数和调用偏函数所传入的参数，会把位置参数和位置参数组合，关键字参数和关键字参数组合。
然后按照*args(位置参数)和**kwargs(关键字参数)传进去
"""
"""
p = partial(add, 1)
p(2, 3)
==> add(1, 2, 3)
p = partial(add, a1=1)
p(2, 3)
==> add(2, 3, a1=1) 报错，a1被重复赋值了
p = partial(add, a1=1)
p(a=2, a3=3)
==> add(a1=1, a2=2, a3=3)
p = partial(add, a2=1)
p(2, 3)
==> add(2, 3, a2=1) 报错，a2被重新赋值了
p = partial(add, a2=1)
p(2, a3=3)
==> add(2, a2=1, a3=3)
"""
手动实现偏函数
def partial(func, *args1, **kwargs1):
def inner(*args2, **kwargs2):
args = args1 + args2
kwargs = dict(list(kwargs1.items()) + list(kwargs2.items()))
return func(*args, **kwargs)
return inner
def add(a1, a2, a3):
print(a1, a2, a3)
p = partial(add, 2)
p(1, 3) # 2 1 3
p = partial(add, a1=2)
p(a2=1, a3=3) # 2 1 3
p = partial(add, a2=2)
p(1, a3=3) # 1 2 3
15.上下文管理前戏：基于列表实现栈
栈是一个后入先出的数据结构，为什么会有这种数据结构。
比如我们调用函数，A调用B，B又调用C，那么函数执行的先后顺序就是C,B,A。
再比如，我们经常用的编辑器的undo操作，就是ctrl+z后退键，这也是一个栈，后退的一定是最后操作的。
class Stack:
def __init__(self):
# 这个是不建议外部访问的，所以设置成私有的
self.__data = []
def push(self, val):
self.__data.append(val)
def pop(self):
return self.__data.pop()
def top(self):
# 查看栈顶元素
return self.__data[-1]
stack = Stack()
stack.push(1)
stack.push(2)
stack.push(3)
print(stack.top()) # 3
print(stack.top()) # 3
print(stack.top()) # 3
print(stack.pop()) # 3
print(stack.pop()) # 2
print(stack.pop()) # 1
16.上下文管理前戏：Local对象
还记得之前实现的Local对象吗？那个太low了，这次我们来实现一个更高级的，使用魔法方法
import threading
class Local:
def __init__(self):
object.__setattr__(self, "storage", {})
def __setattr__(self, key, value):
ident = threading.get_ident()
if ident not in self.storage:
self.storage[ident] = {key: value}
else:
self.storage[ident][key] = value
def __getattr__(self, item):
ident = threading.get_ident()
if ident in self.storage:
return self.storage[ident].get(item)
"""
此时得到的storage就是
{
"线程id1": {key: value},
"线程id2": {key: value}
}
但是我们通过Local对象获取value的时候不需要显示的指定id，只需要指定key即可
因为线程id会自动获取
"""
flask中也是使用了类似的实现方式
17.上下文管理前戏：LocalStack对象
在Local对象的时候，我们看到，可以通过类似于
local.xxx = 123
print(local.xxx) # 123
这样的形式设置和获取值。
如果我们local.stack = []的话
那么还可以进行local.stack.append("xx")等操作
import threading
class Local:
"""由flask中Local类化简得来"""
def __init__(self):
object.__setattr__(self, "storage", {})
def __setattr__(self, key, value):
ident = threading.get_ident()
if ident not in self.storage:
self.storage[ident] = {key: value}
else:
self.storage[ident][key] = value
def __getattr__(self, item):
ident = threading.get_ident()
if ident in self.storage:
return self.storage[ident].get(item)
local = Local()
local.stack = []
print(local.stack) # []
local.stack.append("古明地觉")
local.stack.append("椎名真白")
print(local.stack) # ['古明地觉', '椎名真白']
print(local.stack.pop()) # 椎名真白
print(local.stack) # ['古明地觉']
"""
{
"线程id": {"stack": []}
}
"""
可以看到我们通过append和pop的方式，相当于将value维护成了栈的模式。但是还有一个问题，那就是比较麻烦。因为我们要通过local.stack.append的方式，于是我们能不能通过找一个代理的方式，来帮我们往里面添加元素和pop元素呢？
可以找一个代理，LocalStack，这是还是看flask中的LocalStack
from threading import get_ident
class Local(object):
"""flask源码里面的Local，去掉了一部分类容
可以看到大致思路是类似的
"""
__slots__ = ("__storage__", "__ident_func__")
def __init__(self):
object.__setattr__(self, "__storage__", {})
object.__setattr__(self, "__ident_func__", get_ident)
def __release_local__(self):
self.__storage__.pop(self.__ident_func__(), None)
def __getattr__(self, name):
try:
return self.__storage__[self.__ident_func__()][name]
except KeyError:
raise AttributeError(name)
def __setattr__(self, name, value):
ident = self.__ident_func__()
storage = self.__storage__
try:
storage[ident][name] = value
except KeyError:
storage[ident] = {name: value}
def __delattr__(self, name):
try:
del self.__storage__[self.__ident_func__()][name]
except KeyError:
raise AttributeError(name)
class LocalStack(object):
"""对源码化简。去掉注释"""
def __init__(self):
# 实例化一个Local对象
self._local = Local()
def push(self, obj):
# 从Local对象里面获取stack
# 当一个请求过来的时候，肯定是没有stack的
rv = getattr(self._local, "stack", None)
# 所以如果rv is None的话
if rv is None:
# 给Local对象添加一个stack，当然这个stack肯定是赋值到storage这个大字典里面去了，作为value和线程id作为key
# 组成storage的一个键值对
self._local.stack = rv = []
# 对rv进行append，当然由于列表是可变类型
# 修改也会同步到local.__storage__["线程id"]["stack"]，即：local.stack
rv.append(obj)
return rv
# pop则是从栈顶弹出一个一个元素
def pop(self):
# 获取stack
stack = getattr(self._local, "stack", None)
# 如果是第一次请求，stack是None，那么返回None
if stack is None:
return None
# 如果stack只有一个元素的话
# 那么直接取，就不进行pop了，因为pop之后就没有元素了，但是stack依然存在，是个空列表
elif len(stack) == 1:
# 然后这个方法会将__storage__中，key为当前线程id的键值对给删除
self._local.__release_local__()
return stack[-1]
# 否则的话，弹出来一个元素
else:
return stack.pop()
@property
def top(self):
"""返回栈顶元素"""
try:
return self._local.stack[-1]
except (AttributeError, IndexError):
return None
ls = LocalStack()
ls.push(123)
ls.push(456)
ls.push(456789)
print(ls._local.__storage__[get_ident()]["stack"]) # [123, 456, 456789]
# 或者直接通过.stack的方式，因为我们设置就是自动根据线程id往ls._local.storage__里面设置的
# 取也可以直接取
print(ls._local.stack) # [123, 456, 456789]
print(ls.pop()) # 456789
print(ls.pop()) # 456
# 此时栈里面只剩下123了，说白了就是一个列表，用列表实现栈。
print(ls.top) # 123
因此LocalStack以栈的形式维护Local对象里面数据，以后我们直接通过操作LocalStack对象即可，只需要调用push、pop、top方法即可
Local：为每个线程开辟一个单独的空间
LocalStack：以栈的形式维护Local对象里面数据，可以更加方便去操作。但是如果没有LocalStack，我们能不能操作Local里面的数据呢？答案显然是可以的，只是会稍微麻烦一些。
18.上下文管理前戏：以上综合应用
我们说Local对象里面可以放任何东西，那么我们可不可以放一个类的实例对象到里面去呢。
from threading import get_ident
class Local(object):
__slots__ = ("__storage__", "__ident_func__")
def __init__(self):
object.__setattr__(self, "__storage__", {})
object.__setattr__(self, "__ident_func__", get_ident)
def __release_local__(self):
self.__storage__.pop(self.__ident_func__(), None)
def __getattr__(self, name):
try:
return self.__storage__[self.__ident_func__()][name]
except KeyError:
raise AttributeError(name)
def __setattr__(self, name, value):
ident = self.__ident_func__()
storage = self.__storage__
try:
storage[ident][name] = value
except KeyError:
storage[ident] = {name: value}
def __delattr__(self, name):
try:
del self.__storage__[self.__ident_func__()][name]
except KeyError:
raise AttributeError(name)
class LocalStack(object):
def __init__(self):
self._local = Local()
def push(self, obj):
rv = getattr(self._local, "stack", None)
if rv is None:
self._local.stack = rv = []
rv.append(obj)
return rv
def pop(self):
stack = getattr(self._local, "stack", None)
if stack is None:
return None
elif len(stack) == 1:
self._local.__release_local__()
return stack[-1]
else:
return stack.pop()
@property
def top(self):
try:
return self._local.stack[-1]
except (AttributeError, IndexError):
return None
ls = LocalStack()
class RequestContext:
def __init__(self):
self.request = "request"
self.session = "session"
# 这里也可以传入一个RequestContext对象
# flask也是这么做的
ls.push(RequestContext())
ctx = ls.pop()
print(ctx.request) # request
print(ctx.session) # session
但是呢？我们还可以进一步封装
from threading import get_ident
class Local(object):
__slots__ = ("__storage__", "__ident_func__")
def __init__(self):
object.__setattr__(self, "__storage__", {})
object.__setattr__(self, "__ident_func__", get_ident)
def __release_local__(self):
self.__storage__.pop(self.__ident_func__(), None)
def __getattr__(self, name):
try:
return self.__storage__[self.__ident_func__()][name]
except KeyError:
raise AttributeError(name)
def __setattr__(self, name, value):
ident = self.__ident_func__()
storage = self.__storage__
try:
storage[ident][name] = value
except KeyError:
storage[ident] = {name: value}
def __delattr__(self, name):
try:
del self.__storage__[self.__ident_func__()][name]
except KeyError:
raise AttributeError(name)
class LocalStack(object):
def __init__(self):
self._local = Local()
def push(self, obj):
rv = getattr(self._local, "stack", None)
if rv is None:
self._local.stack = rv = []
rv.append(obj)
return rv
def pop(self):
stack = getattr(self._local, "stack", None)
if stack is None:
return None
elif len(stack) == 1:
self._local.__release_local__()
return stack[-1]
else:
return stack.pop()
@property
def top(self):
try:
return self._local.stack[-1]
except (AttributeError, IndexError):
return None
ls = LocalStack()
class RequestContext:
def __init__(self):
self.request = "request"
self.session = "session"
ls.push(RequestContext())
def get_request():
ctx = ls.top
return ctx.request
def get_session():
ctx = ls.top
return ctx.session
print(get_request()) # request
print(get_session()) # session
# 但是我们能不能定义的再简单一些呢？
def get_session_or_request(name):
ctx = ls.top
return getattr(ctx, name)
print(get_session_or_request("session")) # session
print(get_session_or_request("request")) # request
# 但是还是有点麻烦，我们可以定义的再简单一点
from functools import partial
request = partial(get_session_or_request, "request")
session = partial(get_session_or_request, "session")
print(request()) # request
print(session()) # session
看看flask的源码，flask.globals，去掉注释长这样
from functools import partial
from werkzeug.local import LocalProxy
from werkzeug.local import LocalStack
_request_ctx_err_msg = """\
Working outside of request context.
This typically means that you attempted to use functionality that needed
an active HTTP request. Consult the documentation on testing for
information about how to avoid this problem.\
"""
_app_ctx_err_msg = """\
Working outside of application context.
This typically means that you attempted to use functionality that needed
to interface with the current application object in some way. To solve
this, set up an application context with app.app_context(). See the
documentation for more information.\
"""
# 这个不就是我们定义的get_request_or_session(name)函数吗？
# 根据传入的name来获取ctx里面的属性，
def _lookup_req_object(name):
# _request_ctx_stack就是LocalStack的实例对象，就是我们的ls，名字不同而已
# 进行top返回栈顶元素
top = _request_ctx_stack.top
if top is None:
raise RuntimeError(_request_ctx_err_msg)
return getattr(top, name)
# 这个函数暂时不用管，提前剧透，flask中上下文分为两种
# 一种是上面的请求上下文， 一种是这里的app上下文， app上下文后面介绍
def _lookup_app_object(name):
top = _app_ctx_stack.top
if top is None:
raise RuntimeError(_app_ctx_err_msg)
return getattr(top, name)
# 与app相关的，暂时不用管
def _find_app():
top = _app_ctx_stack.top
if top is None:
raise RuntimeError(_app_ctx_err_msg)
return top.app
# 这里实例化一个LocalStack对象
_request_ctx_stack = LocalStack()
_app_ctx_stack = LocalStack()
current_app = LocalProxy(_find_app)
# 这里的LocalProxy先不用管，看里面的，是不是和我们定义的也是一样的呢？
request = LocalProxy(partial(_lookup_req_object, "request"))
session = LocalProxy(partial(_lookup_req_object, "session"))
g = LocalProxy(partial(_lookup_app_object, "g"))
flask上下文的原理就是通过以上代码实现的。
19.flask请求上下文原理剖析
继续看源码，这次就很简单了。之前我们说过请求到来的时候，会访问app.wsgi_app
app.wsgi_app
def wsgi_app(self, environ, start_response):
# 获取ctx
ctx = self.request_context(environ)
error = None
try:
try:
# 这一步，调用了ctx.push
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except:
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)
ctx.push
def push(self):
if top is not None and top.preserved:
top.pop(top._preserved_exc)
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和self
# 这里的self是不是就是我们的ctx啊，_request_ctx_stack是不是就是我们LocalStack对象啊
# 调用这个对象的push方法，将ctx对象写到Local对象开辟的空间里面去。
# 至于LocalStack里面的push，我想不用介绍了，前戏已经做的够足了
_request_ctx_stack.push(self)
# 如果session是None
if self.session is None:
# 实例化获取session_inteface
session_interface = self.app.session_interface
# 调用内部的open_session方法，获取名为session的cookie
# 写到flask中的session对象里面去
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()
因此我们为什么能from flask import request获取ctx里面的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就是getattr(top, name)
而top是_request_ctx_stack.top，弹出来的就是ctx对象。
flask的Local里面有一个__storage__大字典
__storage__ = {
"线程id": {"stack": [ctx对象(里面有request和session)]}
}
这个线程id是根据当前线程id自动传入的，而stack则是flask默认的名字，已经定死了，当然这些都不用我们去管。
当我们from flask import request的时候，会自动地调用LocalStack对象的top方法，然后根据线程id找到对应的value，此时对应的value依旧是一个字典，然后再找到这个字典里面的key为stack所对应的value，然后再把ctx弹出来，当然这对于我们使用者来说，是不需要关心的。我们只需要知道通过LocalStack可以将Local对象里面的相应线程id对应的ctx取出来即可，然后通过反射，来获取属性，传入"request"获取request，传入"session"获取session
from flask import Flask
from flask.globals import _request_ctx_stack
app = Flask(__name__)
@app.route("/hello")
def hello():
# 我们不通过from flask import request的方式
# 也可以直接通过调用LocakStack对象的top，根据线程id获取ctx，再获取ctx下的request
request = _request_ctx_stack.top.request
return "hello" + request.method
if __name__ == "__main__":
app.run(port=8888, debug=True)
其实前戏做足了，高潮很短暂的。
20.flask请求上下文：LocalProxy对象
先来介绍一下python的魔法方法
class Foo:
def __str__(self):
return "foo"
def __getattr__(self, item):
return "getattr"
def __getitem__(self, item):
return "getitem"
def __add__(self, other):
return other + 3
foo = Foo()
# 打印foo会触发__str__方法
print(foo) # foo
# 通过.的方式获取属性，会触发__getattr__方法
print(foo.abc) # getattr
# 通过[]的方式获取属性，会触发__getitem__方法
print(foo["哈哈哈"]) # getitem
# + 本质上也是一样的,会触发__add__方法
# foo会传给self，1则传给other
print(foo + 1) # 4
"""
可以看到操作foo，本质上就是操作Foo类所对应的魔法方法。
每一个操作符，都会有一个魔法方法对应
"""
我们再来将代码改一下
DATA = {
"request": {
"method": "GET",
"args": {"a": 1, "b": 2}
},
"session": {
"username": "satori",
"password": "123456"
}
}
class LocalProxy:
def __init__(self, key):
self.key = key
def get_data(self):
return DATA[self.key]
def __getattr__(self, item):
data = self.get_data()
return data[item]
request = LocalProxy("request")
print(request.method) # GET
print(request.args) # {'a': 1, 'b': 2}
session = LocalProxy("session")
print(session.username) # satori
print(session.password) # 123456
"""
可以看到，我们其实是可以直接取值的。
但是呢，我们通过找一个代理，让代理帮我们去取值。
这个DATA就可以简单看成是Local中的数据
"""
因此LocalProxy、LocalStack、Local三者的关系。
Local：真正开辟空间，存取数据的
LocalStack：可以以栈的形式更加方便地往Local里面存取数据
LocalProxy：这才是真正意义上的代理，通过代理，帮我们调用LocalStack往Local里面存取数据。
我们加入LocalProxy再来总结一下流程
from flask import Flask
app = Flask(__name__)
@app.route("/hello")
def hello():
return "hello"
if __name__ == "__main__":
app.run(port=8888, debug=True)
"""
第一阶段：请求到来
将request和session相关数据封装到RequestContext对象中
再通过LocalStack将ctx添加到Local中
__storage__ = {
线程id: {"stack": [ctx(request, session)]}
}
第二阶段：视图函数中获取request和session
方法一：直接找LocalStack获取
from flask.globals import _request_ctx_stack
_request_ctx_stack.top.request.method
方法二：通过代理LocalProxy
from flask import request
request.method
方法一也说了， 我们可以直接通过LocalStack获取，但是会比较麻烦。要调用LocalStack对象下的top方法弹出ctx，然后拿到request和session
而LocalProxy则帮助我们更简单的实现了这一过程，可以让我们直接from flask import request, session
如果是request，那么LocalProxy会通过LocalStack获取Local里面的request
如果是session，那么LocalProxy会通过LocalStack获取Local里面的session
"""
request和session本质上就是LocalProxy对象
request = LocalProxy(partial(_lookup_req_object, "request"))
session = LocalProxy(partial(_lookup_req_object, "session"))
本质上都是一样的，只不过参数不一样，一个是request，一个是session
21.flask请求上下文：基本流程
# context locals
_request_ctx_stack = LocalStack()
_app_ctx_stack = LocalStack()
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"))
"""
之前说过，flask上下文分为两种
一种是请求上下文：封装了，request和session
一种是app上下文：封装了，app和g
因此同样会创建一个Local对象，根据封装app上下文。
因此会有两个Local对象，一个存放请求上下文，另一个存放app上下文。
__storage__ = {
"线程id": {"stack": [ctx(request, session)]}
}
__storage__ = {
"线程id": {"stack": [app_ctx(app, g)]}
}
当然两个local的__storage__里面的线程id是一一对应的，只不过存储的数据不一样
"""
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()
# 同样调用push方法
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
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()
请求上下文
class AppContext(object):
"""
源码这里就不分析了，和请求上下文是类似的
初始化的g是一个空的，啥也没有
"""
def __init__(self, app):
self.app = app
self.url_adapter = app.create_url_adapter(None)
self.g = app.app_ctx_globals_class()
self._refcnt = 0
def push(self):
"""Binds the app context to the current context."""
self._refcnt += 1
if hasattr(sys, "exc_clear"):
sys.exc_clear()
_app_ctx_stack.push(self)
appcontext_pushed.send(self.app)
def pop(self, exc=_sentinel):
"""Pops the app context."""
try:
self._refcnt -= 1
if self._refcnt <= 0:
if exc is _sentinel:
exc = sys.exc_info()[1]
self.app.do_teardown_appcontext(exc)
finally:
rv = _app_ctx_stack.pop()
assert rv is self, "Popped wrong app context. (%r instead of %r)" % (rv, self)
appcontext_popped.send(self.app)
def __enter__(self):
self.push()
return self
def __exit__(self, exc_type, exc_value, tb):
self.pop(exc_value)
if BROKEN_PYPY_CTXMGR_EXIT and exc_type is not None:
reraise(exc_type, exc_value, tb)
程序启动：
创建两个Local对象，
local1 = {
}
local2 = {
}
和两个LocalStack对象
_request_ctx_stack
_app_ctx_stack
请求到来：
对数据进行封装：
调用RequestContext，得到ctx(request, session)
调用AppContext，得到app_ctx(app, g)
保存数据：
将包含了(request,session)的ctx对象通过_request_ctx_stack这个LocalStack对象放到local1里面去
将包含了(app,g)的ctx对象通过_app_ctx_stack这个LocalStack对象放到local2里面去
视图函数处理：
通过Localproxy对象(可以是request，可以是g等等)调用LocalStack存取Local里面的数据
请求结束：
调用两个LocalStask的pop方法，直接删除
22.flask上下文管理：g的作用
我们刚才已经知道了，flask当中有一个app上下文，这个app上下文有一个g，那个g是干什么用的
flask中的g对象可以用来绑定变量
那么g对象的生命周期是什么样的呢
我们知道，当请求到来，路由映射的时候，app上下文就已经创建了，也就是说g对象也已经创建好了，那么可以把变量绑定在上面。然后试图函数就可以直接操作了，在一次请求中，所有的视图函数共享g对象。在请求的过程中，有可能还会走一些钩子函数，比如before_request，after_request等等，在这一次请求的生命周期内，g可以直接使用，比如赋值、取值等操作。
但是当请求结束之后，g对象，或者说g对象所在的app上下文就已经被清空了。所以g对象的生命周期和整个请求的生命周期是一致的。
咦，看了g对象的作用，感觉有点像session，那么它和session有什么区别呢
session是跨请求的，因为session会写到cookie里面去，第二次请求过来时，会对cookie进行读取、解密、反序列化，只要未失效，那么获取到的就是同一个合法的session。但是g对象不会写入某个地方然后再读取，请求一次g就改变了一次，或者说呗重新赋了值。
那么g和全局变量一样吗？
显然是不一样的，主要有两点。
一：启动之后，全局变量就在内存里面放着，只要不销毁，那么就会一直存在。而g对象之前说过，只在一次请求周期内有效，一旦请求结束，就会销毁，再次请求时，g对象就会重新生成
二：如果是多线程，会修改同一个全局变量，可能造成资源竞争，而产生不可预期的后果。但是g对象是放在app上下文当中的，存在Local对象(或者说Local对象的__storage__)里面的，是有唯一id作为标识的，所以是不会因为多线程而造成资源竞争的。
那么下面就演示一下，这个g。先挖一个坑，看看你会不会跳进去
from flask import Flask, g
app = Flask(__name__)
@app.route("/hello")
def hello():
g.x = 123
return "hello"
@app.route("/index")
def index():
print(g.x)
return "index"
if __name__ == "__main__":
app.run(port=8888, debug=True)
首先我访问localhost:8888/hello，x肯定绑定到了g对象上面，然后我再访问localhost:8888/index,那么我能不能打印g.x呢？
可以看到当我访问/index的时候，显示没有x这个属性。这是为什么呢？
之前说过，g对象只在一次请求的生命周期内有效，换句话说，g对象的生命周期只在当前请求内有效。
访问/hello的时候，确实将x绑定在了g对象上面，但是当视图函数返回之后，这个请求就结束了，也就意味着g对象被销毁了。
当我访问/index的时候，又会创建一个新的g对象，或者说g对象又被赋值为初始的状态，也就是没有任何变量绑定在上面，所以打印不出来g.x
那如果我把代码改一下呢？
from flask import Flask, g, redirect
app = Flask(__name__)
@app.route("/hello")
def hello():
g.x = 123
return redirect("/index")
@app.route("/index")
def index():
print(g.x)
return "index"
if __name__ == "__main__":
app.run(port=8888, debug=True)
这里我们访问/hello，然后自动跳转到/index
会发现依旧无法访问到x，尽管人没有手动进行二次请求，但是对于程序来说，跳转了等于重新发起了一次请求。
from flask import Flask, g
app = Flask(__name__)
@app.before_request
def hello():
g.x = 123
@app.route("/index")
def index():
print(g.x)
return "index"
if __name__ == "__main__":
app.run(port=8888, debug=True)
通过钩子函数是可以获取的
23.信号机制以及使用场景
from blinker import Namespace
# 1.定义信号
namespace = Namespace()
fire = namespace.signal("fire")
# 2.监听信号
# 首先定义一个回调函数，这个函数里面至少要有一个参数。
def open_fire(sender):
print(sender)
print("open fire`````")
# 监听一个信号，发生则调用回调函数
fire.connect(open_fire)
# 3.发送一个信号
# send里面的参数则会传给open_fire
fire.send("xxx")
"""
步骤就三步：
1.定义信号
namespace = NameSpace()
signal = namespace.signal("signal")
2.监听信号
创建需要至少要接收一个参数的函数，然后通过signal.connect进行注册
3.发送信号
调用signal.send("message")方法，然后会触发绑定的函数，同时message也会传进去，如果不传则为None
如果回调函数里面只有一个参数，直接send即可
但是如果回调函数要接收多个参数，那么第一个sender必须通过位置参数指定，其余要通过关键字参数指定
"""
from blinker import Namespace
# 1.定义信号
namespace = Namespace()
fire = namespace.signal("fire")
def open_fire(sender, a, b, c):
print(sender, a, b, c)
print("open fire`````")
fire.connect(open_fire)
# 回调至少接受一个参数，且必须通过位置参数指定
# 如果不传则相当于传了一个None
# 如果有多个参数，那么剩余的参数要通过关键字参数的方式来传递
try:
fire.send("xxx", 1, 2, 3)
except TypeError as e:
print(e) # send() accepts only one positional argument, 4 given
fire.send("xxx", a=1, b=2, c=3)
"""
xxx 1 2 3
open fire`````
"""
# 第一个参数不指定，则默认传了一个None进去
fire.send(a=1, b=2, c=3)
"""
None 1 2 3
open fire`````
"""
信号机制已经了解了，那么如何与flask进行集成呢？
from flask import Flask, request
from blinker import Namespace
app = Flask(__name__)
# 定义一个登录的信号，以后用户登录进来以后，就发送一个登录信号，然后能够监听这个这个信号
# 在监听到这个信号以后，就记录当前这个用户登录的信息
# 用信号的方式，记录用户的登录信息
# 1.定义信号
namespace = Namespace()
login_signal = namespace.signal("login")
# 2.监听信号
def login_log(sender, user_name):
print(f"用户{user_name},ip:{request.remote_addr}登录了")
login_signal.connect(login_log)
@app.route("/login")
def login():
username = request.args.get("username")
if username:
# 3.发送信号
# 这里面也是可以有参数的，user_name会自动传到login_log里的user_name里
login_signal.send(user_name=username)
return "登陆成功"
else:
return "请输入用户名"
if __name__ == "__main__":
app.run(port=8888, debug=True)
24.flask中提供的信号
template_rendered =_signals.signal('template-rendered') 模板渲染完成后的信号before_render_template =_signals.signal('before-render-template') 模板渲染前的信号request_started =_signals.signal('request-started') 模板开始渲染request_finished =_signals.signal('request-finished') 模板渲染完成request_tearing_down =_signals.signal('request-tearing-down') request对象被销毁got_request_exception =_signals.signal('got-request-exception') 视图函数发生异常。一般可以监听这个信号，来记录网站异常信息appcontext_tearing_down =_signals.signal('appcontext-tearing-down') app上下文被销毁的信号appcontext_pushed =_signals.signal('appcontext-pushed') app上下文被推入到栈上的信号appcontext_popped =_signals.signal('appcontext-popped') app上下文被推出到栈上的信号message_flashed =_signals.signal('message-flashed') 调用了flask的finished方法的信号
代码都在flask.signals中，并且flask中的信号也是用了blinker这个模块
flask.signals.py
# -*- coding: utf-8 -*-
"""
flask.signals
~~~~~~~~~~~~~
Implements signals based on blinker if available, otherwise
falls silently back to a noop.
:copyright: 2010 Pallets
:license: BSD-3-Clause
"""
try:
from blinker import Namespace
signals_available = True
except ImportError:
signals_available = False
class Namespace(object):
def signal(self, name, doc=None):
return _FakeSignal(name, doc)
class _FakeSignal(object):
"""If blinker is unavailable, create a fake class with the same
interface that allows sending of signals but will fail with an
error on anything else. Instead of doing anything on send, it
will just ignore the arguments and do nothing instead.
"""
def __init__(self, name, doc=None):
self.name = name
self.__doc__ = doc
def send(self, *args, **kwargs):
pass
def _fail(self, *args, **kwargs):
raise RuntimeError(
"Signalling support is unavailable because the blinker"
" library is not installed."
)
connect = connect_via = connected_to = temporarily_connected_to = _fail
disconnect = _fail
has_receivers_for = receivers_for = _fail
del _fail
# The namespace for code signals. If you are not Flask code, do
# not put signals in here. Create your own namespace instead.
_signals = Namespace()
# Core signals. For usage examples grep the source code or consult
# the API documentation in docs/api.rst as well as docs/signals.rst
template_rendered = _signals.signal("template-rendered")
before_render_template = _signals.signal("before-render-template")
request_started = _signals.signal("request-started")
request_finished = _signals.signal("request-finished")
request_tearing_down = _signals.signal("request-tearing-down")
got_request_exception = _signals.signal("got-request-exception")
appcontext_tearing_down = _signals.signal("appcontext-tearing-down")
appcontext_pushed = _signals.signal("appcontext-pushed")
appcontext_popped = _signals.signal("appcontext-popped")
message_flashed = _signals.signal("message-flashed")
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>欢迎来到避难小屋</h1>
</body>
</html>
from flask import Flask, request, render_template, template_rendered
app = Flask(__name__)
def template_render_func(sender, template, context):
print(template) # <Template '1.html'>
print(context) # {'g': <flask.g of 'app'>, 'request': <Request 'http://localhost:5000/' [GET]>, 'session': <NullSession {}>}
template_rendered.connect(template_render_func)
'''
当我们渲染模板的时候，会自动传过来template和context
'''
@app.route("/")
def index():
return render_template("1.html")
if __name__ == '__main__':
app.run()