0

当直接在 Flask 应用程序中的 API 调用上使用 SQLAlchemy 执行 SQL 查询时，它们会成功执行，但是当我尝试在我的函数中执行一个使用 Redis 作业队列执行的函数时，我收到以下错误：RuntimeError: No application found. Either work inside a view function or push an application context.函数存储在单独的文件中，而不是app.py.

我的app.py：

app = Flask(__name__)
app.secret_key = os.environ.get('FLASK_SECRET_KEY')

worker_queue = Queue('default', connection=conn)
worker_registry = StartedJobRegistry(queue=worker_queue)

# database configuration
setup_db(app)

在我models.py的 setup_db 中看起来像这样：

db = SQLAlchemy()


def setup_db(app):
    database_name = 'dev'
    default_database_path = "postgresql://{}:{}@{}/{}".format('postgres', 'password', 'localhost:5432', database_name)
    database_path = os.getenv('DATABASE_URL', default_database_path)
    app.config['SQLALCHEMY_DATABASE_URI'] = database_path
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    db.app = app
    db.init_app(app)

然后在我调用作业时触发的函数中：

users = User.query.all()

其中 User 是我在models.py.

4

0 回答 0