如何解决在 Flask 应用程序中运行 APScheduler 作为后台函数而不干扰其他线程
下面是我的 init.py 函数,我想在其中安排一个后台任务在每周一上午 10:30 运行。
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
谁能告诉我我哪里做错了?
我基本上需要运行一个后台任务,以便合并驱动器中的几个文件,然后在交互时根据需要由我的其他应用程序使用。
解决方法
您正在调用 File_merge_auto.sirmerge()
并将其返回值传递给 sched.add_job()
。因此,您的代码等效于:
value = File_merge_auto.sirmerge()
sched.add_job(value,'cron',day_of_week='mon',hour=10,minute=30)
这可能不是您想要的,因此您需要将函数(而不是其返回值)传递给 add_job()
:
sched.add_job(File_merge_auto.sirmerge,minute=30)
如果您有任何其他问题,请参阅文档的 troubleshooting section。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。