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% решение, поэтому вопрос все еще открыт.

→ Ссылка