python- flask current_app详解，与 current_app._get_current_object()的区别以及异步发送邮件实例

Posted 2022-01-13 胖虎是只mao

核心知识

AppContext手动、自动入栈
LocalStack是线程隔离的栈结构
current_app是线程、协程隔离对象
LocalProxy是获取当前线程隔离的代理对象

一、flask中经典错误 working outside application context

错误：

working outside application contex

原因：
在没有获取到应用上下文的情况下，进行了上下文操作。
代码：

from flask import Flask, current_app

app = Flask(__name__)

a = current_app
d = current_app.config['DEBUG']

报错：

app = Flask(__name__)
a= current_app
Traceback (most recent call last):
  File "/Applications/PyCharm.app/Contents/plugins/python/helpers/pydev/_pydev_bundle/pydev_console_utils.py", line 253, in do_get_completions
    return completer.complete(act_tok)
  File "/Applications/PyCharm.app/Contents/plugins/python/helpers/pydev/_pydev_bundle/_pydev_completer.py", line 89, in complete
    return self.attr_matches(text)
  File "/Applications/PyCharm.app/Contents/plugins/python/helpers/pydev/_pydev_bundle/_pydev_completer.py", line 151, in attr_matches
    words = dir2(obj, filter=filter)
  File "/Applications/PyCharm.app/Contents/plugins/python/helpers/pydev/_pydev_bundle/_pydev_imports_tipper.py", line 179, in generate_imports_tip_for_module
    if hasattr(obj_to_complete, '__dict__'):
  File "/Users/zhangweijian/PycharmProjects/ShiYanLou/PyFly/venv/lib/python3.8/site-packages/werkzeug/local.py", line 347, in __getattr__
    return getattr(self._get_current_object(), name)
  File "/Users/zhangweijian/PycharmProjects/ShiYanLou/PyFly/venv/lib/python3.8/site-packages/werkzeug/local.py", line 306, in _get_current_object
    return self.__local()
  File "/Users/zhangweijian/PycharmProjects/ShiYanLou/PyFly/venv/lib/python3.8/site-packages/flask/globals.py", line 51, in _find_app
    raise RuntimeError(_app_ctx_err_msg)
RuntimeError: Working outside of application context.
This typically means that you attempted to use functionality that needed
to interface with the current application object in a way.  To solve
this set up an application context with app.app_context().  See the
documentation for more information.
d = current_app.config['DEBUG']

AppContext、RequestContext、Flask与Request之间的关系

AppContext
应用上下文，是对flask一切对象的封装

RequestContext
请求上下文，是对request请求对象的封装
current_app
类型是LocalProxy
像全局变量一样工作，但只能在处理请求期间且在处理它的线程中访问
返回的栈顶元素不是应用上下文，而是flask的应用实例对象

应用上下文的封装=flask核心对象+和外部协作对象（再flask封装对象上再添加push、pop等）（请求上下文同理）

代码：进入current_app

# context locals
_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'))
g = LocalProxy(partial(_lookup_app_object, 'g'))

current_app与request：
current_app和request都是设计模式中代理设计的代理对象，指向flask核心对象和request的请求类

详解flask上下文与出入栈

在pycharm的flask项目中可以通过：
External Libraries->site-packages->flask->ctx.py
可以看到源码的实现
ctx.py中有AppContext、RequestContext两个函数，都实现了push()和pop()
AppContext：

class AppContext(object):
    """The application context binds an application object implicitly
    to the current thread or greenlet, similar to how the
    :class:`RequestContext` binds request information.  The application
    context is also implicitly created if a request context is created
    but the application is not on top of the individual application
    context.
    """

    def __init__(self, app):
        self.app = app
        self.url_adapter = app.create_url_adapter(None)
        self.g = app.app_ctx_globals_class()

        # Like request context, app contexts can be pushed multiple times
        # but there a basic "refcount" is enough to track them.
        self._refcnt = 0

    def push(self):
        """Binds the app context to the current context."""
        self._refcnt += 1
        if hasattr(sys, 'exc_clear'):
            sys.exc_clear()
        _app_ctx_stack.push(self)
        appcontext_pushed.send(self.app)

    def pop(self, exc=_sentinel):
        """Pops the app context."""
        try:
            self._refcnt -= 1
            if self._refcnt <= 0:
                if exc is _sentinel:
                    exc = sys.exc_info()[1]
                self.app.do_teardown_appcontext(exc)
        finally:
            rv = _app_ctx_stack.pop()
        assert rv is self, 'Popped wrong app context.  (%r instead of %r)' \\
            % (rv, self)
        appcontext_popped.send(self.app)

flask在RequestContext入栈前会检查另外一个AppContext的栈的情况，如果栈顶元素为空或者不是当前对象，就会把AppContext推入栈中，然后RequestContext才进栈。
LocalStack作用是线程隔离
LocalProxy 的作用就是可以根据线程/协程返回对应当前协程/线程的代理对象，也就是说

• 线程 A 往 LocalStack中塞入 A
• 线程 B 往 LocalStack 中塞入 B

无论在是什么地方，调用LocalProxy，

线程 A 永远取到得是 A，线程 B 取到得永远是 B

有关Local、LocalStack的详解可以参考：http://python.jobbole.com/87738/

这就是在 Flask 中可以在代码中直接使用 request、current_app 这样的变量的底层原因。

例如：
过程就好比导游与游客。

• 导游->AppContext

• 游客->RequestContext

• 工作->push

• 消费->push

• 向导->current_app

• 游玩->request

• 旅程->LocalStack
  每批游客都需要一位导游作为这批游客的专属向导，人生地不熟如果没有向导就麻烦了，因此游客开始去游玩前需要有导游带团。在这旅程中，导游和游客虽然分别是工作和消费，但导游（对象:栈顶元素）的任务就是给游客提供向导（属性:app），不提供其他服务，而游客负责游玩。当游客结束这旅程的同时导游的任务也完成（两个栈中的元素会被弹出）了。

手动AppContext进栈

代码：

from flask import Flask, current_app

app = Flask(__name__)

# flask应用实例入栈
ctx = app.app_context()
ctx.push()
a = current_app
ctx.pop()

运行：

app = Flask(__name__)
ctx = app.app_context()
ctx.push()
a = current_app
a
<Flask '<input>'>
app
<Flask '<input>'>
ctx.pop()
a
<LocalProxy unbound>

注意：
但最终current_app得到的栈顶元素不是应用上下文，而是flask的应用实例对象!
进入查看源码

current_app = LocalProxy(_find_app)

查看_find_app函数

def _find_app():
    top = _app_ctx_stack.top
    if top is None:
        raise RuntimeError(_app_ctx_err_msg)
    return top.app

可以发现最后返回的是top对象中的app。
reques和session同理，不过传多一个字符串进行查找。

flask自动入栈

如果是在一个请求中直接使用current_app对象是不用手动把AppContext推入栈中的。如之前所说RequestContext入栈前会检查另外一个AppContext的栈的情况，这个操作会由flask帮你完成。

手动进栈存在的价值

单元测试不在reques请求环境中执行，需要手动AppContext进栈。
离线应用。例如待会介绍的异步邮箱例子。

flask上下文与with语句

可以使用with 来实现自动入栈和出栈，比上面手动push、pop的更优雅，因为在AppContext中已实现两个特殊方法__enter__、exit，也被称为“魔法方法”，凡是实现了这两个特殊方法的对象都可以被with所使用。

    def __enter__(self):
        self.push()
        return self

    def __exit__(self, exc_type, exc_value, tb):
        self.pop(exc_value)

python with的使用这里不详细讨论,在flask中注意的是：

• 实现了上下文协议的对象使用with
• with被称做上下文管理器
• 只要实现enter、exit就是实现上下文协议
• 上下文表达式（app.app_context()）必须返回一个上下文管理器（AppContext）
• exit最后三个参数记录发生异常时的信息，exit返回bool类型，返回true表示正常，false会抛出异常，没有返回值默认为false

代码：

from flask import Flask, current_app

app = Flask(__name__)

with app.app_context():
    a = current_app

运行：

上图中全局变量a在with的作用域中的值为flask应用的实例对象main,当with关闭后就变成了LocalProxy unbound

flask中实际应用

发送密码重置电子邮件

from flask import render_template
from app import app

# ...

def send_password_reset_email(user):
	"""
	令牌，生成密码重置电子邮件
	"""
    token = user.get_reset_password_token()
    # 调用email.py中的send_email函数
    send_email('[Microblog] Reset Your Password',
               sender=app.config['ADMINS'][0],
               recipients=[user.email],
               text_body=render_template('email/reset_password.txt',
                                         user=user, token=token),
               html_body=render_template('email/reset_password.html',
                                         user=user, token=token))

异步电子邮件

from threading import Thread
from flask import current_app
from flask_mail import Message
from app import mail


def send_async_email(app, msg):
    with app.app_context():
        mail.send(msg)


def send_email(subject, sender, recipients, text_body, html_body,
               attachments=None, sync=False):
    msg = Message(subject, sender=sender, recipients=recipients)
    msg.body = text_body
    msg.html = html_body
    if attachments: # 附件
        for attachment in attachments:
            msg.attach(*attachment)
    if sync: # 是否异步
        mail.send(msg)
    else:
        Thread(target=send_async_email,
            args=(current_app._get_current_object(), msg)).start()

1. send_async_email()
mail.send()方法需要访问电子邮件服务器的配置值，而这必须通过访问应用属性的方式来实现。 使用with app.app_context()调用创建的应用上下文使得应用实例可以通过来自Flask的current_app变量来进行访问。

2. send_email()：
在send_email()函数中，应用实例作为参数传递给后台线程，后台线程将发送电子邮件而不阻塞主应用程序。在作为后台线程运行的send_async_email()函数中直接使用current_app将不会奏效，因为current_app是一个与处理客户端请求的线程绑定的上下文感知变量。在另一个线程中，current_app没有赋值。直接将current_app作为参数传递给线程对象也不会有效，因为current_app实际上是一个代理对象，它被动态地映射到应用实例。因此，传递代理对象与直接在线程中使用current_app相同。我需要做的是访问存储在代理对象中的实际应用程序实例，并将其作为app参数传递。 current_app._get_current_object()表达式从代理对象中提取实际的应用实例，所以它就是我作为参数传递给线程的。

current_app._get_current_object()

作用是 获取本线程的应用实例以作为参数传递给其他线程使用，应用实例就是_find_app中返回的top.app

因为current_app = LocalProxy(_find_app)中LocalProxy作用是获得线程隔离对象，在上面详解flask上下文与出入栈有说到

所以哪怕其他线程获得了current_app这个变量，他的线程隔离中也没有任何东西
而把本线程中的属性传递给其他线程却不关线程隔离的事，所以其他线程可以使用应用实例

代码：

原本是要调用_find_app才得到top.app的，现在通过current_app._get_current_object()不经过_find_app也可以获得，无关线程隔离

小结

• 以线程ID号作为key的字典->Local->LocalStack
• AppContext RequestContext -> LocalStack
• Flask -> AppContext Request -> RequestContext
• (LocalStack.top = AppContext top.app=Flask) -> current_app
• (LocalStack.top = RequestContext top.request=Request) -> request
• Flask会为每一个Request创建一个线程，每个线程都有可能需要Flask应用实例app，current_app为此存在

理解：

1. ctx = app.app_context() 实现的前提是有app，那么既然有了app为什么还用current_app呢？
   答：
   上面是为了方便快捷而在app = Flask(name)这个程序入口主线程来进行示范操作。
   好比旅游社如果只有一个创始人的导游在职的话，有没有员工准则手册都是一样，因为他就是准则的制定者。但如果招聘了多个导游，这时就需要颁发手册，让其他人知道准则了。这里的手册就是current_app

current_app的作用是为其他线程提供app = Flask(name)也就是flask应用实例对象app，以便于使用。 上面那个入栈的例子其实是为了快捷方便才在app = Flask(name)下操作演示的 如果说在上面的简单例子中current_app的存在确实是多此一举了，和app是一样的，因为都是在主线程中 app = Flask(name)只有这里实例化了整个flask应用实例对象，只在主线程启动时被实例化了一次。而其他子线程需要用到app就需要current_app了 文章中忘了注明每有一个Request进来都会创建一个新的线程来单独处理 形象理解： 好比旅游社如果只有一个导游在职的话，有没有员工准则手册都是一样，因为他就是准则的制定者。但如果招聘了多个导游，这时就需要颁发手册，让其他人知道准则了。 这里的手册就是current_app

current_app使用场景：
只有主线程的话，可以使用app, 但如果当有其他线程需要用到app时，就需要current_app了

我这样理解对吗 假如说是一个线程接待一个请求，那么异步意味着多线程同时接待多请求。 那么在异步的app.py中，request是这样被处理的，拿异步邮件举例，是在处理一个request的同时，触发/调用了发送异步发送邮件函数，开另一个线程来发送邮件，相当于同一段时间执行两个任务么，虽然对每一个请求而言都是一个线程为他们服务，但是对同一段时间来说，执行了多个任务。。但是对于就干一件事情request来说，异步没有节省时间。。 还有一个问题，current_app.get_current_app()返回的对象类型就和app一样，可以多次调用（给线程分配对象），在写其他异步程序时使用，这样对吗？

1、嗯，异步是为了省时间，如果没有其他耗时任务，只有一个任务，就不需要异步 2、对的