Flask - RuntimeError：未找到应用程序。在视图函数中工作或推送应用程序上下文答案

【问题标题】：Flask - RuntimeError: No application found. Either work inside a view function or push an application contextFlask - RuntimeError：未找到应用程序。在视图函数中工作或推送应用程序上下文
【发布时间】：2020-01-09 06:50:52
【问题描述】：

我现在正在尝试制作一个使用 REST API 处理 HTTP 请求的应用程序。该应用程序将 Flask 用于 Web 框架，并将 Celery 用于异步任务。

这是应用程序结构。

app.py
celery_app.py
controller/
    controller.py
task/
    task.py

(1) app.py

Celery 配置没有行。

from flask import Flask
...

app = Flask(__name__)
...


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

(2) celery.py

from consts.consts import Consts
from kombu.utils.url import quote
from celery import Celery
from config.config import Config

config = Config()
db_uri = 'db+' + config.get_db_uri()

aws_access_key = quote(config.get_config(Consts.AWS, Consts.AWS_ACCESS_KEY))
aws_secret_key = quote(config.get_config(Consts.AWS, Consts.AWS_SECRET_KEY))
broker_url = "sqs://{aws_access_key}:{aws_secret_key}@".format(
    aws_access_key=aws_access_key, aws_secret_key=aws_secret_key
)

celery = Celery(
    broker=broker_url,
    backend=db_uri,
    include=['task']
)

(3) task/task.py

我把所有的任务都放在这里了。

from celery_app import celery
from flask import current_app as app
from model.model import db


@celery.task
def check_friend_status_from_db(user1, user2):
    status = db.engine.execute(
            "QUERY").fetchone()
    return status

现在，controller/controller.py 文件导入并调用任务如下。

(4) controller/controller.py

from flask import Blueprint, request, json, render_template, jsonify
from mysql.connector import Error
from sqlalchemy import or_, and_
from consts.consts import StatusCode
from consts.consts import Consts
from model.model import db, FriendRequest
from outbound.request import Request
from util.logging import logger
from task.task import check_friend_status_from_db


controller_blueprint = Blueprint('controller_blueprint', __name__)
outbound_request = Request()


@controller_blueprint.route('/friend/status/<requested_from>/<requested_to>', methods=['GET'])
def check_friend_status(requested_from, requested_to):
    logger.info('Checking the friend status')
    try:
        status = check_friend_status_from_db.apply_async((requested_from, requested_to)).get()

        if status is None:
            response = {
                Consts.STATUS_CODE: StatusCode.OK,
                Consts.FRIEND_STATUS: StatusCode.NO_RELATION
            }
        else:
            response = {
                Consts.STATUS_CODE: StatusCode.OK,
                Consts.FRIEND_STATUS: status
            }
    except Error as e:
        logger.error("TypeError:", e)
        response = {
            Consts.STATUS_CODE: StatusCode.ERROR
        }
    json_response = jsonify(response)
    logger.info(json_response)
    return json_response

当我运行代码时，我得到了我在标题中提到的错误。

RuntimeError: No application found. Either work inside a view function or push an application context

原来是控制器中try块下的这一部分，错误来自哪里。

status = check_friend_status_from_db.apply_async((requested_from, requested_to)).get()

请问有什么解决办法吗？

【问题讨论】：

标签： python flask celery

【解决方案1】：

您似乎需要注册您的蓝图controller_blueprint。由于此蓝图未注册到您的app，因此您在应用程序的上下文之外工作，因此出现错误。

你可以在你的 app.py 中这样做：

from controller.controller import controller_blueprint

app = Flask(__name__)
app.register_blueprint(controller_blueprint)

【讨论】：