如何从 RQ 作业中连接到 flask-sqlalchemy 数据库答案

【问题标题】：How to connect to flask-sqlalchemy database from inside a RQ job如何从 RQ 作业中连接到 flask-sqlalchemy 数据库
【发布时间】：2021-02-02 18:38:59
【问题描述】：

使用flask-sqlalchemy，如何从redis 任务中连接到数据库？数据库连接在create_app 中创建，其中： db = SQLAlchemy(app)

我从路由中调用作业：

@app.route("/record_occurrences")
def query_library():
    job = queue.enqueue(ApiQueryService(word), word)

然后在redis任务里面，我要更新数据库

class ApiQueryService(object):
    def __init__(self,word):
        resp = call_api()
        db.session.query(Model).filter_by(id=word.id).update({"count":resp[1]})

我找不到访问数据库的方法。我试过用from app import db 导入它。我尝试将其存储在g 中。我尝试用SQLAlchemy(app) 和其他一些东西重新实例化它，但这些都不起作用。当我使用 sqlite 时，所有这些都有效，我可以使用简单称为 sqlite3.connect() 的 get_db 方法轻松地从任何模块连接到数据库。是否有一些简单的方法可以使用类似的 SQLAlchemy 访问它？

【问题讨论】：

你试过from flask import current_app; current_app.db吗？ Flask 会自动尝试从全局上下文中定位应用程序。你可以read more here
是的，我确实尝试过。我用from flask import current_app 导入。但是，调用 current_app 会引发 Working outside of application context. 错误。所以我试图把它放在app.app_context 中，但是我得到一个错误，app 是未定义的。导入 app 然后会产生循环导入错误。
在 rq 任务中调用db = SQLAlchemy() 时有没有办法直接将SQLALCHEMY_DATABASE_URI 传递给它？
您通常可以通过在没有与db = SQLAlchemy() 关联的应用程序的情况下在任务脚本中实例化实际的db 对象来解决循环导入问题，然后在您的工厂函数中导入该db，调用db.init_app(app)。多次声明连接字符串是一种反模式，最好找出根本原因。
如何访问db.init_app(app) 中的app 变量？当我将from app import app 添加到任务文件时，我得到cannot import name 'app' from partially initialized module 。

标签： python-3.x database import sqlalchemy flask-sqlalchemy

【解决方案1】：

如@vulpxn 所述，这可以使用 App Factory 模式解决。

假设我们有这样的配置类：

class Config(object):
    DEBUG = False
    TESTING = False
    DEVELOPMENT = False

    API_PAGINATION = 10

    PROPAGATE_EXCEPTIONS = True  # needed due to Flask-Restful not passing them up

    SQLALCHEMY_TRACK_MODIFICATIONS = False  # ref: https://stackoverflow.com/questions/33738467/how-do-i-know-if-i-can-disable-sqlalchemy-track-modifications/33790196#33790196

class ProductionConfig(Config):
    CSRF_COOKIE_SAMESITE = 'Strict'
    SESSION_PROTECTION = "strong"
    SESSION_COOKIE_SECURE = True
    SESSION_COOKIE_HTTPONLY = True
    SESSION_COOKIE_SAMESITE = 'Strict'

    SECRET_KEY = "super-secret"
    INVITES_SECRET = "super-secret"
    PASSWORD_RESET_SECRET = "super-secret"
    PUBLIC_VALIDATION_SECRET = "super-secret"

    FRONTEND_SERVER_URL = "https://127.0.0.1:4999"

    SQLALCHEMY_DATABASE_URI = "sqlite:///%s" % os.path.join(os.path.abspath(os.path.dirname(__file__)), "..",
                                                            "people.db")

我们创建我们的应用工厂：

from flask_sqlalchemy import SQLAlchemy
from flask import Flask
from development.config import DevelopmentConfig
from rq import Queue
from email_queue.worker import conn

db = SQLAlchemy()

q = Queue(connection=conn)

def init_app(config=ProductionConfig):
    # app creation
    app = Flask(__name__)

    app.config.from_object(config)

    # plugin initialization
    db.init_app(app)
   
    with app.app_context():
        # adding blueprints
        from .blueprints import api
        app.register_blueprint(api, url_prefix='/api/v1')

        return app

我们现在可以使用应用工厂启动我们的应用了：

app = centrifuga4.init_app()
if __name__ == "__main__":
    with app.app_context():
        app.run()

但我们也可以（在我们的 Redis 工作中）执行以下操作：

def my_job():
    app = init_app()  
    with app.app_context():
        return something_using_sqlalchemy()

【讨论】：

这里的光标共享会发生什么？由于我们可能有很多很多任务，sqlalchemy 是否处理 db 任务的线程？
注意在任务中打开了一个上下文，因此每个任务都有一个专用的光标