-
Notifications
You must be signed in to change notification settings - Fork 184
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[PLS HELP] Flask-Appbuilder - accessing scheduler from views.py #216
Comments
I do something similar.. here's how I'm doing it:
from flask_apscheduler import APScheduler
from flask_sqlalchemy import SQLAlchemy # optional for sqlalchemy
from flask_sqlalchemy_caching import CachingQuery # for sqlalchemy
db = SQLAlchemy(query_class=CachingQuery) # for sqlalchemy
atlas_scheduler = APScheduler()
from scheduler.extensions import atlas_scheduler, db
...
# set it up inside you create_app()
db.init_app(app)
with app.app_context():
# pylint: disable=W0611
if atlas_scheduler.running is False:
# pytest imports twice. this will save us on the
# second import.
atlas_scheduler.start()
# here you should also add event listeners if you want them. you can now import the scheduler from the Here's a link to my project: https://github.com/atlas-bi/atlas-automation-hub/tree/main/scheduler |
Hi @christopherpickering
from flask_apscheduler import APScheduler
from flask_apscheduler.auth import HTTPBasicAuth
import logging
from .. import api
def collect_master(master_hostname=None):
nbkt_check_api = api.NetbackupCheckApi()
logging.debug(f'Master {master_hostname} collection in progress')
nbkt_check_api.get_master_events(master_name=master_hostname, internal=True)
logging.debug('------------------------------------------------------')
scheduler = APScheduler()
scheduler.api_enabled = True
scheduler.auth = HTTPBasicAuth()
@scheduler.authenticate
def authenticate(auth):
"""Check auth."""
return auth["username"] == "guest" and auth["password"] == "guest" Initializing from ...
app = Flask(__name__)
app.config.from_object("config")
app.config['APP_NAME'] = app.config['APP_NAME']+" "+'.'.join(map(str,app.config['VERSION']))
db = SQLA(app)
migrate = Migrate(app, db)
appbuilder = AppBuilder(app, db.session, indexview=MyIndexView, base_template="mybase.html", security_manager_class=MySecurityManager, menu=Menu(extra_classes="navbar-fixed-top "))
from . import models, views, api# noqa
from .scheduler.extensions import scheduler, collect_master
scheduler.start()
# Add tasks to collect data from active masters
active_masters = db.session.query(models.MasterServer).filter_by(active=true()).all()
if active_masters is not None:
logging.debug(f"Active masters found: {active_masters}")
if app.config['SCHEDULER_DEFAULT_COLLECTION_INTERVAL'] >0:
for master in active_masters:
logging.debug(f"Starting scheduled job for: {master.hostname}, default collection interval: {app.config['SCHEDULER_DEFAULT_COLLECTION_INTERVAL']}")
with app.app_context():
job = scheduler.add_job(
trigger = 'interval',
id=f"{master.hostname}",
name = f"{master.hostname}",
seconds=app.config['SCHEDULER_DEFAULT_COLLECTION_INTERVAL'],
misfire_grace_time=900,
replace_existing=True,
func = collect_master,
args = [master.hostname],
) And finally ...
from .scheduler.extensions import scheduler, collect_master
...
def post_update(self, item):
super()
logging.debug(f"Currently scheduled jobs: [{scheduler.get_jobs()}]")
logging.debug(f"Is current item scheduled? [{scheduler.get_job(item.hostname)}]")
if not item.active and scheduler.get_job(item.hostname):
logging.debug(f"Master {item} is not active anymore, removing if from the scheduler")
scheduler.remove_job(item.hostname)
logging.debug(f"Currently scheduled jobs: [{scheduler.get_jobs()}]")
if item.active and scheduler.get_job(item.hostname) is None:
logging.debug(f"Master {item} is active now, adding it to the scheduler")
scheduler.add_job(
trigger = 'interval',
id=f"{item.hostname}",
name = f"{item.hostname}",
seconds=appbuilder.app.config['SCHEDULER_DEFAULT_COLLECTION_INTERVAL'],
misfire_grace_time=900,
replace_existing=True,
func = collect_master,
args = [item.hostname],
)
logging.debug(f"Currently scheduled jobs: [{scheduler.get_jobs()}]") Thanks again and Happy New year :) |
Hi all,
I would like to ask you for a help. I have a Flask application (based on flask-appbuilder). The scheduler is initialized and started in
__init__.py
:I have
views.py
where the data views are defined. There is apost_update
method invoked when a model data are modified. I would like to add/remove/modify a scheduler task based on model changes - usingscheduler.add_job
,scheduler.remove_job
Here's a snippet from
views.py
My question is: How to access the scheduler created in
__init__.py
from withinviews.py
?Thanks a lot.
Pavol
The text was updated successfully, but these errors were encountered: