Reputation: 368
Initial Notes: The project uses Blueprints and below are the file structure and extracts of the code used...
/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
# Function to get all posts from DB
def get_all_posts():
post = Post.query.order_by(Post.id.asc())
return post
# 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())
# 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())
# __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()
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.
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.
Upvotes: 0
Views: 920
Reputation: 5315
There is no blueprint in crons.py
, which is expected as it's about scheduling, not routing.
Not being in a route, you get the error that the application context is missing, because effectively everything not in a route does not have an application context.
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 application context transparently.
PS. In users/routes.py
you define a users
Blueprint, but them seem to use a posts
blueprint in the route decorator.
Ok, sorry for delay but here is a demo of using Flask-SQLAlchemy and Flask-APSscheduler to access the app context from scheduled tasks.
from flask import Blueprint, Flask
from flask_apscheduler import APScheduler
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
scheduler = APScheduler()
users = Blueprint("users", __name__)
class User(db.Model):
id = db.Column(db.Integer(), primary_key=True)
username = db.Column(db.String(100), unique=True, nullable=False)
def __repr__(self):
return "<Name %r>" % self.id
def get_all_users():
return User.query.all()
@users.route("/all", methods=["GET"])
def handle_users_all():
return get_all_users()
@scheduler.task("cron", id="job_get_all_users", second="*/10")
def scheduled_get_all_users():
with scheduler.app.app_context():
get_all_users()
def create_app():
app = Flask(__name__)
app.register_blueprint(users)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:////tmp/74975778.db"
db.init_app(app)
with app.app_context():
db.create_all()
app.config["SCHEDULER_API_ENABLED"] = True
scheduler.init_app(app)
scheduler.start()
return app
NB. this does not do anything useful in the scheduled task, I leave it up to you, but it can execute the utility function get_all_users
on schedule.
Upvotes: 2
Reputation: 71
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
# 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
# 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.
Upvotes: 1