2

Using flask-sqlalchemy, how is it possible to connect to a database from within a redis task? The database connection is created in create_app with: db = SQLAlchemy(app)

I call a job from a route:

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

Then inside the redis task, I want to make an update to the database

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

I can't find a way to access the db. I've tried importing it with from app import db. I tried storing it in g. I tried reinstantiating it with SQLAlchemy(app), and several other things, but none of these work. When I was using sqlite, all of this worked, and I could easily connect to the db from any module with a get_db method that simply called sqlite3.connect(). Is there some simple way to access it with SQLAlchemy that's similar to that?

Ando
  • 165
  • 1
  • 14
  • Have you tried `from flask import current_app; current_app.db`? Flask will automatically try to locate the application from the global context. You can [read more here](https://flask.palletsprojects.com/en/1.1.x/appcontext/) – vulpxn Oct 20 '20 at 01:03
  • Yeah, I did try that. I imported with `from flask import current_app`. Calling `current_app`, however, throws a `Working outside of application context.` error. So I tried to put that in `app.app_context`, but then I get an error that `app` is undefined. Importing `app` then creates a circular import error. – Ando Oct 20 '20 at 17:29
  • Is there a way when calling `db = SQLAlchemy()` inside the rq task to just directly pass it the `SQLALCHEMY_DATABASE_URI`? – Ando Oct 20 '20 at 17:31
  • You can usually resolve the circular import issue by instantiating the actual `db` object within your tasks script without an associated app with `db = SQLAlchemy()`, and then importing that `db` inside your factory function, calling `db.init_app(app)`. Declaring your connection string multiple times is an antipattern, and it's much better to figure out the root cause. – vulpxn Oct 20 '20 at 17:34
  • 1
    How do I access that `app` variable in `db.init_app(app)`? When I add `from app import app` to the task file, I get `cannot import name 'app' from partially initialized module `. – Ando Oct 20 '20 at 18:14
  • If your application is growing to beyond a single file, you'll probably want to use a [factory function](https://flask.palletsprojects.com/en/1.1.x/patterns/appfactories/#factories-extensions) – vulpxn Oct 20 '20 at 18:15

1 Answers1

1

This can be solved using the App Factory pattern, as mentioned by @vulpxn.

Let's assume we have our configuration class somewhere like this:

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")

We create our app factory:

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

We will now be able to start our app using the app factory:

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

But we will also be able to (in our Redis job), do the following:

def my_job():
    app = init_app()  
    with app.app_context():
        return something_using_sqlalchemy()
miquelvir
  • 1,748
  • 1
  • 7
  • 21