正文
flask框架-下
小程序：扫一扫查出行【扫一扫了解最新限行尾号】
复制小程序
Local与偏函数threasing.local
多个线程修改同一个数据，复制多份变量给每个线程用，为每个线程开辟一块空间进行数据存储。
不使用therading.local
# 不用local
from threading import Thread
import time
cxw = -1
def task(arg):
global cxw
cxw = arg
# time.sleep(2)
print(cxw)for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
结果分析：在运行时结果全部显示为9，打印的数据是混乱，当把time.sleep删除后，结果正常打印1，2，3，4....10。
使用threading.local
from threading import Thread
from threading import local
import time
from threading import get_ident
# 特殊的对象
cxw = local()
def task(arg):
# 对象.val = 1/2/3/4/5
cxw.value = arg
time.sleep(2)
print(cxw.value)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
使用local()时数据能正常显示，不混乱。
通过字典自定义threaing.local(函数版)
# 线程下 获取id
from threading import get_ident,Thread
import time
storage = {}
def set(k,v):
index = get_ident()
if index in storage:
storage[index][k] = v
else:
storage[index] = {k:v}
def get(k):
index = get_ident()
return storage[index][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()
面向对象版
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()
通过setattr和getattr实现
# 反射
# hasattr(object,name) # 判断对象是否拥有某个属性
# setattr(object,name,value) #为对象增加新的属性
# getattr(object,name,default) #从对象中获取某个属性
# delattr(object,name) #从对象中删除某个属性
#
from threading import get_ident,Thread
import time
class Local(object):
storage = {}
def __setattr__(self, key, value):
ident = get_ident()
if ident in Local.storage:
Local.storage[ident][key] = value
else:
Local.storage[ident] = {key:value}
def __getattr__(self, key):
ident = get_ident()
return Local.storage[ident][key]
obj = Local()
def task(arg):
obj.val = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
每个对象有自己的存储空间（字典）
from threading import get_ident,Thread
import time
class Local(object):
def __init__(self):
object.__setattr__(self,'storage',{})
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()
兼容线程和协程（源码到request中去看，看local的getattr,setattr）
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()
以上的几种Local 实现方法是为了推导出以下源码实现：获取get_ident，setattr和getattr方法的实现。
源码入口：app.__call__>>>wsgi_>>>push>>>Local>>
内部核心源码：
partial偏函数#偏函数的第二个部分(可变参数)，按原有函数的参数顺序进行补充，参数将作用在原函数上，最后偏函数返回一个新函数
from functools import partial
def test(a,b,c,d):
return a+b+c+dtes=partial(test,1,2) # a=1,b=2 c=3,d=4
print(tes(3,4))
必须遵循传参统一方式，应用场景，传参形式比较随意简便，不必设置默认的变量值。
session源码分析SecureCookieSessionInterface
-open_session
-save_session
请求上下文请求步骤源码分析，跳转实现的生命周期。
from flask import Flask,request
app=Flask(__name__)
@app.route("/")
def index():
print(request.form)
return "ok"
if __name__ == '__main__':
#self,是app,app(),--->Flask对象，Flask的__call__
app.__call__
app.run()'''
1 app.__call__
2 wsgi_app(environ, start_response)
2.1 ctx = self.request_context(environ) 2.1.1 return RequestContext(self, environ)
这里的self是app,environ请求相关
2.1.2 return RequestContext(self, environ)
得到了RequestContext的对象，而且有request属性
2.2 2.1中的ctx就是RequestContext的对象 2.3 ctx.push()执行这个，就是RequestContext的对象的push方法
2.3.1 #执行这个，self-->ctx
_request_ctx_stack.push(self)
2.3.1.1 我们发现_request_ctx_stack = LocalStack()
他的push方法的源码：
def push(self, obj):
rv = getattr(self._local, "stack", None)
if rv is None:
# self._local=>stack-->storage['线程id']['stack']=[ctx,]
self._local.stack = rv = []
rv.append(obj)
return rv3在请求中获取request.form
3.1 request是LocalProxy的对象，当获取属性的时候会走__getattr__
def __getattr__(self, name):
if name == "__members__":
return dir(self._get_current_object())
#name-->form,
#self._get_current_object()===>ctx.request,form
#_get_current_object()---》self.__local() return getattr(self._get_current_object(), name) 3.1.1 self._get_current_object():源码：最终：partial(_lookup_req_object, "request")
def _get_current_object(self): if not hasattr(self.__local, "__release_local__"):
#local==>partial(_lookup_req_object, "request")
#def __init__(self, local, name=None):
# object.__setattr__(self, "_LocalProxy__local", local)
#self.__local()===>local()
return self.__local()
try:
return getattr(self.__local, self.__name__)
except AttributeError:
raise RuntimeError("no object bound to %s" % self.__name__)
4 partial(_lookup_req_object, "request")偏函数的源码
def _lookup_req_object(name):
#name是request
#ctx
top = _request_ctx_stack.top
if top is None:
raise RuntimeError(_request_ctx_err_msg)
#ctx-->request
return getattr(top, name)
4.1中_request_ctx_stack.top
@property
def top(self): try:
return self._local.stack[-1]
except (AttributeError, IndexError):
return None
'''
截图源码跳转流程：
#偏函数的第二个部分(可变参数)，按原有函数的参数顺序进行补充，参数将作用在原函数上，最后偏函数返回一个新函数
from functools import partial
def test(a,b,c,d):
return a+b+c+dtes=partial(test,1,2) # a=1,b=2 c=3,d=4
print(tes(3,4))
必须遵循传参统一方式，应用场景，传参形式比较随意简便，不必设置默认的变量值。
SecureCookieSessionInterface
-open_session
-save_session
请求上下文请求步骤源码分析，跳转实现的生命周期。
from flask import Flask,request
app=Flask(__name__)
@app.route("/")
def index():
print(request.form)
return "ok"
if __name__ == '__main__':
#self,是app,app(),--->Flask对象，Flask的__call__
app.__call__
app.run()'''
1 app.__call__
2 wsgi_app(environ, start_response)
2.1 ctx = self.request_context(environ) 2.1.1 return RequestContext(self, environ)
这里的self是app,environ请求相关
2.1.2 return RequestContext(self, environ)
得到了RequestContext的对象，而且有request属性
2.2 2.1中的ctx就是RequestContext的对象 2.3 ctx.push()执行这个，就是RequestContext的对象的push方法
2.3.1 #执行这个，self-->ctx
_request_ctx_stack.push(self)
2.3.1.1 我们发现_request_ctx_stack = LocalStack()
他的push方法的源码：
def push(self, obj):
rv = getattr(self._local, "stack", None)
if rv is None:
# self._local=>stack-->storage['线程id']['stack']=[ctx,]
self._local.stack = rv = []
rv.append(obj)
return rv3在请求中获取request.form
3.1 request是LocalProxy的对象，当获取属性的时候会走__getattr__
def __getattr__(self, name):
if name == "__members__":
return dir(self._get_current_object())
#name-->form,
#self._get_current_object()===>ctx.request,form
#_get_current_object()---》self.__local() return getattr(self._get_current_object(), name) 3.1.1 self._get_current_object():源码：最终：partial(_lookup_req_object, "request")
def _get_current_object(self): if not hasattr(self.__local, "__release_local__"):
#local==>partial(_lookup_req_object, "request")
#def __init__(self, local, name=None):
# object.__setattr__(self, "_LocalProxy__local", local)
#self.__local()===>local()
return self.__local()
try:
return getattr(self.__local, self.__name__)
except AttributeError:
raise RuntimeError("no object bound to %s" % self.__name__)
4 partial(_lookup_req_object, "request")偏函数的源码
def _lookup_req_object(name):
#name是request
#ctx
top = _request_ctx_stack.top
if top is None:
raise RuntimeError(_request_ctx_err_msg)
#ctx-->request
return getattr(top, name)
4.1中_request_ctx_stack.top
@property
def top(self): try:
return self._local.stack[-1]
except (AttributeError, IndexError):
return None
'''
截图源码跳转流程：
from flask import Flask,request
app=Flask(__name__)
@app.route("/")
def index():
print(request.form)
return "ok"
if __name__ == '__main__':
#self,是app,app(),--->Flask对象，Flask的__call__
app.__call__
app.run()'''
1 app.__call__
2 wsgi_app(environ, start_response)
2.1 ctx = self.request_context(environ) 2.1.1 return RequestContext(self, environ)
这里的self是app,environ请求相关
2.1.2 return RequestContext(self, environ)
得到了RequestContext的对象，而且有request属性
2.2 2.1中的ctx就是RequestContext的对象 2.3 ctx.push()执行这个，就是RequestContext的对象的push方法
2.3.1 #执行这个，self-->ctx
_request_ctx_stack.push(self)
2.3.1.1 我们发现_request_ctx_stack = LocalStack()
他的push方法的源码：
def push(self, obj):
rv = getattr(self._local, "stack", None)
if rv is None:
# self._local=>stack-->storage['线程id']['stack']=[ctx,]
self._local.stack = rv = []
rv.append(obj)
return rv3在请求中获取request.form
3.1 request是LocalProxy的对象，当获取属性的时候会走__getattr__
def __getattr__(self, name):
if name == "__members__":
return dir(self._get_current_object())
#name-->form,
#self._get_current_object()===>ctx.request,form
#_get_current_object()---》self.__local() return getattr(self._get_current_object(), name) 3.1.1 self._get_current_object():源码：最终：partial(_lookup_req_object, "request")
def _get_current_object(self): if not hasattr(self.__local, "__release_local__"):
#local==>partial(_lookup_req_object, "request")
#def __init__(self, local, name=None):
# object.__setattr__(self, "_LocalProxy__local", local)
#self.__local()===>local()
return self.__local()
try:
return getattr(self.__local, self.__name__)
except AttributeError:
raise RuntimeError("no object bound to %s" % self.__name__)
4 partial(_lookup_req_object, "request")偏函数的源码
def _lookup_req_object(name):
#name是request
#ctx
top = _request_ctx_stack.top
if top is None:
raise RuntimeError(_request_ctx_err_msg)
#ctx-->request
return getattr(top, name)
4.1中_request_ctx_stack.top
@property
def top(self): try:
return self._local.stack[-1]
except (AttributeError, IndexError):
return None
'''
1：源码入口__call__>>>wsgi_app
2：查看请求方式内，所传的参数信息
3：reruest-context，内需要传的参数信息
4：倒退回来，切入点ctx.push
5：
6：
7：核心部分
8：request请求的方式
蓝图
对程序进行目录结构划分，类似书籍的目录。
不使用蓝图时，自己分文件，目录结构：
-templates
-views
--__init__.py
-suer.py
-order.py
-app.py
app.py
from views import app
if __name__ == "__main__"
app.run()
__init__.py
from flask import Flask,request
app = Flask(__name__)
# 不导入这个不行from.import account
from.import order
from.import user
user.py
from.import app
@app.route("/user")
def user():
return "user"
order.py
from.import app
@app.route("order")def reder():
return "order"
创建蓝图并注册
推荐使用的两种蓝图模板pro_flask_简单应用程序目录示例
创建蓝图
pro_flask_大型应用目录示例
注册蓝图
总结：
1、xxx = Blueprint("account",name,url_prefix="/xxx"):蓝图url前缀，表示url的前缀，表示该蓝图下的所有url都加前缀。
2、xxx = Blueprint("account",name,url_prefix="/xxx",template_folder="tpls"):给当前的蓝图单独使用，向上查找，当找不到，会找总的templates。
3、蓝图的befort_request，对当前蓝图有效。
4、大型项目，可以模拟出类似于django中的多个APP概念。
g对象
专门用来存储用户信息的对象，g的全称：global，g对象在一次请求中的所有的代码地方，都是可以使用的。
#g对象的特性：
"""
1、当前请求内你设置就可以取，必须先设置，后取，当前请求可以取无限次
2、就算你当前请求，设置了，如果不取，其他请求过来，也取不到。"""
from flask import Flask,g,redirect,request
app = Flask(__name__)
app.debug = True# @app.before_request
# def a():
# if request.path == '/':
# request.name = "hello"
# g.name = "hello"def set_g():
g.name = "hello"@app.route("/")
def index():
set_g()
return redirect("/index")@app.route("/index")
def logon():
print(g.name)
return "ok"
信号
Flask框架中的信号基于blinker,其主要就是让开发者可以在flask请求过程中定制一些用户行为。
安装：
pip3 install blinker
信号的使用
from flask import Flask,signals,render_templateapp = Flask(__name__)# 往信号中注册函数
#1给信号绑定要执行的函数
#无需管调用，因为flask,已经给我们设置调用点def func(*args,**kwargs):
print('触发型号',args,kwargs)
#与该信号进行绑定
signals.request_started.connect(func)
# signals.request_started.send
# 触发信号： signals.request_started.send()
@app.before_first_request
def before_first1(*args,**kwargs):
print("befor_first_request")@app.before_request
def before_first3(*args,**kwargs):
print("befor_request")@app.route('/',methods=['GET',"POST"])
def index():
print('视图')
return "视图"if __name__ == '__main__':
# app.wsgi_app
app.run()
自定义信号绑定+注册
from flask import Flask, current_app, flash, render_template
from flask.signals import _signalsapp = Flask(import_name=__name__)# 自定义信号
xxxxx = _signals.signal('xxxxx')def func(sender,a):
print(sender,a)
print("我是自定义信号")# 自定义信号中注册函数
xxxxx.connect(func)@app.route("/x")
def index():
# 触发信号
xxxxx.send("sb",a="")
return 'Index'if __name__ == '__main__':
app.run()
flask-session
from flask import Flask, current_app, flash, render_template
from flask.signals import _signalsapp = Flask(import_name=__name__)# 自定义信号
xxxxx = _signals.signal('xxxxx')def func(sender,a):
print(sender,a)
print("我是自定义信号")# 自定义信号中注册函数
xxxxx.connect(func)@app.route("/x")
def index():
# 触发信号
xxxxx.send("sb",a="")
return 'Index'if __name__ == '__main__':
app.run()
作用：将默认保存的签名cookie中的值，保存到redis/meecached/file/mysql/Mongodb等数据库。
安装
pip3 install flask-session
存和取：通过保存在cookie中的key，取出vlaue，到数据库拼接picke序列化，反解出来获取session。
使用
from flask import Flask,session
from flask_session import RedisSessionInterface
import redis
app = Flask(__name__)
app.secret_key="shgihh"
conn=redis.Redis(host='127.0.0.1',port=6379)
#use_signer是否对key签名
app.session_interface=RedisSessionInterface(conn,key_prefix="jack",use_signer=True,permanent=False)
@app.route('/')
def iindex():
session["gook"] = "jack"
return "hello world"@app.route("/index")
def login():
print(session["ok"])
return "full"if __name__ == '__main__':
app.run()
使用2：设置配置，方便更换数据库来保存session数据。
from flask import Flask,session
import redis
from flask_session import Session
app = Flask(__name__)
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_REDIS'] =redis.Redis(host='127.0.0.1',port='')
app.config['SESSION_KEY_PREFIX']="jason"
Session(app)@app.route('/')
def hello_world():
session['sb']='jason'
return 'Hello World!'@app.route("/index")
def index():
print(session['sb'])
return "ok"if __name__ == '__main__':
app.run()
