[PLS HELP] Flask-Appbuilder - accessing scheduler from views.py

[palkoc]

[PLS HELP] Flask-Appbuilder - accessing scheduler from views.py

[palkoc]

Hi @christopherpickering
Thanks a lot for your hint. It's helped and now the code is working fine! Here's my version:

extenstions.py:

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 __init___.py:

...

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 views.py:

...
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 :)

[christopherpickering]

I do something similar.. here's how I'm doing it:

1. file called extensions.py where scheduler is initialized.. an also sqlachemy if you are using 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()

2. then in __init__.py you can import it.

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 extensions.py wherever you want to use it. You will probably need to wrap w/ an app_context to make everything work.

Here's a link to my project: https://github.com/atlas-bi/atlas-automation-hub/tree/main/scheduler