Flask框架（三）—— 请求扩展、中间件、蓝图、session源码分析
目录
请求扩展、中间件、蓝图、session源码分析
一、请求扩展
1、before_request
类比django中间件中的process_request，在请求收到之前绑定一个函数做一些事情
#基于它做用户登录认证
@app.before_request
def process_request(*args,**kwargs):
if request.path == '/login':
return None
user = session.get('user_info')
if user:
return None
return redirect('/login')2、after_request
类比django中间件中的process_response，每一个请求之后绑定一个函数，如果请求没有异常
@app.after_request
def process_response1(response):
print('process_response1 走了')
return response3、before_first_request
第一次请求时会走，跟浏览器无关
@app.before_first_request
def first():
pass4、teardown_request
每次请求走，都会走它，包括视图函数出了异常
@app.teardown_request
def ter(e):
pass5、errorhandler
路径不存在时404，服务器内部错误500
@app.errorhandler(404)
def error_404(arg):
return "404错误了"6、template_global
标签
@app.template_global()
def sb(a1, a2):
return a1 + a2
# {{sb(1,2)}}7、template_filter
过滤器，注意模板层中参数的位置，一个写前面，其余写后面
@app.template_filter()
def db(a1, a2, a3):
return a1 + a2 + a3
# {{ 1|db(2,3)}}二、中间件
from flask import Flask
app = Flask(__name__)
@app.route('/')
def index():
return 'Hello World!'
# 模拟中间件
class Md(object):
def __init__(self,old_wsgi_app):
self.old_wsgi_app = old_wsgi_app
def __call__(self, environ, start_response):
print('开始之前')
ret = self.old_wsgi_app(environ, start_response)
print('结束之后')
return ret
if __name__ == '__main__':
# 把原来的wsgi_app替换为自定义的
app.wsgi_app = Md(app.wsgi_app)
app.run()
三、蓝图（blueprint）
对程序进行目录结构划分
1、不使用蓝图，自己分文件
目录结构：
-templates
-views
-__init__.py
-user.py
-order.py
-app.pyapp.py
from views import app
if __name__ == '__main__':
app.run()init.py
from flask import Flask,request
app = Flask(__name__)
# 不导入这个不行，容易产生交叉导入问题
from . import order
from . import useruser.py
from . import app
@app.route('/user')
def user():
return 'user'order.py
from . import app
@app.route('/order')
def order():
return 'order'2、使用蓝图之中小型系统
蓝图模板下载：
链接：https://pan.baidu.com/s/1nosJD2PCtk2jQa4d3wLL4g
提取码：yhkq -
xxx = Blueprint('account', __name__,url_prefix='/xxx',)url_prifix 表示添加在原路由中加中间路由，此时须访问127.0.0.1/xxx/index才能访问
目录结构：
-flask_pro
-flask_test
-__init__.py
-static
-templates
-views
-order.py
-user.py
-manage.py
__init__.py
from flask import Flask
app=Flask(__name__)
# 3.注册蓝图到app中
from flask_test.views import user
from flask_test.views import order
app.register_blueprint(user.us)
app.register_blueprint(order.ord)manage.py
from flask_test import app
if __name__ == '__main__':
app.run(port=8008)user.py
from flask import Blueprint
# 1.创建蓝图
us=Blueprint('user',__name__，)
# 2.使用蓝图
@us.route('/login')
def login():
return 'login'order.py
from flask import Blueprint
ord=Blueprint('order',__name__)
@ord.route('/test')
def test():
return 'order test'3、使用蓝图之大型系统
蓝图模板下载：
链接：https://pan.baidu.com/s/1Ce2npcn2PTRsvrfHjG9b2A
提取码：rnk94、总结
（1）xxx = Blueprint('account', name,url_prefix='/xxx') ：蓝图URL前缀，表示url的前缀，在该蓝图下所有url都加前缀
（2）xxx = Blueprint('account', name,url_prefix='/xxx',template_folder='tpls')：给当前蓝图单独使用templates，向上查找，当前找不到，会找总templates
（3）蓝图的befort_request，对当前蓝图有效
（4）大型项目，可以模拟出类似于django中app的概念
四、threading.local
多个线程修改同一个数据，复制多份变量给每个线程用,为每个线程开辟一块空间进行数据存储
1、不用threading.local
# 不用local
from threading import Thread
import time
lqz = -1
def task(arg):
global lqz
lqz = arg
# time.sleep(2)
print(lqz)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()2、threading.local使用
from threading import Thread
from threading import local
import time
from threading import get_ident
# 特殊的对象
lqz = local()
def task(arg):
# 对象.val = 1/2/3/4/5
lqz.value = arg
time.sleep(2)
print(lqz.value)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()3、通过字典自定义threading.local（函数）
from threading import get_ident,Thread
import time
storage = {}
def set(k,v):
ident = get_ident()
if ident in storage:
storage[ident][k] = v
else:
storage[ident] = {k:v}
def get(k):
ident = get_ident()
return storage[ident][k]
def task(arg):
set('val',arg)
v = get('val')
print(v)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()4、面向对象版
from threading import get_ident,Thread
import time
class Local(object):
storage = {}
def set(self, k, v):
ident = get_ident()
if ident in Local.storage:
Local.storage[ident][k] = v
else:
Local.storage[ident] = {k: v}
def get(self, k):
ident = get_ident()
return Local.storage[ident][k]
obj = Local()
def task(arg):
obj.set('val',arg)
v = obj.get('val')
print(v)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()5、通过setattr和getattr实现
from threading import get_ident,Thread
import time
class Local(object):
storage = {}
def __setattr__(self, k, v):
ident = get_ident()
if ident in Local.storage:
Local.storage[ident][k] = v
else:
Local.storage[ident] = {k: v}
def __getattr__(self, k):
ident = get_ident()
return Local.storage[ident][k]
obj = Local()
def task(arg):
obj.val = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()6、每个对象有自己的存储空间（字典）
from threading import get_ident,Thread
import time
class Local(object):
def __init__(self):
object.__setattr__(self,'storage',{})
def __setattr__(self, k, v):
ident = get_ident()
if ident in self.storage:
self.storage[ident][k] = v
else:
self.storage[ident] = {k: v}
def __getattr__(self, k):
ident = get_ident()
return self.storage[ident][k]
obj = Local()
def task(arg):
obj.val = arg
obj.xxx = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()7、兼容线程和协程（终极版）——flask中使用版本
try:
from greenlet import getcurrent as get_ident
except Exception as e:
from threading import get_ident
from threading import Thread
import time
class Local(object):
def __init__(self):
object.__setattr__(self,'storage',{})
def __setattr__(self, k, v):
ident = get_ident()
if ident in self.storage:
self.storage[ident][k] = v
else:
self.storage[ident] = {k: v}
def __getattr__(self, k):
ident = get_ident()
return self.storage[ident][k]
obj = Local()
def task(arg):
obj.val = arg
obj.xxx = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()五、session源码分析
1.请求来了,生成一个空session放到ctx中:ctx = self.request_context(environ)
2.ctx.push():
if self.session is None:
session_interface = self.app.session_interface
# 正常情况下session中有值了
# 什么情况下没有:请求中没有cookie,session仍为空
# 从cookie中取出value,(有解密过程)转成session对象
self.session = session_interface.open_session(
self.app, self.request
)
if self.session is None:
# 生成一个空session
self.session = session_interface.make_null_session(self.app)
3.请求走了:
self.full_dispatch_request()
1 执行before_request
2 执行视图函数
3 把session写入
if not self.session_interface.is_null_session(ctx.session):
self.session_interface.save_session(self, ctx.session, response)
六、session修改的坑
# 1.增加值
session['user']={'name':'aaa','age':18}
# 2.修改session的坑
session['user']['name']='bbb'
# 这样无法修改session，这样触发的是session['user']字典的__setitem__方法，没有modified，在save_session时没有保存，所以session并没有改变
# 解决方法一
session['user']={'name':'bbb','age':18}
# 这样是在__setitem__方法中写了modified，并设置为True，在save_session时因为是True所以保存
# 解决方法二
session['user']['name']='bbb'
session.modified=True # 修改完后，手动指定modified为True，这样再save_session时就会保存
博客内容仅供参考，部分参考他人优秀博文，仅供学习使用