Working outside of application context при запуске BackgroundScheduler
При выполнении задачи scheduler появляется ошибка о запуске вне контекста, сам шедулер забирает некоторые данные с конфига. Пробовал уже инициализировать планировщик в самом контексте, создавать задачи в нем же соответственно и запускать. Так же пробовал обернуть саму задачу в with app.app_context(), но это тоже не помогло. Вот куски кода:
# __init__.py
from flask import Flask
from flask_login import LoginManager
from app.models import User
from app.DBManagement import accounts_db
from app.DBManagement.app_db import db
from app.scheduler import build_cards, worker_check
from apscheduler.schedulers.background import BackgroundScheduler
from flask import current_app
app = Flask(__name__)
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'auth.login'
@login_manager.user_loader
def load_user(user_id):
"""Загрузка списка пользователей"""
users = accounts_db.load_accounts()
if user_id in users:
user_data = users[user_id]
return User(user_id, user_data['role'])
return None
def create_app():
app.config.from_object('app.config.Config')
from app.routes.main import main_bp
from app.routes.auth import auth_bp
from app.routes.two_fa import two_fa_bp
from app.api.v1 import api_bp
app.register_blueprint(auth_bp)
app.register_blueprint(main_bp)
app.register_blueprint(two_fa_bp)
# app.register_blueprint(api_bp, url_prefix='/api/v1')
app.register_blueprint(api_bp)
db.init_app(app)
with app.app_context():
db.create_all()
# Инициализация планировщика
scheduler = BackgroundScheduler()
# Добавление задач
scheduler.add_job(build_cards, 'cron', hour=9, minute=0)
scheduler.add_job(worker_check, 'interval', seconds=5)
# Запуск планировщика
scheduler.start()
return app
# scheduler.py
from datetime import datetime
from app.DBManagement import cards_db, servers_db
from app.utils import load_progress, save_progress, save_state
import random
import requests
from flask import current_app
def worker_check(bot_token, chat_id):
"""Проверка состояния серверов (пинги)"""
servers = servers_db.load_servers()
bot_token = current_app.config['BOT_TOKEN']
chat_id = current_app.config['CHAT_ID']
for server in servers:
ip = server['ip']
server_name = server['name']
try:
response = requests.get(f"http://{ip}/ping", timeout=5)
if response.status_code == 200:
print(f"Сервер {server_name} работает нормально.")
else:
requests.get(
f'https://api.telegram.org/{bot_token}/sendmessage?chat_id={chat_id}&text=? Сервер {server_name} вернул статус {response.status_code}')
except requests.exceptions.RequestException:
requests.get(
f'https://api.telegram.org/{bot_token}/sendmessage?chat_id={chat_id}&text=? Ошибка при подключении к серверу {server_name}')
# config.py
import os
import json
class Config:
# Общие настройки
SECRET_KEY = os.environ.get('SECRET_KEY', 'SecretCode')
STATE_FILE = "data/sys/work_state.json"
BOT_TOKEN = os.environ.get('BOT_TOKEN', 'botToken')
CHAT_ID = os.environ.get('CHAT_ID', 'BotChat')
# Настройки базы данных
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL', 'sqlite:///app.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
Получаю эту ошибку:
RuntimeError: Working outside of application context.
This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.
Ответы (1 шт):
Сам нашел ответ на проблему, думаю, не совсем, конечно, правильно, но тем не менее работает, если вдруг это прям грубая ошибка - прошу поправить.
Установил flask_apscheduler и переписал init
:
from flask import Flask
from flask_login import LoginManager
from app.models import User
from app.DBManagement import accounts_db
from app.DBManagement.app_db import db
from app.scheduler import worker_check_with_context, build_cards_with_context
from flask_apscheduler import APScheduler
app = Flask(__name__)
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'auth.login'
@login_manager.user_loader
def load_user(user_id):
"""Загрузка списка пользователей"""
users = accounts_db.load_accounts()
if user_id in users:
user_data = users[user_id]
return User(user_id, user_data['role'])
return None
def create_app():
app.config.from_object('app.config.Config')
from app.routes.main import main_bp
from app.routes.auth import auth_bp
from app.routes.two_fa import two_fa_bp
from app.api.v1 import api_bp
app.register_blueprint(auth_bp)
app.register_blueprint(main_bp)
app.register_blueprint(two_fa_bp)
app.register_blueprint(api_bp)
db.init_app(app)
# Инициализация планировщика
scheduler = APScheduler()
scheduler.init_app(app)
# Добавление задач
scheduler.add_job(
id='build_cards',
func=lambda: build_cards_with_context(app),
trigger='cron',
hour=9,
minute=0
)
scheduler.add_job(
id='worker_check',
func=lambda: worker_check_with_context(app),
trigger='interval',
minutes=10
)
with app.app_context():
db.create_all()
scheduler.start()
return app
в scheduler.py
добавил дополнительную функцию, в которую передается сам контекст:
def worker_check():
"""Проверка состояния серверов (пинги)"""
with current_app.app_context():
servers = servers_db.load_servers()
bot_token = current_app.config['BOT_TOKEN']
chat_id = current_app.config['CHAT_ID']
for server in servers:
ip = server['ip']
server_name = server['name']
try:
response = requests.get(f"http://{ip}/ping", timeout=5)
if response.status_code == 200:
print(f"Сервер {server_name} работает нормально.")
else:
requests.get(
f'https://api.telegram.org/{bot_token}/sendmessage?chat_id={chat_id}&text=? Сервер {server_name} вернул статус {response.status_code}')
except requests.exceptions.RequestException:
requests.get(
f'https://api.telegram.org/{bot_token}/sendmessage?chat_id={chat_id}&text=? Ошибка при подключении к серверу {server_name}')
def worker_check_with_context(app):
with app.app_context():
return worker_check()
Опять же, не думаю, что это прям 100% решение, поэтому вопрос все еще открыт.