从Blueprint部署的排队后台工作线程中缺少Flask应用程序上下文

如何解决从Blueprint部署的排队后台工作线程中缺少Flask应用程序上下文

我有一个使用了工厂功能模式的Flask应用程序（来自入门教程），并且正在尝试将一个长时间运行的工作卸载到具有Redis队列的后台工作人员。我正在从蓝图调用后台工作，并且无法随调用一起传递应用程序上下文。目的是使用应用程序上下文和SQLite配置从后台线程对其执行写操作。我在这里想念什么？我认为这可能更像是“您对Flask的工作原理还不了解”问题，如果是这种情况，请让我知道我做错了！谢谢。

错误

RuntimeError: Working outside of application context.

dy.py

import sqlite3

import click
from flask import current_app,g
from flask.cli import with_appcontext


def get_db():
    if 'db' not in g:
        g.db = sqlite3.connect(
            current_app.config['DATABASE'],detect_types=sqlite3.PARSE_DECLTYPES
        )
        g.db.row_factory = sqlite3.Row
    
    return g.db

蓝图模块：

from flask import (
    Blueprint,flash,g,redirect,render_template,request,url_for,Response,current_app
)
import sqlite3

from app.db import get_db

from rq import Queue
from redis import Redis

bp = Blueprint('perform_work',__name__)
q = Queue(connection=Redis())


def some_func():

    db = get_db()
    ...
    

def generate_work():

    result = q.enqueue(some_func)
    ...


@bp.route('/perform_work',methods=['POST'])
def perform_work():
    ...
    generate_work()

worker.py

import os

import redis
from rq import Worker,Queue,Connection

listen = ['default']

redis_url = os.getenv('REDISTOGO_URL','redis://localhost:6379')

conn = redis.from_url(redis_url)


if __name__ == '__main__':
    with Connection(conn):
        worker = Worker(list(map(Queue,listen)))
        worker.work()

解决方法

好的，我很高兴我猜出来了。应用程序的上下文从未在工作程序中注册过，这应该发生在worker.py中，因此，是的，工作人员对应用程序本身一无所知。这是注册应用程序上下文的更新后的worker.py：

import os

import redis
from rq import Worker,Queue,Connection
from app import create_app

listen = ['default']

redis_url = os.getenv('REDISTOGO_URL','redis://localhost:6379')

conn = redis.from_url(redis_url)

app = create_app()
app.app_context().push()

if __name__ == '__main__':
    with Connection(conn):
        worker = Worker(list(map(Queue,listen)))
        worker.work()