Answer a question

Below is my init.py funciton within which i would like to schedule a background task to be run every monday at 10:30 AM.

The code shows no error but doesnt work either.

from flask import Flask, render_template
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from sqlalchemy import event
from sqlalchemy.engine import Engine
from sqlite3 import Connection as SQLite3Connection
from apscheduler.schedulers.background import BackgroundScheduler
from Scripts.File_merge import File_merge_auto

db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()

sched = BackgroundScheduler(daemon=True)
sched.add_job(File_merge_auto.sirmerge(),'cron',day_of_week='mon',hour=10,minute=30)
sched.start()

def create_app():
    """Application-factory pattern"""
    # Initialize our core application
    app = Flask(__name__)
    if app.config['ENV'] == 'development':
        app.config.from_object('config.DevelopmentConfig')
    else:
        app.config.from_object('config.ProductionConfig')

    # Error Handling
    @app.errorhandler(404)
    def page_not_found(error):
        return render_template('error_pages/404.html'), 404

    @app.errorhandler(500)
    def internal_error(error):
        return render_template('error_pages/500.html'), 500

    @app.errorhandler(400)
    def bad_request(error):
        return render_template('error_pages/400.html'), 400

    @app.errorhandler(401)
    def unauthorized(error):
        return render_template('error_pages/401.html'), 401

    @app.errorhandler(403)
    def forbidden(error):
        return render_template('error_pages/403.html'), 403

    @app.errorhandler(405)
    def method_not_allowed(error):
        return render_template('error_pages/405.html'), 405

    @app.errorhandler(503)
    def service_unavailable(error):
        return render_template('error_pages/503.html'), 503

    

    with app.app_context():
        # Import Routes
        from .views.login import login_bp
        from .views.home import home_bp
        from .views.admin import admin_bp
        from .views.flash_dashboard import flash_dashboard_bp
        from .views.blank_dashboard1 import blank_dashboard1_bp
        from .views.blank_dashboard2 import blank_dashboard2_bp
        
        # Register Blueprints
        app.register_blueprint(login_bp)
        app.register_blueprint(home_bp)
        app.register_blueprint(admin_bp)
        app.register_blueprint(flash_dashboard_bp)
        app.register_blueprint(blank_dashboard1_bp)
        app.register_blueprint(blank_dashboard2_bp)
        
        # Initialize extensions
        db.init_app(app)
        migrate.init_app(app, db)
        login_manager.init_app(app)
        
        return app

Could anyone please tell me where i am doing wrong?

I basically need to run a background task in order to merge few files in drive which are then consumed by my other apps on need basis when interacted.

Answers

You are calling File_merge_auto.sirmerge() and passing its return value to sched.add_job(). Your code is thus equivalent to this:

value = File_merge_auto.sirmerge()
sched.add_job(value,'cron',day_of_week='mon',hour=10,minute=30)

This is probably not what you want, so you need to pass the function (and not its return value) to add_job():

sched.add_job(File_merge_auto.sirmerge,'cron',day_of_week='mon',hour=10,minute=30)

If you have any further trouble, consult the troubleshooting section of the documentation.