Initial Notes: The project uses Blueprints and below are the file structure and extracts of the code used...
File Structure:
/app
├── flaskapp/
│ ├── posts/
│ │ ├── __init__.py
│ │ ├── forms.py
│ │ ├── routes.py
│ │ ├── utils.py
│ ├── users/
│ │ ├── __init__.py
│ │ ├── forms.py
│ │ ├── routes.py
│ │ ├── utils.py
│ ├── main/
│ │ ├── __init__.py
│ │ ├── crons.py
│ │ ├── routes.py
│ ├── templates/
│ │ ├── users.html
│ ├── __init__.py
│ ├── config.py
│ ├── models.py
├── run.py
posts/utils.py
# Function to get all posts from DB
def get_all_posts():
post = Post.query.order_by(Post.id.asc())
return post
users/routes.py
# Importing 'get_all_posts' function from 'posts/utils.py'
from flaskapp.posts.utils import get_all_posts
users = Blueprint('users', __name__)
#All Users Route Related Posts
@posts.route("/posts", methods=['GET'])
@login_required
def all_users():
users = User.query.order_by(User.id.asc())
return render_template('users.html', USERS=users, POSTS=get_all_posts())
main/crons.py
# Importing 'get_all_posts' function from 'posts/utils.py'
from flaskapp.posts.utils import get_all_posts
# A function to be called using 'scheduler' from 'flaskapp/__init__.py' on launch
def list_expired_posts():
posts = get_all_posts()
for p in posts:
if p.expired == 1:
print(p.title)
scheduler = BackgroundScheduler()
scheduler.add_job(func=list_expired_posts, trigger="interval", seconds=60)
scheduler.start()
# Terminate Scheduler on APP termination
atexit.register(lambda: scheduler.shutdown())
flaskapp/init.py
# __init__.py Main
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from flaskapp.config import Config
db = SQLAlchemy()
login_manager = LoginManager()
login_manager.login_view = 'users.login'
login_manager.login_message_category = 'warning'
# Create APP Function
def create_app(config_class=Config):
app = Flask(__name__)
# Import Configuration from Config File Class
app.config.from_object(Config)
db.init_app(app)
login_manager.init_app(app)
# Import Blueprint objects
from flaskapp.posts.routes import posts
from flaskapp.users.routes import users
from flaskapp.main.routes import main
# Register Blueprints
app.register_blueprint(posts)
app.register_blueprint(users)
app.register_blueprint(main)
# END
return(app)
# Calling scheduler function 'list_expired_posts' FROM '/main/crons.py' as a scheduled job to be triggered on app initiation
from flaskapp.main.crons import list_expired_posts
list_expired_posts()
Explanation:
I have the function 'get_all_posts()' located in 'posts/utils.py' which works fine when I import it and use it in another blueprint's routes.py file (ex.** users/routes.py**) as shown above.
But I'm getting the below error when importing the same function in another blueprint (specifically.** main/crons.py**) as shown above.
I'm trying to use the 'get_all_posts()' function from 'posts/utils.py' within the 'list_expired_posts()' in 'main/crons.py' and then calling the 'list_expired_posts()' function from 'flaskapp/init.py' to trigger it on launch and keep on executing it every 60 minutes.
Error:
RuntimeError: Working outside of application context.
This typically means that you attempted to use functionality that needed the current application. To solve this, set up an application context with app.app_context(). See the documentation for more information.
*Conclusion Notes Steps attempted: *
I have already eliminated the 'Scheduler' temporarily and tried working only with the function itself and not even calling it from 'flaskapp/init.py'.
I have also tried moving the below code to the 'def create_app(config_class=Config)' section without any luck
from flaskapp.main.crons import list_expired_posts
list_expired_posts()
I have also tried creating a specific Blueprint for 'crons.py' and registering it to my 'flaskapp/init.py' but still got the same result.
As a final outcome, I am trying to call the 'get_all_posts()' FROM 'posts/utils.py', then filter out the 'expired posts' using the 'list_expired_posts()' function FROM 'main/crons.py' and schedule it to print the title of the expired posts every 60 minutes.
Since I've eliminated the scheduler already to test out, I'm quite sure this is not a scheduler issue but some import mixup I'm not figuring out.
I am also aware that the 'list_expired_posts()' can become as another function in 'posts/utils.py' and then directly call the function from there using the scheduler which I've also tried but keep getting the same error.
I also tried manually configuring the app's context as instructed in other posts but I keep getting the same error.
with app.app_context():
I'm not really a Python pro and I always try seeking multiple online resources prior to posting a question here but it seems like i'm out of luck this time. Your help is truly appreciated.
CodePudding user response：
I also don't think this is directly related to the scheduler. I think it works in your users/routes.py because it is inside a route. So it will only get called when you visit that route. And by then your app is fully created and has its context and whatever.
But in main/crons.py it is not inside a route so I think it will try to run it when it passes this import line in your flaskapp/init.py
from flaskapp.main.crons import list_expired_posts
Maybe you could try this: putting the scheduler part into a function
main/crons.py
# Importing 'get_all_posts' function from 'posts/utils.py'
from flaskapp.posts.utils import get_all_posts
# A function to be called using 'scheduler' from 'flaskapp/__init__.py' on launch
def list_expired_posts():
posts = get_all_posts()
for p in posts:
if p.expired == 1:
print(p.title)
def initiate_scheduler():
scheduler = BackgroundScheduler()
scheduler.add_job(func=list_expired_posts, trigger="interval", seconds=60)
scheduler.start()
and then call that function in your run.py after you create the app
run.py
# first create the app
app = create_app()
# then start the scheduler
initiate_scheduler()
if __name__ == '__main__':
app.run()
# if you visit '/posts' in a browser it will call the get_all_posts()
# at this point in time after all the other code has ran.
CodePudding user response：
There is no blueprint in crons.py, which is expected as it's about scheduling, not a routing.
You might want to look into Flask-APScheduler, which you'd initialise in a similar way to Flask-SQLAlchemy and then use in your crons.py, it should carry the context Flask-SQLAchemy needs, transparently.
NB. I don't have time now, but will produce a longer demo tomorrow.
PS. In users/routes.py you define a users Blueprint, but them seem to use a posts blueprint in the route decorator.