python—如何在具有redis队列作业的flask应用程序中使用postgresql和sqlalchemy

t98cgbkg  于 2021-09-13  发布在  Java
关注(0)|答案(0)|浏览(156)

在flask app中直接对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 设置数据库如下所示:

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 .

暂无答案!

目前还没有任何答案,快来回答吧!

相关问题