我在练习Flask .我已经做了一个简单的Flask 应用程序,它发送一封邮箱给管理员,说"新用户已加入."一旦有人输入他们的名字并在表单上提交.

这是邮箱组件:

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


# Send aync email functions
def send_aysnc_email(app, msg):
    with app.app_context():
        mail.send(msg)


def send_mail(to, subject, template, **kwargs):
    app = current_app
    msg = Message(
        app.config["FLASKY_MAIL_SUBJECT_PREFIX"] + subject,
        sender=app.config["FLASKY_MAIL_SENDER"],
        recipients=[to],
    )

    msg.body = render_template(template + ".txt", **kwargs)
    msg.html = render_template(template + ".html", **kwargs)
    thr = Thread(target=send_aysnc_email, args=[app, msg])
    thr.start()
    return thr

当我运行这款应用程序,在表格上输入姓名并按下提交时,邮箱不会被发送.

UI of my app

我在日志(log)上看到一个错误,说

Exception in thread Thread-3 (send_aysnc_email):
Traceback (most recent call last):
  File "/usr/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
127.0.0.1 - - [02/Sep/2023 14:39:29] "POST / HTTP/1.1" 302 -
    self.run()
  File "/usr/lib/python3.10/threading.py", line 953, in run
    self._target(*self._args, **self._kwargs)
  File "/home/cadbay53/Desktop/practice/flask/app/emails.py", line 9, in send_aysnc_email
    with app.app_context():
  File "/home/cadbay53/Desktop/practice/flask/venv/lib/python3.10/site-packages/werkzeug/local.py", line 311, in __get__
    obj = instance._get_current_object()
  File "/home/cadbay53/Desktop/practice/flask/venv/lib/python3.10/site-packages/werkzeug/local.py", line 508, in _get_current_object
    raise RuntimeError(unbound_message) from None
RuntimeError: Working outside of application context.

This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.

我已经找到了解决这个问题的方法,但我不明白为什么当前的代码不起作用.

这是我的应用程序工厂:

from flask import Flask, render_template
from flask_bootstrap import Bootstrap
from flask_mail import Mail
from flask_moment import Moment
from flask_sqlalchemy import SQLAlchemy
from config import config

bootstrap = Bootstrap()
mail = Mail()
moment = Moment()
db = SQLAlchemy()


def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)

    bootstrap.init_app(app)
    mail.init_app(app)
    moment.init_app(app)
    db.init_app(app)

    from .main import main as main_blueprint

    app.register_blueprint(main_blueprint)

    return app

这是我创建应用程序的文件实例:

import os
from app import create_app
from app.models import User, Role
from flask_migrate import Migrate
from app import db

app = create_app("default")
migrate = Migrate(app, db)


@app.shell_context_processor
def make_shell_context():
    return dict(db=db, User=User, Role=Role)


@app.cli.command()
def test():
    """Run unit tests."""
    import unittest

    tests = unittest.TestLoader().discover("tests")
    unittest.TextTestRunner(verbosity=2).run(tests)

以下是配置:

import os

basedir = os.path.abspath(os.path.dirname(__file__))


class Config:
    SECRET_KEY = "***"
    MAIL_SERVER = "smtp.gmail.com"
    MAIL_PORT = 587
    MAIL_USE_TLS = True
    MAIL_USERNAME = "***@***.com"
    MAIL_PASSWORD = "****"
    FLASKY_MAIL_SUBJECT_PREFIX = "[Flasky]"
    FLASKY_MAIL_SENDER = "Flasky Admin <***@***.com>"
    FLASKY_ADMIN = "***@***.com"
    SQLALCHEMY_TRACK_MODIFICATIONS = False

    @staticmethod
    def init_app(app):
        pass


class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///" + os.path.join(basedir, "data-dev.sqlite")


class TestConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///"


class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = "sqlite:///" + os.path.join(basedir, "data.sqlite")


config = {
    "development": DevelopmentConfig,
    "testing": TestConfig,
    "production": ProductionConfig,
    "default": DevelopmentConfig,
}

以下是我的工作内容:

# Send aync email functions
def send_aysnc_email(app, msg):
    with app.app_context():
        mail.send(msg)


def send_mail(to, subject, template, **kwargs):
    app = current_app._get_current_object()
    msg = Message(
        app.config["FLASKY_MAIL_SUBJECT_PREFIX"] + subject,
        sender=app.config["FLASKY_MAIL_SENDER"],
        recipients=[to],
    )

    msg.body = render_template(template + ".txt", **kwargs)
    msg.html = render_template(template + ".html", **kwargs)
    thr = Thread(target=send_aysnc_email, args=[app, msg])
    thr.start()
    return thr

我不明白为什么app.app_context()不能访问当前的应用程序实例,而将当前的应用程序作为参数提供,并且可以很好地与current_app._get_current_object()一起工作.

推荐答案

当提供当前应用程序作为参数时,为什么app.app_context()不能访问当前应用程序实例?

app_context()实际上返回一个代理对象,它不应该传递给另一个线程(docs):

Flask 提供的某些对象是其他对象的代理.对于每个工作线程,访问代理的方式相同,但指向绑定到每个工作线程的唯一对象[...]

大多数情况下,您不必关心这一点,但在某些例外情况下,最好知道该对象实际上是一个代理: [.] 在某些情况下,例如发送信号或passing data to a background thread,需要引用代理对象.

这就是为什么你必须使用app._get_current_object().由于传统上它是一个私有方法,因此用# noqa来 suppress 林特的警告可能是个好主意:


def send_mail(to, subject, template, **kwargs):

    app = current_app._get_current_object()  # noqa: there is no workaround
    ...

Python相关问答推荐

将列中的滚动值集转换为单元格中的单个值

在有限数量的唯一字母的长字符串中,找到包含重复不超过k次的所有唯一字母的最长子字符串

使用Python从HTTP打印值

使用pandas MultiIndex进行不连续 Select

ValueRight:参数目标和输出必须具有相同的形状.接收:目标.形状=(无,512),输出.形状=(无,3)

如何在Power Query中按名称和时间总和进行分组

Twilio:CallInstance对象没有来自_的属性'

Python主进程和分支进程如何共享gc信息?

使用GEKKO在简单DTE系统中进行一致初始化

在Python和matlab中显示不同 colored颜色 的图像

Python会扔掉未使用的表情吗?

大Pandas 胚胎中产生组合

_repr_html_实现自定义__getattr_时未显示

如何在类和classy-fastapi -fastapi- followup中使用FastAPI创建路由

用NumPy优化a[i] = a[i-1]*b[i] + c[i]的迭代计算

如何在Python脚本中附加一个Google tab(已经打开)

Pandas—合并数据帧,在公共列上保留非空值,在另一列上保留平均值

导入...从...混乱

启用/禁用shiny 的自动重新加载

使用Python查找、替换和调整PDF中的图像'