Как силами flask организовать проверку обновлений в БД по таймеру?

Создал небольшое приложение на flask с БД куда сенсоры на arduino периодически отправляют данные.

Как сделать так, чтобы, например, каждые 20 минут проверять БД на наличие новых записей и если их нет, то отправлять alarm.
Вопросов как организовать alarm или проверку у меня нет (проверяю по времени и если timedelta больше скольких-то минут, отправляю по api в телеграм уведомление).

Подскажите, как организовать таймер в 20 минут?
Cron нет - хотелось бы именно силами этого же приложения flask.

Во views.py создал такую функцию:
def check_alive_Sensors():
    while True:
        TELEGRAM_URL = "https://api.telegram.org/bot"
        part_url_for_1 = "/sendMessage?chat_id="
        chat_id = BaseConfig.CHAT_ID
        part_url_for_2 = "&text="
        text = "Внимание: ОТКЛЮЧЕНИЕ ЭЛЕКТРИЧЕСТВА. ДАТА: "
        BOT_TOKEN = BaseConfig.BOT_TOKEN

        now = datetime.now() + timedelta(hours=3)

        obj1, obj2 = Sensors.query.order_by(Sensors.id.desc()).limit(2)
        print(obj1.date_send)

        delta = obj1.date_send - now
        last_send = obj1.date_send
        delta1 = int(delta.total_seconds() / 60)

        if delta1 > 2:
            text = "Внимание: Долго не было новых данных. Последняя отправка: "
            request_telegram = TELEGRAM_URL + BOT_TOKEN + part_url_for_1 + chat_id + part_url_for_2 + text \
                               + str(last_send) + "мин."
            # return redirect(request_telegram)
            resp = requests.get(request_telegram)
            api_answer = resp.json()
            api_answer_status = api_answer["ok"]
            print(type(api_answer))
            print(type(api_answer_status))

        time.sleep(20)

thread = Thread(target=check_alive_Sensors)
thread.start()


Вылезает следующий Exception:

C:\Users\xabor\PycharmProjects\flask\TODO-app\venv\Scripts\python.exe C:\Users\xabor\PycharmProjects\flask\TODO-app\runner.py 
Exception in thread Thread-1 (check_alive_Sensors):
Traceback (most recent call last):
  File "C:\Program Files\Python311\Lib\threading.py", line 1045, in _bootstrap_inner
    self.run()
  File "C:\Program Files\Python311\Lib\threading.py", line 982, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\xabor\PycharmProjects\flask\TODO-app\app\views.py", line 33, in check_alive_Sensors
    obj1, obj2 = Sensors.query.order_by(Sensors.id.desc()).limit(2)
                 ^^^^^^^^^^^^^
  File "C:\Users\xabor\PycharmProjects\flask\TODO-app\venv\Lib\site-packages\flask_sqlalchemy\model.py", line 31, in __get__
    cls, session=cls.__fsa__.session()  # type: ignore[arg-type]
                 ^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\xabor\PycharmProjects\flask\TODO-app\venv\Lib\site-packages\sqlalchemy\orm\scoping.py", line 47, in __call__
    sess = self.registry()
           ^^^^^^^^^^^^^^^
  File "C:\Users\xabor\PycharmProjects\flask\TODO-app\venv\Lib\site-packages\sqlalchemy\util\_collections.py", line 1006, in __call__
    key = self.scopefunc()
          ^^^^^^^^^^^^^^^^
  File "C:\Users\xabor\PycharmProjects\flask\TODO-app\venv\Lib\site-packages\flask_sqlalchemy\session.py", line 102, in _app_ctx_id
    return id(app_ctx._get_current_object())  # type: ignore[attr-defined]
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\xabor\PycharmProjects\flask\TODO-app\venv\Lib\site-packages\werkzeug\local.py", line 508, in _get_current_object
    raise RuntimeError(unbound_message) from None
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.
 * Serving Flask app 'app'
 * Debug mode: off
INFO:werkzeug:WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on http://127.0.0.1:5000
INFO:werkzeug:Press CTRL+C to quit
  • Вопрос задан
  • 61 просмотр
Пригласить эксперта
Ответы на вопрос 1
@Everything_is_bad
Cron нет - хотелось бы именно силами этого же приложения flask.
Cron - да, flask это веб сервер, не нужно на него вешать, то что он не должен делать
Ответ написан
Ваш ответ на вопрос

Войдите, чтобы написать ответ

Войти через центр авторизации
Похожие вопросы