flask 上下文,flask 发送请求

  flask 上下文,flask 发送请求

  先看一个例子:

  #!/usr/bin/env python

  # -*-编码:utf-8 -*-

  导入线程

  # local_values=threading.local()

  类Foo(对象):

  def __init__(self):

  self.name=0

  local_values=Foo()

  定义函数(数字):

  本地值。名称=数量

  导入时间

  时间。睡眠(1)

  print(local_values.name,threading.current_thread().姓名)

  对于范围(5)中的我:

  th=线程Thread(target=func,args=(i,),name=线程% s“% I”)

  th.start()4线程0

  四线程一

  四线程3

  四线程2

  四线程四

  上述结果不是我们想要的,local_values.name的值被最后一个覆盖了....

  瓶的请求和会议设置方式比较新颖,如果没有这种方式,那么就只能通过参数的传递。

  瓶是如何做的呢?

  1.本地线程,保证即使是多个线程,自己的值也是互相隔离。

  #!/usr/bin/env python

  # -*-编码:utf-8 -*-

  导入线程

  local_values=threading.local()

  定义函数(数字):

  本地值。名称=数量

  导入时间

  时间。睡眠(1)

  print(local_values.name,threading.current_thread().姓名)

  对于范围内的我(20岁):

  th=线程Thread(target=func,args=(i,),name=线程% s“% I”)

  th.start()

  从线程导入获取标识

  导入线程

  定义任务(数量):

  print(get_ident())

  对于范围(10)内的我:

  th=线程Thread(target=task,args=(i,),name=’线程% s“% I”)

  th.start()查看代码

  如上图所示,每个线程都有一个唯一标识

  __setattr__,__getattr__的用法

  类Foo(对象):

  def __init__(self):

  及格

  def __setattr__(self,key,value):

  打印(关键字,值)

  def __getattr__(self,item):

  打印(项目)

  obj=Foo()

  xxx=123

  对象XXX _ _ setattr _ _ _ _ _ getattr _ _

  自定义当地的

  一导入线程

  2尝试:

  3从一种绿色小鸟导入获取当前值作为get_ident #协程

  四除了导入错误:

  5尝试:

  6从线程导入获取标识

  七除了导入错误:

  8来自线程导入get_ident #线程

  9

  10

  11类局部(对象):

  12

  13 def __init__(self):

  14 #递归执行__setattr__

  15 #自己。__存储__={}

  16 #自己__ident_func__=get_ident

  17反对. self _ _ setattr _ _(self, __storage__ ,{})

  18对象__setattr__(self, __ident_func__ ,get_ident)

  19

  20

  21 def __getattr__(self,name):

  22尝试:

  23回归自我。__存储__[self ._ _ ident _ func _ _()][姓名]

  24除了关键错误:

  25引发属性错误(名称)

  26

  27 def __setattr__(self,name,value):

  28标识=自我__ident_func__()

  29储存=自我。__存储_ _

  30尝试:

  31存储[标识][名称]=值

  32除了关键错误:

  33存储[标识]={名称:值}

  34

  35 def __delattr__(self,name):

  36试试:

  德尔赛尔夫街37号。__存储__[self ._ _ ident _ func _ _()][姓名]

  38除了关键错误:

  39引发属性错误(名称)

  40

  41

  42本地值=本地()

  43

  44

  45定义任务(数量):

  46本地值。名称=数量

  47导入时间

  48时间。睡眠(1)

  49 print(local_values.name,threading.current_thread().姓名)

  50

  51

  对于范围(5)中的我为52:

  第53届=穿线Thread(target=task,args=(i,),name=’线程% s“% I”)

  54 th.start()

  55

  56

  57

  58 执行结果:

  59 0 线程0

  60 1 线程一

  61 2 线程2

  62 4 线程四

  63 3 线程3查看代码

  2.上下文原理

  #!/usr/bin/env python

  # -*-编码:utf-8 -*-

  从函数工具导入部分

  从flask.globals导入本地堆栈,本地代理

  ls=LocalStack()

  类请求上下文(对象):

  def __init__(self,environ):

  自我要求=环境

  定义_查找_请求_对象(名称):

  top=ls.top

  如果顶端为无:

  引发运行时错误(最小二乘)

  return getattr(top,name)

  会话=本地代理(部分(_ lookup _ req _ object, request ))

  ls.push(RequestContext(c1)) #当请求进来时,放入

  打印(会话)#视图函数使用

  打印(会话)#视图函数使用

  ls.pop() #请求结束流行音乐

  ls.push(RequestContext(c2 ))

  打印(会话)

  ls.push(RequestContext(c3 ))

  打印(会话)视图代码

  3.瓶内部实现

  #!/usr/bin/env python

  # -*-编码:utf-8 -*-

  从一种绿色小鸟导入获取当前值作为获取标识

  定义发布_本地(本地):

  本地的. release_local__()

  类本地(对象):

  __slots__=(__storage__ , __ident_func__ )

  def __init__(self):

  #自我。__存储__={}

  #自我__ident_func__=get_ident

  对象. self _ _ setattr _ _(self, __storage__ ,{})

  对象__setattr__(self, __ident_func__ ,get_ident)

  def __release_local__(self):

  自我。__存储_ _。波普(自我__ident_func__(),无)

  def __getattr__(self,name):

  尝试:

  回归自我。__存储__[self ._ _ ident _ func _ _()][姓名]

  除了关键错误:

  引发属性错误(名称)

  def __setattr__(self,name,value):

  标识=自我__ident_func__()

  存储=自我。__存储_ _

  尝试:

  存储[标识][名称]=值

  除了关键错误:

  存储[ident]={name: value}

  def __delattr__(self,name):

  尝试:

  德尔赛尔夫。__存储__[self ._ _ ident _ func _ _()][姓名]

  除了关键错误:

  引发属性错误(名称)

  类本地堆栈(对象):

  def __init__(self):

  自我. local=Local()

  def __release_local__(self):

  自我本地的。__release_local__()

  定义推送(self,obj):

  将新项目推入堆栈

  rv=getattr(self ._local, stack ,None)

  如果轮状病毒为无:

  自我. local.stack=rv=[]

  rv .追加(对象)

  返回轮状病毒

  定义弹出(自身):

  从堆栈中移除最顶端的项,将返回

  旧值,如果堆栈已经为空,则为没有.

  stack=getattr(self ._local, stack ,None)

  如果堆栈为无:

  不返回

  elif len(stack)==1:

  发布_本地(自我。_本地)

  返回堆栈[-1]

  否则:

  返回stack.pop()

  @属性

  定义顶部(自身):

   堆栈中最上面的项目。如果堆栈是空的,

  "无"被返回。

  尝试:

  回归自我. local.stack[-1]

  except (AttributeError,IndexError):

  不返回

  stc=LocalStack()

  stc.push(123)

  v=stc.pop()

  打印视图代码

  -源码分析(重点): -

  首先看这段代码:

  if __name__==__main__ :

  # 1.1

  应用程序.__呼叫_ _

  应用程序。请求_类

  响应类=响应

  app.run()

  在上面提到了上下文的原理:

  -本地斯塔克-本地-

  globals.py中有这么段代码,定义了请求

  # \flask\globals.py

  #上下文局部变量

  # 全局变量

  _request_ctx_stack=LocalStack()

  # _app_ctx_stack=LocalStack()

  # current _ app=本地代理(_ find _ app)

  请求=本地代理(部分(_ lookup _ req _ object, request ))

  session=本地代理(部分(_ lookup _ req _ object, session ))

  # g=本地代理(部分(_ lookup _ app _ object, g ))

  #本地类

  类本地(对象):

  __slots__=(__storage__ , __ident_func__ )

  def __init__(self):

  对象. self _ _ setattr _ _(self, __storage__ ,{})

  对象__setattr__(self, __ident_func__ ,get_ident)

  def __getattr__(self,name):

  尝试:

  回归自我。__存储__[self ._ _ ident _ func _ _()][姓名]

  除了关键错误:

  引发属性错误(名称)

  def __setattr__(self,name,value):

  标识=自我__ident_func__()

  存储=自我。__存储_ _

  {

  唯一标识(__ident_func__):{

  堆栈:[ctx/requestcontext,]//栈

  }

  }

  尝试:

  存储[标识][名称]=值

  除了关键错误:

  存储[ident]={name: value}

  #本地堆栈类

  def __init__(self):

  自我. local=Local()

  # LocalStack()就是我们之前写过的一个多线程数据相互隔离的类,每个线程都有一个唯一标识

  # 每个线程都有一个独立的存储数据和访问数据的空间

  # __存储_ _ _ _识别_功能_ _

  # \werkzeug\local.py

  #本地类每个线程唯一标识可以存储数据

  #本地堆栈类实例化当地的类

  #本地堆栈类

  定义推送(self,obj):

  # 刚开始stack=[]

  rv=getattr(self ._local, stack ,None)

  如果轮状病毒为无:

  自我. local.stack=rv=[]

  rv.append(obj)//执行当地的的__setattr__方法

  返回轮状病毒

  定义弹出(自身):

  stack=getattr(self ._local, stack ,None)

  如果堆栈为无:

  不返回

  elif len(stack)==1:

  发布_本地(自我。_本地)

  返回堆栈[-1]

  否则:

  返回stack.pop()

  以上都是启动会加载的文件,创建唯一标识,并给这个标识赋值堆栈=[],堆栈是个栈,存储着每个用户请求的对象(数据)请求上下文对象(ctx)

  {

  唯一标识(__ident_func__):{

  堆栈:[ctx/requestcontext,]//栈

  }

  }

  -接下来开始看看执行流程:-(应用程序)

  应用程序.__呼叫_ _

  应用程序。请求_类

  响应类=响应

  app.run()

  执行应用程序.__呼叫_ _方法

  app.run()=》run_simple(主机,端口,自身,* *选项)=》app .__呼叫_ _

  # \flask\app.py

  # 主要从这里入手分析代码

  #烧瓶类(app)

  app.py

  def __call__(self,environ,start_response):

  return self.wsgi_app(environ,start_response)app.py

  def wsgi_app(self,environ,start_response):

  #将请求相关的数据包围封装到了请求上下文对象中

  #再将对象封装到当地的中(每个线程/每个协程独立空间存储)

  #请求上下文对象中有如下:

  #ctx.app当前应用

  #ctx.request请求对象(封装请求相关的东西)

  #ctx.session

  #请求上下文类ctx.py

  ctx=self.request_context(环境)

  错误=无

  尝试:

  尝试:

  #请求上下文。push()-本地堆栈。推

  #将封装了请求相关的请求上下文对象中的对象添加到LocalStack.push(self)

  #_request_ctx_stack=LocalStack()

  {

  唯一标识(__ident_func__):{

  堆栈:[ctx/requestcontext,]//栈

  }

  }

  # 将中强通过本地堆栈添加到当地的中

  # 此时中强中的会议中已经有值了

  ctx.push()

  回应=自我。完整分派请求()

  例外情况为e:

  错误=e

  response=self.handle_exception

  除了:

  error=sys.exc_info()[1]

  上升

  返回响应(环境,启动响应)

  最后:

  if self.should_ignore_error(错误):

  错误=无

  #请求上下文。pop()-本地堆栈。弹出(_请求_ CTX _堆栈。pop())

  #_request_ctx_stack=LocalStack()

  #最后将自己请求在当地的中保存的数据清除

  ctx.auto_pop错误)

  ctx=self.request_context(环境)记住了!

  定义请求上下文(自身,环境):

  #self=app

  返回请求上下文(自身,环境)

  中强就是请求上下文对象

  #请求上下文对象中有如下属性:

  #ctx.app当前应用

  #ctx.request请求对象(封装请求相关的东西)

  #ctx.session

  ctx.push():

  #执行请求上下文。push()-本地堆栈。推

  #将封装了请求相关的请求上下文对象中的对象添加到LocalStack.push(self)

  {

  唯一标识(__ident_func__):{

  堆栈:[ctx/requestcontext,]//栈

  }

  }

  -请求上下文- ctx.py -

  # \flask\ctx.py

  #请求上下文类

  def __init__(self,app,environ,request=None):

  self.app=app

  如果请求为无:

  请求=app.request_class(环境)

  自我请求=请求

  自我。URL _ adapter=app。create _ URL _ adapter(自身。请求)

  自我。闪光=无

  self.session=无

  自我. implicit_app_ctx_stack=[]

  self.preserved=False

  自我. preserved_exc=None

  自我. after_request_functions=[]

  self.match_request()

  请求=app.request_class(环境)

  #ctx.push

  定义推送(自身):

  top=_request_ctx_stack.top

  如果顶端不是没有人并且顶部。保存:

  top.pop(top ._preserved_exc)

  app_ctx=_app_ctx_stack.top

  如果app_ctx是没有人或者app_ctx.app!=self.app:

  # 应用上下文

  # app_ctx=AppContext对象

  # app_ctx.app=Flask对象

  # app_ctx.g=对象(用于一个周期内要保存的值,相当于全局变量)

  app_ctx=self.app.app_context()

  app_ctx.push()

  自我.隐式应用程序CTX堆栈。附加(应用程序_ CTX)

  否则:

  自我. implicit_app_ctx_stack.append(无)

  if hasattr(sys, exc_clear ):

  sys.exc_clear()

  #selft是请求上下文对象也是应用里面的ctx,包含了请求所有数据

  #_request_ctx_stack=LocalStack()

  #相当于执行LocalStack.push()

  #自我就是请求上下文类

  _request_ctx_stack.push()

  如果自我会话为无:

  会话接口=self . app . session _接口

  自我。会话=会话接口。打开会话(

  自我应用,自我请求

  )

  如果自我会话为无:

  自我。会话=会话接口。make _ null _ session(self。app)

  请求上下文对象中有请求,会话等对象,#ctx.push执行的就是请求上下文.推送其中有这么段代码

  _request_ctx_stack.push()#self是请求上下文对象也是应用里面的ctx,包含了请求所有数据

  #_request_ctx_stack=LocalStack()

  #相当于执行LocalStack.push()

  #自我就是请求上下文类

  _request_ctx_stack.push()

  组中执行的是当地的中存储[标识][名称]=值

  最终目的就是把请求上下文对象也就是中强放到本地堆栈中的唯一标识字典中的堆栈中

  以上明白了吗?不明白可以去看看源码。 现在已经出现了

  ctx=请求上下文应用程序本地堆栈本地四个类

  globals.py刚开始启动加载的文件

  到这里请求上下文已经差不多了,每个请求都有一个独立数据存储空间,互补影响彼此。就是把请求对象放到请求上下文属性里面,然后又把请求上下文对象放到每个线程(请求)或者是协程的一个唯一标识的堆栈中

  存储:{

  唯一标识(__ident_func__):{

  堆栈:[ctx/requestcontext,]//栈

  }

  }

  视图中执行打印(请求)发生了什么?

  -请求-

  # \flask\globals.py

  # 视图中执行打印(请求)发生了什么?

  # 执行本地代理类中的__str__方法

  # 在globals.py文件中(最开始在加载的文件)

  #上下文局部变量

  _request_ctx_stack=LocalStack()

  _app_ctx_stack=LocalStack()

  当前应用程序=本地代理(_查找应用程序)

  请求=本地代理(部分(_ lookup _ req _ object, request ))

  session=本地代理(部分(_ lookup _ req _ object, session ))

  g=本地代理(部分(_ lookup _ app _ object, g ))

  # 偏函数做了啥?

  #partial(_lookup_req_object,请求)

  # 函数_查找_请求_对象去本地堆栈获取唯一标识栈里面的值(请求上下文对象)

  定义_查找_请求_对象(名称):

  # _request_ctx_stack=LocalStack()

  # 返回的其实就是刚开始塞到每个线程唯一标识里面的那个CTX/请求上下文对象

  top=_request_ctx_stack.top

  如果顶端为无:

  引发运行时错误(_请求_ CTX _错误_消息)

  # 到请求上下文对象取请求属性值请求对象

  return getattr(top,name)

  #本地代理类中的__str__方法最终就是执行_查找_请求_对象函数而已

  __setattr__=lambda x,n,v:setattr(x . _ get _ current _ object(),n,v)

  在本地堆栈类中:

  @属性

  定义顶部(自身):

   堆栈中最上面的项目。如果堆栈是空的,

  "无"被返回。

  尝试:

  回归自我. local.stack[-1]

  except (AttributeError,IndexError):

  不返回

  定义_获取_当前_对象(自身):

  如果不是哈撒特尔(self .__local, __release_local__ ):

  # 执行_查找_请求_对象函数

  回归自我。_ _本地()

  尝试:

  返回getattr(self ._ _本地,自我__姓名_ _)

  除了属性错误:

  引发RuntimeError(没有对象绑定到%s % self .__name__)

  请求=本地代理(部分(_ lookup _ req _ object, request ))

  这段代码要仔细看看了打印(请求)=本地代理。__str__=LocalProxy ._ _ str _ _=lambda x:str(x . _ get _ current _ object())=(_ get _ current _ object里面执行了)自我。_ _本地()

  本地代理。_ _本地就是传入的偏函数(

  部分(_lookup_req_object,请求))

  最后执行自我。_ _本地()相当于执行偏函数,返回请求对象了

  本地代理类中

  def __init__(self,local,name=None):

  # Local=partial(_ lookup _ req _ object, request )

  #_local=_LocalProxy__local私有字段访问

  对象__setattr__(self, _LocalProxy__local ,local)

  对象. self _ _ setattr _ _(self, __name__ ,name)

  如果可调用(本地)且不具有属性(本地, __release_local__ ):

  对象__setattr__(自身, __wrapped__ ,本地)

  定义_获取_当前_对象(自身):

  如果不是哈撒特尔(self .__local, __release_local__ ):

  回归自我。_ _本地()

  尝试:

  返回getattr(self ._ _本地,自我__姓名_ _)

  除了属性错误:

  _请求_ ctx _堆栈的应用

  # _请求_ ctx _堆栈的应用

  来自烧瓶。全局导入_请求_ CTX _堆栈

  从函数工具导入部分

  定义_查找_请求_对象(名称):

  # name=请求

  # top=ctx

  top=_request_ctx_stack.top

  如果顶端为无:

  引发RuntimeError(不存在)

  #返回请求

  return getattr(top,name)

  类Foo(对象):

  def __init__(self):

  self.xxx=123

  self.ooo=888

  req=partial(_lookup_req_object, xxx )

  xxx=partial(_lookup_req_object, ooo )

  # 当前求刚进来时

  _request_ctx_stack.push(Foo())

  # 使用

  # obj=_request_ctx_stack.top

  # obj.xxx

  v1=req()

  打印(v1)

  v2=xxx()

  打印(v2)

  # 请求终止,将当地的中的值移除

  请求CTX堆栈。pop()_请求_ CTX _堆栈

  上下文

  -穿线。当地的和瓶自定义当地的对象

  -请求到来

  - ctx=封装请求上下文(请求,会话)

  中强放到当地的中

  -执行视图时

  -导入请求

  -打印(请求)-本地代理对象的__str__

  -请求方法-本地代理对象的__getattr__

  -请求1 -本地代理对象的__添加_ _

  -调用_查找_请求_对象函数:去当地的中将请求上下文想获取到,再去请求上下文中获取请求或会议

  -请求结束

  - ctx.auto_pop()

  中强从当地的中移除。

  多应用应用

  从werkzeug.wsgi导入调度员中间件

  从werkzeug .上菜导入运行_简单

  从烧瓶导入烧瓶,当前应用程序

  app1=Flask(app01 )

  app2=Flask(app02 )

  @app1.route(/index )

  定义索引():

  返回"应用程序01 "

  @app2.route(/index2 )

  定义索引2():

  返回"应用程序2 "

  # http://www.oldboyedu.com/index

  # http://www.oldboyedu.com/sec/index2

  dm=DispatcherMiddleware(app1,{

  /sec: app2,

  })

  if __name__==__main__ :

  附录2。__呼叫_ _

  run_simple(本地主机,5000,dm)查看代码

  离线脚本:

  从瓶导入Flask,当前应用程序,全局,_应用程序ctx _堆栈

  app1=Flask(app01 )

  app1.debug=False #用户/密码/邮箱

  # app_ctx=AppContext(self):

  # app_ctx.app

  # app_ctx.g

  app2=Flask(app02 )

  app2.debug=True #用户/密码/邮箱

  # app_ctx=AppContext(self):

  # app_ctx.app

  # app_ctx.g

  用app 1。app _ context():# _ _ enter _ _方法- push - app_ctx添加到_app_ctx_stack.local

  # {位于0x00000000036E2340的格林莱特,格林莱特对象:{stack: [位于0x00000000037CA438的flask.ctx.AppContext对象]}}

  print(_app_ctx_stack ._本地.__存储_ _)

  打印(当前应用程序。配置[调试])

  使用app2.app_context():

  # {位于0x00000000036E2340的格林莱特,格林莱特对象:{stack: [位于0x00000000037CA438的flask.ctx.AppContext对象]}}

  print(_app_ctx_stack ._本地.__存储_ _)

  打印(当前应用程序。配置[调试])

  打印(当前应用程序。配置[调试])查看代码

  需求:不用数据库连接池,显示数据库连接

  类的理解类(对象):

  定义打开(自身):

  及格

  定义提取(自身,sql):

  及格

  定义关闭(自己):

  及格

  def __enter__(自身):

  self.open()

  回归自我

  def __exit__(self,exc_type,exc_val,exc_tb):

  self.close()

  # obj=SQLHelper()

  # obj.open()

  # obj.fetch(select . )

  # obj.close()

  用SQLHelper()作为目标:#自动调用类中的__输入_ _方法,对象就是__输入_ _返回值

  对象获取( xxxx )

  # 当执行完毕后,自动调用类__出口_ _方法

  # 以后如果遇到:查看代码

郑重声明:本文由网友发布,不代表盛行IT的观点,版权归原作者所有,仅为传播更多信息之目的,如有侵权请联系,我们将第一时间修改或删除,多谢。

留言与评论(共有 条评论)
   
验证码: