slashboywang个人页面-阿里云开发者社区

百度AI智能学习

春色将阑，莺声渐老，红英落尽青梅小。画堂人静雨蒙蒙，屏山半掩余香袅。密约沉沉，离情杳杳，菱花尘满慵将照。倚楼无语欲销魂，长空黯淡连芳草。首先安装python包 pip3 install baidu-aip 再注册百度AI得到相关参数 https://ai.baidu.com/ 一、语音合成 from aip import AipSpeech APP_ID = '14446020' API_KEY = 'GnaoLWrIiTKP10disiDHMiNZ' SECRET_KEY = 'FYaMNBsH5NFsgWcRsyBfaHDV70MvvE6u' #实例化AipSpeech对象 client = AipSpeech(APP_ID, API_KEY, SECRET_KEY) #调用方法语音合成 result = client.synthesis("欢迎来到王者荣耀", options={ "per": 1, "spd": 9, "pit": 9, "vol": 15, }) if not isinstance(result, dict): with open('auido.mp3', 'wb') as f: f.write(result) 二、语音识别 from aip import AipSpeech import os APP_ID = '14446020' API_KEY = 'GnaoLWrIiTKP10disiDHMiNZ' SECRET_KEY = 'FYaMNBsH5NFsgWcRsyBfaHDV70MvvE6u' client = AipSpeech(APP_ID, API_KEY, SECRET_KEY) def get_file_content(filePath): cmd_str = f"ffmpeg -y -i {filePath} -acodec pcm_s16le -f s16le -ac 1 -ar 16000 {filePath}.pcm" os.system(cmd_str) with open(f"{filePath}.pcm", 'rb') as fp: return fp.read() res = client.asr(speech=get_file_content("auido.mp3"),options={ "dev_pid":1536, }) print(res["result"][0]) 这里用到一个格式转换的软件，百度地址：https://pan.baidu.com/s/1MadxSh-A0Pzo1Su_wKdktQ 提取码：x5xi 固定的格式转换命令：（需要将bin文件添加环境变量，在cmd中执行） ffmpeg -y -i filePath -acodec pcm_s16le -f s16le -ac 1 -ar 16000 filePath.pcm 三、短文本相似度 from aip import AipNlp APP_ID = '14446020' API_KEY = 'GnaoLWrIiTKP10disiDHMiNZ' SECRET_KEY = 'FYaMNBsH5NFsgWcRsyBfaHDV70MvvE6u' client = AipNlp(APP_ID,API_KEY,SECRET_KEY) ret = client.simnet("你今年几岁了？","多大年龄了？") print(ret) {'log_id': 4545309161914786697, 'texts': {'text_2': '多大年龄了？', 'text_1': '你今年几岁了？'}, 'score': 0.742316} score 是两个测试的短文本相似度，一般大于0.72的两个短文本的意思是相似的句子！四、代码实现对接图灵 import requests def tuling_test(question): url = "http://openapi.tuling123.com/openapi/api/v2" data = { "reqType":0, "perception": { "inputText": { "text": question }, "inputImage": { }, }, "userInfo": { "apiKey": "2f4e809b8b3049ce82a6b4787bad65bb", "userId": "wangjifei" } } return requests.post(url=url,json=data).json() ret = tuling_test("心情不好") print(ret.get("results")[0]["values"]["text"]) 五、简单实现人机交流基本步骤：用户录制音频---传入函数---格式转化---语音识别---匹配答案---语音合成---语音文件流写入文件---os执行文件---删除文件 from aip import AipSpeech from aip import AipNlp from uuid import uuid4 import os import requests import time APP_ID = '14446007' API_KEY = 'QrQWLLg5a8qld7Qty7avqCGC' SECRET_KEY = 'O5mE31LSl17hm8NRYyf9PwlE5Byqm0nr' client = AipSpeech(APP_ID, API_KEY, SECRET_KEY) nlp_client = AipNlp(APP_ID, API_KEY, SECRET_KEY) def tuling_test(question): """接入图灵，为问题匹配答案""" url = "http://openapi.tuling123.com/openapi/api/v2" data = { "reqType": 0, "perception": { "inputText": { "text": question }, "inputImage": { }, }, "userInfo": { "apiKey": "2f4e809b8b3049ce82a6b4787bad65bb", "userId": "wangjifei" } } ret = requests.post(url=url, json=data).json() return ret.get("results")[0]["values"]["text"] def get_file_content(filePath): """音频的格式转换""" cmd_str = f"ffmpeg -y -i {filePath} -acodec pcm_s16le -f s16le -ac 1 -ar 16000 {filePath}.pcm" os.system(cmd_str) with open(f"{filePath}.pcm", 'rb') as fp: return fp.read() def custom_reply(text): """根据问题得到相应的答案，可以通过短文本相似来自定义，也可以调用图灵问题库""" if nlp_client.simnet("你叫什么名字", text).get("score") >= 0.72: return "我不能告诉你" return tuling_test(text) def learn_say(file_name): """机器人学说话""" # 语音识别成文字 res = client.asr(speech=get_file_content(file_name), options={ "dev_pid": 1536, }) os.remove(f"{file_name}.pcm") text = res.get("result")[0] # 根据问题得到相关答案 text1 = custom_reply(text) # 答案语音合成 res_audio = client.synthesis(text1, options={ "vol": 8, "pit": 8, "spd": 5, "per": 4 }) # 通过uuid 生成文件名 ret_file_name = f"{uuid4()}.mp3" # 将生成的语音流写入文件中 with open(ret_file_name, "wb") as f: f.write(res_audio) # 执行音频文件 ret = os.system(ret_file_name) time.sleep(2) os.remove(ret_file_name) if __name__ == '__main__': learn_say("auido.m4a") 六、网页版智能机器人对话 flask_ws.py from flask import Flask, request, render_template from uuid import uuid4 from geventwebsocket.websocket import WebSocket from gevent.pywsgi import WSGIServer from geventwebsocket.handler import WebSocketHandler from learn_say import learn_say app = Flask(__name__) # type:Flask @app.route("/ws") def ws(): user_socket = request.environ.get("wsgi.websocket") # type:WebSocket while True: msg = user_socket.receive() q_file_name = f"{uuid4()}.wav" with open(q_file_name, "wb") as f: f.write(msg) ret_file_name = learn_say(q_file_name) user_socket.send(ret_file_name) if __name__ == '__main__': http_serv = WSGIServer(("127.0.0.1", 8006), app, handler_class=WebSocketHandler) http_serv.serve_forever() flask_app.py from flask import Flask, request, render_template, send_file app = Flask(__name__) # type:Flask @app.route("/index") def index(): return render_template("index.html") @app.route("/get_audio/<audio_name>") def get_audio(audio_name): return send_file(audio_name) if __name__ == '__main__': app.run("127.0.0.1", 8008, debug=True) learn_say.py from aip import AipSpeech from aip import AipNlp from uuid import uuid4 import os import requests import time APP_ID = '14446007' API_KEY = 'QrQWLLg5a8qld7Qty7avqCGC' SECRET_KEY = 'O5mE31LSl17hm8NRYyf9PwlE5Byqm0nr' client = AipSpeech(APP_ID, API_KEY, SECRET_KEY) nlp_client = AipNlp(APP_ID, API_KEY, SECRET_KEY) def tuling_test(question): """接入图灵，为问题匹配答案""" url = "http://openapi.tuling123.com/openapi/api/v2" data = { "reqType": 0, "perception": { "inputText": { "text": question }, "inputImage": { }, }, "userInfo": { "apiKey": "2f4e809b8b3049ce82a6b4787bad65bb", "userId": "wangjifei" } } ret = requests.post(url=url, json=data).json() return ret.get("results")[0]["values"]["text"] def get_file_content(filePath): """音频的格式转换""" cmd_str = f"ffmpeg -y -i {filePath} -acodec pcm_s16le -f s16le -ac 1 -ar 16000 {filePath}.pcm" os.system(cmd_str) with open(f"{filePath}.pcm", 'rb') as fp: return fp.read() def custom_reply(text): """根据问题得到相应的答案，可以通过短文本相似来自定义，也可以调用图灵问题库""" if nlp_client.simnet("你叫什么名字", text).get("score") >= 0.72: return "我不能告诉你" return tuling_test(text) def learn_say(file_name): """机器人学说话""" # 语音识别成文字 res = client.asr(speech=get_file_content(file_name), options={ "dev_pid": 1536, }) os.remove(file_name) os.remove(f"{file_name}.pcm") text = res.get("result")[0] # 根据问题得到相关答案 text1 = custom_reply(text) # 答案语音合成 res_audio = client.synthesis(text1, options={ "vol": 8, "pit": 8, "spd": 5, "per": 4 }) # 通过uuid 生成文件名 ret_file_name = f"{uuid4()}.mp3" # 将生成的语音流写入文件中 with open(ret_file_name, "wb") as f: f.write(res_audio) return ret_file_name index.html <!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Title</title> </head> <body> <audio src="" autoplay controls id="player"></audio> <button onclick="start_reco()">录制消息</button> <br> <button onclick="stop_reco()">发送语音消息</button> </body> <script src="/static/Recorder.js"></script> <script type="application/javascript"> var serv = "http://127.0.0.1:8008"; var ws_serv = "ws://127.0.0.1:8006/ws"; var get_music = serv + "/get_audio/"; var ws = new WebSocket(ws_serv); ws.onmessage = function (data) { document.getElementById("player").src = get_music + data.data }; var reco = null; var audio_context = new AudioContext(); navigator.getUserMedia = (navigator.getUserMedia || navigator.webkitGetUserMedia || navigator.mozGetUserMedia || navigator.msGetUserMedia); navigator.getUserMedia({audio: true}, create_stream, function (err) { console.log(err) }); function create_stream(user_media) { var stream_input = audio_context.createMediaStreamSource(user_media); reco = new Recorder(stream_input); } //录制消息 function start_reco() { reco.record(); } //先停止录制，再获取音频 function stop_reco() { reco.stop(); get_audio(); reco.clear(); } //获取音频，发送音频 function get_audio() { reco.exportWAV(function (wav_file) { // wav_file = Blob对象 ws.send(wav_file); }) } </script> </html>

Flask的上下文管理

庭院深深深几许，杨柳堆烟，帘幕无重数。玉勒雕鞍游冶处，楼高不见章台路。雨横风狂三月暮，门掩黄昏，无计留春住。泪眼问花花不语，乱红飞过秋千去。一、通过Threading的local实现的本地存储隔离当我们开启多线程来执行func函数，通过自定义的Foo类来存储数据时，我们发现最终的输出结果是全部的线程打印的都是一个最终的数字10，这是因为这样存储的数据线程之间是共享的，当最后一个线程执行func函数时，由于func函数time.sleep了1秒钟，第一个线程还没有来得及打印数据呢，就被最后一个线程传的参数给覆盖了，从而线程们打印的结果都是相同的。 from threading import local import threading import time class Foo(object): pass def func(num): Foo.num = num time.sleep(1) print(Foo.num,threading.current_thread().ident) for i in range(1,11): th = threading.Thread(target=func,args=(i,)) th.start() 输出结果.png 我们使用Threading中的local来存储数据就可以做到数据在线程之间数据隔离。从而就可以理解为什么flask的上下文管理是基于local来实现的数据存储了。 from threading import local import threading import time foo = local() def func(num): foo.num = num time.sleep(1) print(foo.num,threading.current_thread().ident) for i in range(1,11): th = threading.Thread(target=func,args=(i,)) th.start() 输出结果.png 二、实例化一个Flask应用使用app = Flask(__name__)，可以实例化一个Flask应用。实例化的Flask应用有一些要点或特性需要注意一下：对于请求和响应的处理，Flask使用werkzeug库中的Request类和Response类。对于URL模式的处理，Flask应用使用werkzeug库中的Map类和Rule类，每一个URL模式对应一个Rule实例，这些Rule实例最终会作为参数传递给Map类构造包含所有URL模式的一个“地图”。这个地图可以用来匹配请求中的URL信息。当实例化一个Flask应用app（这个应用的名字可以随便定义）之后，对于如何添加URL模式，Flask采取了一种更加优雅的模式，对于这点可以和Django的做法进行比较。Flask采取装饰器的方法，将URL规则和视图函数结合在一起写，其中主要的函数是route。这样写视图函数，会将'/'这条URL规则和视图函数index()联系起来，并且会形成一个Rule实例，再添加进Map实例中去。当访问'/'时，会执行index()。实例化Flask应用时，会创造一个Jinja环境，这是Flask自带的一种模板引擎。实例化的Flask应用是一个可调用对象。在前面讲到，Web应用要遵循WSGI规范，就要实现一个函数或者一个可调用对象webapp(environ, start_response)，以方便服务器或网关调用。Flask应用通过__call__(environ, start_response)方法可以让它被服务器或网关调用。注意到调用该方法会执行wsgi_app(environ, start_response)方法，之所以这样设计是为了在应用正式处理请求之前，可以加载一些“中间件”,以此改变Flask应用的相关特性。 Flask应用还有一些其他的属性或方法，用于整个请求和响应过程。三、调用Flask应用时会发生什么上面部分分析了实例化的Flask应用长什么样子。当一个完整的Flask应用实例化后，可以通过调用app.run()方法运行这个应用。 Flask应用的run()方法会调用werkzeug.serving模块中的run_simple方法。这个方法会创建一个本地的测试服务器，并且在这个服务器中运行Flask应用。当服务器开始调用Flask应用后，便会触发Flask应用的_call_(environ, start_response)方法。其中environ由服务器产生，start_response在服务器中定义。上面我们分析到当Flask应用被调用时会执行wsgi_app(environ, start_response)方法。可以看出，wsgi_app是真正被调用的WSGI应用，之所以这样设计，就是为了在应用正式处理请求之前，wsgi_app可以被一些“中间件”装饰，以便先行处理一些操作。四、上下文（application context 和 request context）在 flask 中，视图函数需要知道它执行情况的请求信息（请求的 url，参数，方法等）以及应用信息（应用中初始化的数据库等），才能够正确运行。最直观地做法是把这些信息封装成一个对象，作为参数传递给视图函数。但是这样的话，所有的视图函数都需要添加对应的参数，即使该函数内部并没有使用到它。 flask 的做法是把这些信息作为类似全局变量的东西，视图函数需要的时候，可以使用 from flask import request 获取。但是这些对象和全局变量不同的是——它们必须是动态的，因为在多线程或者多协程的情况下，每个线程或者协程获取的都是自己独特的对象，不会互相干扰。那么如何实现这种效果呢？如果对 python 多线程比较熟悉的话，应该知道多线程中有个非常类似的概念 threading.local，可以实现多线程访问某个变量的时候只看到自己的数据。内部的原理说起来也很简单，这个对象有一个字典，保存了线程 id 对应的数据，读取该对象的时候，它动态地查询当前线程 id 对应的数据。 flask 中有两种上下文：application context 和 request context。上下文有关的内容定义在 globals.py 文件。 from functools import partial from werkzeug.local import LocalStack, LocalProxy _request_ctx_err_msg = '''\ Working outside of request context. This typically means that you attempted to use functionality that needed an active HTTP request. Consult the documentation on testing for information about how to avoid this problem.\ ''' _app_ctx_err_msg = '''\ Working outside of application context. This typically means that you attempted to use functionality that needed to interface with the current application object in some way. To solve this, set up an application context with app.app_context(). See the documentation for more information.\ ''' def _lookup_req_object(name): top = _request_ctx_stack.top if top is None: raise RuntimeError(_request_ctx_err_msg) return getattr(top, name) def _lookup_app_object(name): top = _app_ctx_stack.top if top is None: raise RuntimeError(_app_ctx_err_msg) return getattr(top, name) def _find_app(): top = _app_ctx_stack.top if top is None: raise RuntimeError(_app_ctx_err_msg) return top.app # context locals _request_ctx_stack = LocalStack() # {"__storage__":{ },"__ident_func__" : get_ident} _app_ctx_stack = LocalStack() current_app = LocalProxy(_find_app) request = LocalProxy(partial(_lookup_req_object, 'request')) session = LocalProxy(partial(_lookup_req_object, 'session')) g = LocalProxy(partial(_lookup_app_object, 'g')) flask 提供两种上下文：application context 和 request context 。application context 又演化出来两个变量 current_app 和 g，而 request context 则演化出来 request 和 session。这里的实现用到了两个东西：LocalStack 和 LocalProxy。它们两个的结果就是我们可以动态地获取两个上下文的内容，在并发程序中每个视图函数都会看到属于自己的上下文，而不会出现混乱。 LocalStack 和 LocalProxy 都是 werkzeug 提供的，定义在 local.py 文件中。在分析这两个类之前，我们先介绍这个文件另外一个基础的类 Local。Local 就是实现了类似 threading.local 的效果——多线程或者多协程情况下全局变量的隔离效果。下面是它的代码： import copy from functools import update_wrapper from werkzeug.wsgi import ClosingIterator from werkzeug._compat import PY2, implements_bool try: from greenlet import getcurrent as get_ident except ImportError: try: from thread import get_ident except ImportError: from _thread import get_ident class Local(object): __slots__ = ('__storage__', '__ident_func__') def __init__(self): // 数据保存在 __storage__ 中，后续访问都是对该属性的操作 object.__setattr__(self, '__storage__', {}) // {"__storage__":{}} object.__setattr__(self, '__ident_func__', get_ident) // {"__storage__":{} , "__ident_func__" : get_ident} def __iter__(self): return iter(self.__storage__.items()) def __call__(self, proxy): """Create a proxy for a name.""" return LocalProxy(self, proxy) // 清空当前线程/协程保存的所有数据 def __release_local__(self): self.__storage__.pop(self.__ident_func__(), None) // 下面三个方法实现了属性的访问、设置和删除。 // 注意到，内部都调用 `self.__ident_func__` 获取当前线程或者协程的 id，然后再访问对应的内部字典。 // 如果访问或者删除的属性不存在，会抛出 AttributeError。 // 这样，外部用户看到的就是它在访问实例的属性，完全不知道字典或者多线程/协程切换的实现 def __getattr__(self, name): try: return self.__storage__[self.__ident_func__()][name] except KeyError: raise AttributeError(name) def __setattr__(self, name, value): ident = self.__ident_func__() storage = self.__storage__ try: storage[ident][name] = value except KeyError: storage[ident] = {name: value} def __delattr__(self, name): try: del self.__storage__[self.__ident_func__()][name] except KeyError: raise AttributeError(name) 可以看到，Local 对象内部的数据都是保存在 _storage_ 属性的，这个属性变量是个嵌套的字典：{"ident":{"key1":"value1","key2":"value2",...}}。最外面字典 key 是线程或者协程的 identity，value 是另外一个字典，这个内部字典就是用户自定义的 key-value 键值对。用户访问实例的属性，就变成了访问内部的字典，外面字典的 key 是自动关联的。self._ident_func_() 就是执行get_ident()，得到的是协程的 get_current 或者线程的 get_ident，从而获取当前代码所在线程或者协程的 id。除了这些基本操作之外，Local 还实现了_release_local_ ，用来清空（析构）当前线程或者协程的数据（状态）。_call_ 操作来创建一个 LocalProxy 对象，LocalProxy 会在下面讲到。理解了 Local，我们继续回来看另外两个类。 LocalStack 是基于 Local 实现的栈结构。如果说 Local 提供了多线程或者多协程隔离的属性访问，那么 LocalStack 就提供了隔离的栈访问。下面是它的实现代码，可以看到它提供了 push、pop 和 top 方法。 _release_local_ 可以用来清空当前线程或者协程的栈数据，_call_ 方法返回当前线程或者协程栈顶元素的代理对象。 class LocalStack(object): def __init__(self): self._local = Local() // {"__storage__":{},"__ident_func__":get_ident} def __release_local__(self): self._local.__release_local__() def __call__(self): def _lookup(): rv = self.top if rv is None: raise RuntimeError('object unbound') return rv return LocalProxy(_lookup) // push、pop 和 top 三个方法实现了栈的操作， // 可以看到栈的数据是保存在 self._local.stack 属性中的 def push(self, obj): """Pushes a new item to the stack""" rv = getattr(self._local, 'stack', None) // {"__storage__":{},"__ident_func__":get_ident} if rv is None: self._local.stack = rv = [] rv.append(obj) return rv def pop(self): stack = getattr(self._local, 'stack', None) if stack is None: return None elif len(stack) == 1: release_local(self._local) return stack[-1] else: return stack.pop() @property def top(self): try: return self._local.stack[-1] except (AttributeError, IndexError): return None 我们在之前看到了 request context 的定义，它就是一个 LocalStack 的实例： _request_ctx_stack = LocalStack() // {"__storage__":{},"__ident_func__":get_ident} 它会把当前线程或者协程的请求都保存在栈里，等使用的时候再从里面读取。 LocalProxy 是一个 Local 对象的代理，负责把所有对自己的操作转发给内部的 Local 对象。LocalProxy 的构造函数介绍一个 callable 的参数，这个 callable 调用之后需要返回一个 Local 实例，后续所有的属性操作都会转发给 callable 返回的对象。 @implements_bool class LocalProxy(object): __slots__ = ('__local', '__dict__', '__name__', '__wrapped__') def __init__(self, local, name=None): object.__setattr__(self, '_LocalProxy__local', local) object.__setattr__(self, '__name__', name) if callable(local) and not hasattr(local, '__release_local__'): # "local" is a callable that is not an instance of Local or # LocalManager: mark it as a wrapped function. object.__setattr__(self, '__wrapped__', local) def _get_current_object(self): """Return the current object""" if not hasattr(self.__local, '__release_local__'): return self.__local() try: return getattr(self.__local, self.__name__) except AttributeError: raise RuntimeError('no object bound to %s' % self.__name__) @property def __dict__(self): try: return self._get_current_object().__dict__ except RuntimeError: raise AttributeError('__dict__') def __getattr__(self, name): if name == '__members__': return dir(self._get_current_object()) return getattr(self._get_current_object(), name) def __setitem__(self, key, value): self._get_current_object()[key] = value def __delitem__(self, key): del self._get_current_object()[key] 这里实现的关键是把通过参数传递进来的 Local 实例保存在 __local 属性中，并定义了 _get_current_object() 方法获取当前线程或者协程对应的对象。看下面这段代码就能看明白，由于_request_ctx_stack 是多线程或者协程隔离的栈结构，request 每次都会调用 _lookup_req_object 栈头部的数据来获取保存在里面的 requst context。 _request_ctx_stack = LocalStack() // {"__storage__":{},"__ident_func__":get_ident} request = LocalProxy(partial(_lookup_req_object, 'request')) session = LocalProxy(partial(_lookup_req_object, 'session')) 那么请求上下文信息是什么时候被放在 stack 中呢？还记得之前介绍的 wsgi_app() 方法有下面代码吗？ def __call__(self, environ, start_response): return self.wsgi_app(environ, start_response) def wsgi_app(self, environ, start_response): ctx = self.request_context(environ) error = None try: try: ctx.push() response = self.full_dispatch_request() except Exception as e: error = e response = self.handle_exception(e) except: error = sys.exc_info()[1] raise return response(environ, start_response) finally: if self.should_ignore_error(error): error = None ctx.auto_pop(error) 每次在调用 app._call_ 的时候，都会把对应的请求信息压栈，最后执行完请求的处理之后把它出栈。我们来看看request_context，这个方法只有一行代码： def request_context(self, environ): return RequestContext(self, environ) 它调用了 RequestContext，并把 self 和请求信息的字典 environ 当做参数传递进去。追踪到 RequestContext 定义的地方，它出现在 ctx.py 文件中，代码如下： class RequestContext(object): def __init__(self, app, environ, request=None): self.app = app if request is None: request = app.request_class(environ) self.request = request self.url_adapter = app.create_url_adapter(self.request) self.flashes = None self.session = None def match_request(self): """Can be overridden by a subclass to hook into the matching of the request. """ try: url_rule, self.request.view_args = \ self.url_adapter.match(return_rule=True) self.request.url_rule = url_rule except HTTPException as e: self.request.routing_exception = e def push(self): top = _request_ctx_stack.top # {"__storage__":{},"__ident_func__":get_ident} if top is not None and top.preserved: top.pop(top._preserved_exc) # Before we push the request context we have to ensure that there # is an application context. app_ctx = _app_ctx_stack.top if app_ctx is None or app_ctx.app != self.app: app_ctx = self.app.app_context() app_ctx.push() self._implicit_app_ctx_stack.append(app_ctx) else: self._implicit_app_ctx_stack.append(None) if hasattr(sys, 'exc_clear'): sys.exc_clear() _request_ctx_stack.push(self) if self.session is None: session_interface = self.app.session_interface self.session = session_interface.open_session( self.app, self.request ) if self.session is None: self.session = session_interface.make_null_session(self.app) def pop(self, exc=_sentinel): app_ctx = self._implicit_app_ctx_stack.pop() try: clear_request = False if not self._implicit_app_ctx_stack: self.preserved = False self._preserved_exc = None if exc is _sentinel: exc = sys.exc_info()[1] self.app.do_teardown_request(exc) if hasattr(sys, 'exc_clear'): sys.exc_clear() request_close = getattr(self.request, 'close', None) if request_close is not None: request_close() clear_request = True finally: rv = _request_ctx_stack.pop() if clear_request: rv.request.environ['werkzeug.request'] = None if app_ctx is not None: app_ctx.pop(exc) assert rv is self, 'Popped wrong request context. ' \ '(%r instead of %r)' % (rv, self) def auto_pop(self, exc): if self.request.environ.get('flask._preserve_context') or \ (exc is not None and self.app.preserve_context_on_exception): self.preserved = True self._preserved_exc = exc else: self.pop(exc) def __enter__(self): self.push() return self def __exit__(self, exc_type, exc_value, tb): self.auto_pop(exc_value) if BROKEN_PYPY_CTXMGR_EXIT and exc_type is not None: reraise(exc_type, exc_value, tb) def __repr__(self): return '<%s \'%s\' [%s] of %s>' % ( self.__class__.__name__, self.request.url, self.request.method, self.app.name, ) 每个 request context 都保存了当前请求的信息，比如 request 对象和 app 对象。在初始化的最后，还调用了 match_request 实现了路由的匹配逻辑。 push 操作就是把该请求的 ApplicationContext（如果 _app_ctx_stack 栈顶不是当前请求所在 app ，需要创建新的 app context）和 RequestContext 有关的信息保存到对应的栈上，压栈后还会保存 session 的信息； pop 则相反，把 request context 和 application context 出栈，做一些清理性的工作。到这里，上下文的实现就比较清晰了：每次有请求过来的时候，flask 会先创建当前线程或者进程需要处理的两个重要上下文对象，把它们保存到隔离的栈里面，这样视图函数进行处理的时候就能直接从栈上获取这些信息。

slashboywang个人页面-阿里云开发者社区

个人介绍

擅长的技术

2018-10-23

python之time和datetime模块

百度AI智能学习

Flask的上下文管理

Vue全知识脉络梳理

Django中ajax技术和form表单两种方式向后端提交文件

分页功能的实现

MySQL的索引原理与查询优化

Django中的admin自定制相关

Django之admin源码解析

一张图搞定Vue的生命周期

Django之数据库ORM基础到进阶操作

Django自带的auth登录认证

django面试题(21道)

RBAC角色权限设计

npm、webpack、vue-cli 快速上手版

基于Django实现 RESTful API 之RestFramework框架2

基于Django实现 RESTful API 之RestFramework框架3

python链表冒泡排序、二叉树顺序递归遍历、顺序表的快排

初识Flask

Flask基础知识点1

Vue全知识脉络梳理

Flask的上下文管理

百度AI智能学习

python之time和datetime模块