使用 local() 函数创建的变量，可以被各个线程调用，但和公共资源不同，各个线程在使用 local() 函数创建的变量时，都会在该线程自己的内存空间中拷贝一份。这意味着，local() 函数创建的变量看似全局变量（可以被各个线程调用），但各线程调用的都是该变量的副本（各调用各的，之间并无关系）。
import time from threading import Thread,local,get_ident # get_ident获取线程id #####这是使用local对象 # a = local() # # # # def task(arg): # # a.value = arg # # # # time.sleep(1) # # # # print(a.value) #####用函数的形式来模拟local # storage = {} # def set(k,v): # # 获取线程id # 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) # time.sleep(1) # v = get("val") # print(v) #面向对象版1 # class Local: # 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) # time.sleep(1) # v = obj.get("val") # print(v) # #面向对象的setattr,getattr版 # class Local: # 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 # # time.sleep(1) # # print(obj.val) # 面向对象，将storage变成对象的属性 class Local(object): def __init__(self): #防止getattr递归 object.__setattr__(self,"storage",{}) # 用父类的方法初始化时创建设置storage为{}，注意此storage为父类的属性 #self.storage = {} def __setattr__(self,k,v): ident = get_ident() if ident in self.storage: self.storage[ident][k] = v # 此处storage是父类的属性，所以不会触发setattr方法 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 time.sleep(1) print(obj.val) for i in range(10): t = Thread(target=task,args=(i,)) t.start()
#偏函数partial from functools import partial def fun(a,b,c,d): return a+b+c+d #第一个参数是函数的对象，第二个到后面都是函数的参数，严格按照python函数的传参规则 tes = partial(fun,1) print(tes) print(tes(2,3,4))
流程图
流程图上半部分
流程图下半部分
_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"))
app.wsig_app源码：
def wsgi_app(self, environ, start_response): # RequestContext的对象，其中包含当前请求的东西。 ctx = self.request_context(environ) error = None try: try: #RequestContext的对象，ctx.push，就执行是RequestContext对象的push方法 #ctx放入local中 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)
# 1 我们self.request_context(environ)执行是：所以ctx是RequestContext(self, environ)的返回值： def request_context(self, environ)： #等一下返回的ctx就是RequestContext（）执行的结果。 return RequestContext(self, environ) #2 RequestContext(self, environ)的执行结果是什么？是RequestContext的对象，这个对象中包含了reuqest等信息 class RequestContext(object): #app, environ，ctx，就是RequestContext的对象， def __init__(self, app, environ, request=None, session=None): self.app = app if request is None: request = app.request_class(environ) self.request = request
#1 ctx.push方法就执行RequestContext的对象的push方法: #源码： def push(self): #当前的self是RequestContext的对象,也就是ctx, _request_ctx_stack.push(self) #2 _request_ctx_stack,我们发现它是全局LocalStack()的对象。 #_request_ctx_stack.push(self)就是执行LocalStack中的push方法，并且把ctx传给了#_request_ctx_stack.push的push方法。self=ctx #LocalStack中的push源码： def push(self, obj): rv = getattr(self._local, "stack", None) if rv is None: #storage[线程，协程id]["stack"] = [] self._local.stack = rv = [] rv.append(obj) return rv #上述代码中的self._local是Local对象，用来保证数据安全。 #当代码执行self._local.stack = rv = []，再Local对象中的存储结构是：storage[线程，协程id]["stack"] = [] # rv.append(obj)，把storage[线程，协程id]["stack"] = [obj],又由于obj=ctx，就算就是把ctx放入到Local对象中。结构如下Local().storage[线程，协程id]["stack"] = [ctx]
#1 #这句话我们执行是请扩展与响应函数 #再响应函数和请求扩展中就可以使用request,它是如果和实现的呢？ from flask import Flask,request,session app = Flask(__name__) def index(): print(request.method) print(session) return "ok" #比如上面代码中request.method，是如何拿到请求相关的内容呢？ #2 #我们点进去看request是什么？ request = LocalProxy(partial(_lookup_req_object, "request")) #我们发现request就是 LocalProxy的对象。 #那我们再LocalProxy应该能找到method属性，但是发现没有，既然没有，它就一定写了__getattr__ #我们看__getattr__的源码： def __getattr__(self, name): #request.method,现在的name就是method if name == "__members__": return dir(self._get_current_object()) #现在的name是method，self._get_current_object()执行结果就偏函数的执行结果 #偏函数的执行结果中来获取method #偏函数的执行结果就是request. #etattr(request, method) return getattr(self._get_current_object(), name) #3 #我们再看看_get_current_object()它的执行结果是什么？ #self._get_current_object()的源码： def _get_current_object(self): return self.__local() #上面的self._get_current_object()执行结果就是self.__local()的执行结果 # 4 #self.__local是什么？ #我们发现这个self.__local是实例化的时候传递的，也就local,代码如下 def __init__(self, local, name=None): #self.__local= _LocalProxy__local=local = partial(_lookup_req_object, "request") #self.__local设置为隐藏属性， object.__setattr__(self, "_LocalProxy__local", local) # self.__local=local #local是实例化的时候传递的。我们实例化这个 LocalProxy(partial(_lookup_req_object, "request")) # 我们self._get_current_object=self.__local=local=partial(_lookup_req_object, "request") #5 #那最终我们明白了： def __getattr__(self, name): #request.method,现在的name就是method if name == "__members__": return dir(self._get_current_object()) return getattr(self._get_current_object(), name) #self._get_current_object() 执行结果，就partial(_lookup_req_object, "request")的执行结果。 #就是ctx.request，所以getattr(self._get_current_object(), name)相当于getattr(ctx.request, method) #6 #partial(_lookup_req_object, "request")（）执行的结果 #_lookup_req_object的代码如下： def _lookup_req_object(name): #这里的name,是"request" top = _request_ctx_stack.top #top是谁？是ctx. if top is None: raise RuntimeError(_request_ctx_err_msg) #top里面找request，也就是top= ctx--》ctx里面找request. return getattr(top, name) # 7 #又因为_request_ctx_stack是LocalStack()的对象，所以 _request_ctx_stack.top就是执行LocalStack()的对象中的top: #_request_ctx_stack.top的源码： @property def top(self): try: # storage[线程，协程id]["stack"] = [ctx] #现在top的retrun值就ctx return self._local.stack[-1] except (AttributeError, IndexError): return None #代码中的self._local是Local对象，是我们存ctx的时候的Local对象。
#g ,global,当前请求的全局, #g，对象只当前请求内有效，一旦不是一个请求，就不能获取 from flask import Flask,g,request app = Flask(__name__) # @app.after_request # def af(res): # # print(g.name) # print("befor1") # return res @app.before_request def befor(): #你知道request里面有哪些？ #g 它就怕你和全局request改变了它里面的属性 g.name = "tank" @app.route("/") def index(): print(g.name) return "ok" @app.route("/index") def index1(): #print(g.name) return "ok1" if __name__ == ‘__main__‘: app.run()
# 要用flask-session,就必须装pip install flask-session #最开始的时候，我讲session,sesssion大字典是cookie中的val #如果我们想要存redis怎么办？ from flask import Flask,session from flask_session import RedisSessionInterface import redis app = Flask(__name__) conn=redis.Redis(host=‘127.0.0.1‘,port=6379,password=‘admin123‘) # 可以不配，host和port都是默认 app.secret_key = "123" # 若user_signer=True,此处需要设定密钥；未False则不用设定 #permanent=True,如果这个值为True,表示关闭浏览器，不失效，如果为Fasle,表示关闭浏览器，session就失效 app.session_interface=RedisSessionInterface(conn,key_prefix=‘lqz‘,use_signer=True) @app.route(‘/‘) def hello_world(): session[‘name‘]=‘lqz‘ return ‘Hello World!‘ @app.route("/index") def index(): print(session[‘name‘]) return "ok" if __name__ == ‘__main__‘: app.run()
from redis import Redis from flask import Flask,session from flask_session import Session app = Flask(__name__) app.config[‘SESSION_TYPE‘] = ‘redis‘ app.config[‘SESSION_REDIS‘] = Redis(host=‘127.0.0.1‘,port=‘6379‘,password=‘admin123‘) Session(app) # 内部参数 ‘sqlalchemy‘为mysql，oracle关系型数据库都行 @app.route("/") def index(): session[‘name‘] = "tank" return "ok" @app.route("/index") def index1(): print(session[‘name‘]) return "ok" if __name__ == ‘__main__‘: app.run()
#pip install blinker from flask import Flask,signals app = Flask(__name__) # 触发信号要执行函数 def si(*args,**kwargs): print(‘触发这个信号‘,args,kwargs) # 等同于时间钩子 signals.request_started.connect(si) # 信号与函数相绑定 request_started为请求到来之前执行，其他参数参考小猿取经flask09-01 @app.route(‘/‘) def index(): return ‘123‘ if __name__ == ‘__main__‘: # app.__call__ app.run()
from flask import Flask from flask.signals import _signals app = Flask(__name__) # 自定义信号 xxxxx = _signals.signal(‘xxxxx‘) # xxxxx为信号的名字 # 接收参数的时候，就算send方法不传，也要接收一个参数，后面按照关键字接收参数 def func(arg,on1): print("123123",arg,on1) # 自定义信号中注册函数 xxxxx.connect(func) # 信号名绑定函数 @app.route("/x") def index(): # 触发信号 #传参的时候，一个必须是位置参数，后面的产生必须是关键字参数 xxxxx.send("sb",on1="sb2") return ‘Index‘ if __name__ == ‘__main__‘: app.run()
# pip install flask-script from flask_script import Manager from flask import Flask app = Flask(__name__) manage = Manager(app) @app.route(‘/‘) def index(): return ‘ok‘ if __name__ == ‘__main__‘: manage.run() # 命令行启动命令如下 python fl-Manage.py runserver
from flask_script import Manager from flask import Flask app = Flask(__name__) manage = Manager(app) @app.route(‘/‘) def index(): return ‘ok‘ @manage.command def cd(arg): # 命令的名字和函数名一致，有参数必须传参数 print(arg) if __name__ == ‘__main__‘: manage.run() # 命令行 python fl-Manage.py cd 123
from flask_script import Manager from flask import Flask app = Flask(__name__) manage = Manager(app) @app.route(‘/‘) def index(): return ‘ok‘ @manage.command def cd(arg): # 命令的名字和函数名一致，有参数必须传参数 print(arg) @manage.option(‘-n‘,‘--name‘,dest=‘name‘) # n为命令简写,name为全写(前后可以互换),dest参数值必须与函数变量名一致 @manage.option(‘-u‘,‘--url‘,dest=‘url‘) def cmd(name,url): print(name,url) if __name__ == ‘__main__‘: manage.run() # 命令行 # python fl-Manage.py cmd -n 123 #只有@manage.option(‘-n‘,‘--name‘,dest=‘name‘) 装饰器时 python fl-Manage.py cmd -u asd --name asfsf # -u 和--name顺序无所谓,只写一个参数也不会报错
flask框架3 local对象补充，偏函数，请求上下文，g对象，flask-session组件，flask的信号，自定义信号(了解)，用命令启动flask项目
原文：https://www.cnblogs.com/ludingchao/p/12554824.html