APScheduler运行时错误:找不到应用程序

Question

我有个问题。我有IP地址，使用APScheduler，我尝试每10秒ping一次，并更新我的数据库。对于APScheduler，我知道我需要使用with app.app_context()，但问题是我不知道把它放在哪里，无论我在哪里尝试放它，它都会引发：RuntimeError: No application found，并且无法更新数据库

init.py:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from os import path
db = SQLAlchemy()
DB_NAME = "database.db"
def create_app():
app = Flask(__name__)
app.config['SECRET_KEY'] = 'hjshjhdjah kjshkjdhjs'
app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{DB_NAME}'
db.init_app(app)

from .views import views
from .auth import auth
app.register_blueprint(views, url_prefix='/')
app.register_blueprint(auth, url_prefix='/')
from .models import Servers
create_database(app)
return app
def create_database(app):
if not path.exists('websiteflaskupdatedd/' + DB_NAME):
db.create_all(app=app)
print('Created Database!')

auth.py:

from flask import Blueprint, render_template, request, flash, redirect, url_for, escape, session
from .models import Servers, ping
from . import db, create_app
from apscheduler.schedulers.background import BackgroundScheduler
@auth.route('/servers', methods = ['POST', 'GET'])
def servers():
if request.method == 'POST':
server_ip = request.form.get("server_ip")
new_server = Servers(server_ip = server_ip)
try:
db.session.add(new_server)
db.session.commit()
return redirect("/servers")
except:
return "There was an error adding your server"
else:
if not Servers.query.get(1) == None:
servers = Servers.query.order_by(Servers.date_created)
return render_template('servers.html', servers=servers)

def update_up_status():
with create_app().app_context():
server_status_column = Servers.query.filter(Servers.server_status.in_(["0","1"]))
for server in server_status_column:
server.server_status = ping(server.server_ip)

db.session.commit()

scheduler = BackgroundScheduler()
if not scheduler.running:
scheduler.add_job(func=update_up_status, trigger="interval", seconds=10)
scheduler.start()

models.py:

from . import db
from datetime import datetime
import platform    
import subprocess
def ping(host):
cmd = ['ping', '-w', "1", "-n", '1', host]

return subprocess.call(cmd) == 0
class Servers(db.Model):
id = db.Column(db.Integer, primary_key=True)
server_ip = db.Column(db.String(12), nullable=False, unique=True)
server_status = db.Column(db.String(12), nullable=False)
date_created = db.Column(db.DateTime, default= datetime.now)

Answer 1

看起来您的应用程序没有运行。

我经常使用这个设置(flask+apscheduler(，并有一个main.py文件，其中包含以下内容(包括从其他文件导入的必要模块(：

if __name__ == "__main__":
scheduler = BackgroundScheduler()
scheduler.add_job(func=update_up_status, trigger="interval", seconds=10)
scheduler.start()
app.run(host="0.0.0.0", debug=True)

后台调度器在另一个应用程序的后台工作，不会被主线程阻塞，在这种情况下，主线程是烧瓶应用程序。与阻塞调度程序相反，阻塞调度程序将阻塞主线程，并且不能在另一个应用程序内部运行。

Answer 2

就是这样。

def insertArticles(articles):
from app import db
from app.models.articles import Article
from app import scheduler
# 解决 flask apscheduler RuntimeError: No application found
with scheduler.app.app_context():
for article in articles:
# 字典 解包，直接插入字典
new_article = Article(**article)
db.session.add(new_article)
db.session.flush()
db.session.commit()

with scheduler.app.app_context()是重要的线路。