Создал небольшое приложение на 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
