Flask-APScheduler Documentation

repository·master·Indexed 22 days ago

https://github.com/viniciuschiele/flask-apscheduler

A Flask extension that integrates the APScheduler library to schedule tasks within a Flask application. It features a built-in REST API for job management, support for the Application Factory pattern, and the ability to define jobs via decorators or Flask configuration. It provides tools for managing jobstores, executors, and accessing the Flask application context within background tasks and event listeners.

Tokens
4.5K
Snippets
18
Records
22
Agent score
78%

What's inside Flask-APScheduler

  1. Overview of Flask-APScheduler features

    master

    Flask-APScheduler is a Flask extension that integrates APScheduler into your Flask application. Key capabilities include:

    • Configuration Integration: Loads both scheduler configuration and job definitions directly from your Flask configuration.
    • Hostname Specification: Allows you to specify the hostname on which the scheduler runs.
    • Job Management API: Provides a REST API to manage scheduled jobs.
    • API Security: Includes authentication support for the REST API.
    • Blueprint Support: Integrates with Flask Blueprints, making it compatible with the application factory pattern.
  2. Access the Flask app context within a job

    master

    If your scheduled job needs to access Flask application features (like current_app, g, or database models), you must wrap the logic inside a with scheduler.app.app_context(): block.

    Note for SQLAlchemy users: If you are performing database operations within a job using Flask-SQLAlchemy, you must explicitly call db.session.commit() in addition to providing the Flask app context.

    def blah():
        with scheduler.app.app_context():
            # do stuff
  3. Access Flask app context within scheduler event listeners

    master

    When a function triggered by a scheduler event needs to interact with your Flask application (e.g., accessing database models, configuration, or other app-bound resources), you must explicitly wrap the logic within the Flask application context. You can access the app instance via scheduler.app.app_context().

    def blah():
        with scheduler.app.app_context():
            # do stuff that requires Flask app context
    
    scheduler.add_listener(blah, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
  4. Manage scheduler startup with FLASK_DEBUG

    master

    Flask-APScheduler's startup behavior depends on the FLASK_DEBUG environment variable:

    • If FLASK_DEBUG=true and using Flask's Werkzeug server, the scheduler will start.
    • If FLASK_DEBUG=false and using a production server (e.g., Gunicorn), the scheduler will start.

    To manage server execution patterns during development, you can use get_debug_flag() to differentiate between running the app directly and running via a WSGI server.

    from flask.helpers import get_debug_flag
    if get_debug_flag():
        app.run()
    else
        .... wsgi server run forever
  5. Setup Flask-APScheduler in a Flask application

    master

    To use Flask-APScheduler, you need to import APScheduler, configure it (optionally), initialize it with your Flask app instance using init_app(app), and then call start().

    Configuration can be handled via the Flask app.config object using keys like SCHEDULER_API_ENABLED. Alternatively, you can set options directly on the scheduler instance before calling init_app.

    from flask import Flask
    from flask_apscheduler import APScheduler
    
    # set configuration values
    class Config:
        SCHEDULER_API_ENABLED = True
    
    # create app
    app = Flask(__name__)
    app.config.from_object(Config())
    
    # initialize scheduler
    scheduler = APScheduler()
    # if you don't wanna use a config, you can set options here:
    # scheduler.api_enabled = True
    scheduler.init_app(app)
    scheduler.start()
    
    
    if __name__ == '__main__':
        app.run()
  6. Add jobs using the @scheduler.task decorator

    master

    You can define jobs using the @scheduler.task decorator. These jobs are created when the decorated functions are imported. Ensure these functions are imported before app.run() is called.

    Supported trigger types include:

    • 'interval': Runs at fixed time intervals (e.g., every X seconds).
    • 'cron': Runs based on a cron-like schedule (e.g., specific minutes, days of the week, or weeks).
    # interval example
    @scheduler.task('interval', id='do_job_1', seconds=30, misfire_grace_time=900)
    def job1():
        print('Job 1 executed')
    
    
    # cron examples
    @scheduler.task('cron', id='do_job_2', minute='*')
    def job2():
        print('Job 2 executed')
    
    
    @scheduler.task('cron', id='do_job_3', week='*', day_of_week='sun')
    def job3():
        print('Job 3 executed')
    
    
    scheduler.start()
  7. Avoid loading tasks in __init__.py when using persistent jobstores

    master

    When using a persistent jobstore, tasks registered via decorators or add_job should not be loaded directly in your app/__init__.py.

    If you must load them upon app creation, use the @app.before_first_request hook to import the tasks module. This ensures tasks are registered the first time a request is made to the Flask app, preventing issues with persistent storage initialization.

    # app/__init__.py
    
    scheduler = APScheduler()
    db = SQLAlchemy()
    
    <other stuff>
    
    def create_app(config_class=Config):
        app = Flask(__name__)
        app.config.from_object(config_class)
        db.init_app(app)
        scheduler.init_app(app)
        scheduler.start()
        <other stuff>
    
        @app.before_first_request
        def load_tasks():
            from app import tasks
    
        return app
    
    # app/tasks.py
    
    @scheduler.task('cron', id='do_renewals', hour=9, minute=5)
    def scheduled_function():
        # your scheduled task code here
  8. Use Flask-APScheduler with the Application Factory pattern

    master

    When using the Application Factory pattern in Flask, you should initialize the APScheduler extension outside the factory and then call init_app(app) inside the factory function. This ensures the scheduler is correctly attached to the Flask application instance.

    from flask import Flask
    from flask_apscheduler import APScheduler
    
    scheduler = APScheduler()
    
    def create_app():
        app = Flask(__name__)
        app.config.from_object('config.Config')
        
        scheduler.init_app(app)
        # ... other initializations
        
        return app
  9. Configure jobs via Flask configuration

    master

    Instead of hardcoding job schedules in Python, you can define them in your Flask configuration object. This allows for more flexible deployment and environment-specific scheduling.

    # In your Flask config
    APSCHEDULER_JOBS = [
        {
            'id': 'job1',
            'func': 'myapp.tasks.my_task',
            'trigger': 'interval',
            'second': 10
        }
    ]