一。local
在线程的处理中，常常遇到这样的问题。当一系列线程遇到io操作的时候，就会执行其他线程，这时候就会出现数据冲突，发生数据方面的问题：
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()
这样的线程是错误的。所以为了区别所有线程，在存储数据的时候往往会加上线程号作为key，取数据的时候也会加上线程号。
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: # cxw['线程id']['value']=arg #storage[1][val]=arg #storage={1:{val:agr}} storage[ident] = {k:v} def get(k): ident = get_ident() #1 #storage = {1: {val: agr}} return storage[ident][k] def task(arg): set('val',arg) time.sleep(1) v = get('val') print(v) for i in range(10): t = Thread(target=task,args=(i,)) t.start()
这就是local中的内部原理，但是根据类进行封装就需要使用__setattr__，__getattr__，在init中创建存储字典storage防止递归。
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.set(val,arg) obj.val = arg print(obj.val) for i in range(10): t = Thread(target=task,args=(i,)) t.start()
偏函数
偏函数就是可以定义任意函数，通过偏函数先定义一部分值，再加上另外的值就可以执行的函数，：
from functools import partial def test(a,b,c,d): return a+b+c+d tes=partial(test,2,2) print(tes) print(tes(300,4)) # functools.partial(<function test at 0x00000181D14F1EA0>, 2, 2) # 308
二。请求上下文。
当在全局调用request方法的时候，就可以从其中获取request相关的数据，是在其生成的对象被调用 的时候将request放入。流程如下：
1.app.__call__
2.wsgi_app(environ,start_response),将app，环境和请求等信息传入。
3.ctx = self.request_context(environ)，其中是生成一个request对象的函数。此时ctx代表的就是请求对象。
4.ctx.push（），执行请求对象中的push方法。
5._request_ctx_stack.push(self)
这里的_request_ctx_stack，其实就是LocalStack()，所以在这个函数中寻找push方法
6.push方法
以上的push方法中需要rv，rv是从_local中获取到，而local其实就是代表线程容器的Local()
如果没有获得stake就会在其中创建rv，并将该对象添加
以form为例子，从request中取出该元素，流程如下。
1.request = LocalProxy(partial(_lookup_req_object, "request"))
request是licalproxy对象，当点方法从中取东西时，会走__getattr__
2.getattr,会走_get_current_object
3._get_current_object
最终的反射获取的时__init__中的封装方法"_LocalProxy__local"
而其中的local代表的就是LocalProxy的参数，一个偏函数。
4.偏函数partial(_lookup_req_object, "request")
其中的函数走的时_lookup_req_object，最后获取的元素就是top
5.top
top代表的就是LocalStack()，所以需要运行其中dtop方法：
而_local就是Local(),代表线程容器，从中取出刚刚传入的request对象。
''' 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 rv 3在请求中获取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
三。蓝图。
蓝图就是一个项目的框架，可以参照django写一个，
一个蓝图需要包函以下东西。可以将app下兴建一个pro文件，在这个文件下创建项目相关
1.manage
里面放启动的文件和语句，需要导入app
2.templates
在其同一个文件夹下的文件都可以时从其中文件夹下导入模板。
3.视图
视图下需要导入app进行修饰。
4.__init__
在该文件夹下创建初始化文件，创建app，并导入view，让manage启动时提供view。
可以看到，不使用蓝图创建框架会使得循环导入，可以从view层的app使用蓝图，将其循环链断开。
在视图函数中创建蓝图：
us=Blueprint("user",__name__) @us.route("/") def index(): return render_template("index.html")
这样在使用路由后，就可以调用该蓝图。
当蓝图被创建之后，需要在app下进行注册。
from flask import Flask app=Flask(__name__) from pro import views app.register_blueprint(views.us)
蓝图中所取的名字，代表每个蓝图的头，可以调用不同蓝图中相同名字 的视图函数，只要前缀添加即可。
所以，在蓝图中不能出现名字相同的 蓝图
各个蓝图文件下都有自己的请求扩展周期，执行哪个蓝图就执行哪个蓝图的请求扩展。
小型flask框架：
#!/usr/bin/env python # -*- coding:utf-8 -*- from flask import Blueprint from flask import render_template from flask import request account = Blueprint('acc', __name__) @account.route('/login.html', methods=['GET', "POST"]) def login(): return render_template('login.html')
#!/usr/bin/env python # -*- coding:utf-8 -*- from flask import Blueprint,url_for blog = Blueprint('blog', __name__) @blog.before_request def a(): print("wo我是blog 的请求扩展") @blog.route("/index") def index(): print(url_for("acc.login")) return "ok"
#!/usr/bin/env python # -*- coding:utf-8 -*- from flask import Blueprint user = Blueprint('user', __name__)
#!/usr/bin/env python # -*- coding:utf-8 -*- from flask import Flask app = Flask(__name__,template_folder='templates',static_folder='statics',static_url_path='/static') @app.before_request def a(): print("我是app里面的befor_request") from .views.account import account from .views.blog import blog from .views.user import user app.register_blueprint(account) app.register_blueprint(blog) app.register_blueprint(user)
from pro_flask import app if __name__ == '__main__': app.run()
url_prefix代表的时这个蓝图的前缀，在前端访问该蓝图下的视图时需要加前缀来区分视图：
大型flask框架
#!/usr/bin/env python # -*- coding:utf-8 -*- from flask import Blueprint admin = Blueprint( 'admin', __name__, template_folder='templates', static_folder='static' ) from . import views
#!/usr/bin/env python # -*- coding:utf-8 -*- from . import admin @admin.route('/index') def index(): return 'Admin.Index'
#!/usr/bin/env python # -*- coding:utf-8 -*- from flask import Blueprint web = Blueprint( 'web', __name__, template_folder='templates', static_folder='static' ) from . import views
#!/usr/bin/env python # -*- coding:utf-8 -*- from . import web @web.route('/index') def index(): return 'Web.Index'
#!/usr/bin/env python # -*- coding:utf-8 -*- from flask import Flask from .admin import admin from .web import web app = Flask(__name__) app.debug = True app.register_blueprint(admin, url_prefix='/admin') app.register_blueprint(web)
from pro_flask import app if __name__ == '__main__': app.run()
四。g对象
g对象的特性是，当前请求之内，可以被取出，但是需要先设置，当前请求可以无线取出，和闪现不同的是，这次请求之后，就不再有了，也就是其他请求不能取。
from flask import Flask,request,g def set_g(): g.name='nb' @app.route("/index") def login(): print(g.name) return "2"
应用场景有可以再请求扩展中，添加判断，给指定路由进行传参。
五。信号
信号就是当到达某个阶段就会触发某个函数。
信号需要基于blinker，所以需要安装模块：
pip install blinker
信号系统自带的有：
型号的使用：
# 往信号中注册函数 #1给信号绑定要执行的函数 #无需管调用，因为flask,已经给我们设置调用点 def func(*args,**kwargs): print('触发型号',args,kwargs) #与该信号进行绑定 signals.request_started.connect(func) # signals.request_started.send # 触发信号： signals.request_started.send()
该信号是再请求扩展之后，view视图之前。
自定义信号。
仿照源码，自定义信号需要先创建一个信号对象，再进行绑定，最后使用send进行调用：
自定义的参数默认传入一个参数，这个参数需要再函数中接受，可以不传，不传值为none，如果需要传其他参数，需要使用关键字参数。
from flask.signals import _signals app = 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="1") return 'Index' if __name__ == '__main__': app.run()
六。flask-session
这个模块的作用就是将原本保存cookies的值保存在redis中。
需要安装模块：
pip install flash-session
应用如下：
from flask import Flask,session from flask_session import RedisSessionInterface import redis app = Flask(__name__) app.secret_key="ajksda" conn=redis.Redis(host='127.0.0.1',port=6379) #use_signer是否对key签名 app.session_interface=RedisSessionInterface(conn,key_prefix='jason',use_signer=True, permanent=False) @app.route('/') def hello_world(): session['nb']='jason' return 'Hello World!' @app.route("/index") def index(): print(session['nb']) return "ok" if __name__ == '__main__': app.run()
首先需要调用redis创建一个数据库连接，和密钥
将app_session_interface接口改成flask-session，其中设置一个前缀。
保存session时，首先需要将前端返回的session中id获取，将其前缀和id作为redis的name，将其session序列化传入到redis中，并再前端设置一样的value，所以该值和redis中的值是一样的。
当取值时，通过实现保存到key值，从redis获取到val，再反序列化获得其值。
session的参数;
use_signer=True,当这个值为True时，需要设置密钥，否则不需要设置密钥。
permanent=True，当关闭浏览器时，这个session是否失效。True代表有效，False代表无效。
简便方法：使用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='6379') 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()
这个只需要传入存储方式，存储路径，和存储前缀，最后生成Session，就可以完成配置