Flask 中怎样用 while 写一个定时执行的任务？（定时查询最新数据）

[miniyao]

用 before_app_first_request 的钩子，希望 app 首次启动时，开始运行一个每隔 1 小时定时查询最新数据的任务。结果发现这个任务每隔 1 小时去查询最新的数据，通过 flask_sqlalchemy 初始化之后的工厂模式，查询到的数据都是一样的。如下：

def async_cron_task(app):

with app.app_context():

while True:

new_user = User.query.order_by(User.reg_time.desc()).first()

print('New User: ', new_user.id)

time.sleep(3600)



def cron_task():

app = current_app._get_current_object()

thr = Thread(target=async_cron_task, args=[app])

thr.start()

return thr

但是，如果把数据库写成硬连接，就可以查到最新数据：

db = MySQLdb.connect(host='localhost', port=3306, user='xxx', passwd='xxx', db='xxx', charset='utf8')

cursor = db.cursor()

由于原生 SQL 查询语句太复杂了，希望用 SQLAlchemy 的方式连接数据库，要怎么样才能定时查到新数据呢？

[fhy1994]

apscheduler

[fansfans]

我前几天刚好也遇到了类似的问题在接口中提交数据但是在线程池中的任务没有获取到对应的数据 应该是 sqlalchemy 的缓存问题 后面直接用 celery 了 调式控制都方便很多。

[frostming]

https://github.com/frostming/flask-crontab

用 crontab 运行周期任务，自动激活 app_context，减少代码量

[wobuhuicode]

定时任务用 crontab ～

[RRRoger]

可以试试改成自己调自己的接口

[amoyiki]

目前项目使用很多年 celery, 出现过屈指可数的程序问题。一般就是 redis 获取连接池失败。
不使用 celery 的话可以自己创建一个接口，外部 1h 调用一次。外部直接用 crontab 写定时

[l4ever]

用 flask-apscheduler

[www5070504]

丢到 apscheduler 里

[wzwwzw]

apscheduler flask 有插件

[Vegetable]

对自己好一点，apscheduler 吧

[676529483]

1. 你这个就是数据库连接处理有问题额，问题在 with app.app_context():这行，你应该想要使用 sqlalchemy 的连接初始化，要手动调用获取数据库连接
2. 我以前也这样自己处理，发现各种调度时间问题。现在更倾向于另启动一个 cron 服务，定时去调用 api 接口

[nonduality]

针对你的需求，要获取最新数据，用相关事件信号触发动作不是更直截了当还稳定可靠？

[cz5424]

看了一下的评论，猜测是事务的问题，同一个上下文里面会一直处于同一个事务，可重复读会一直读到相同数据，另一个方法是可以修改 sqlalchemy 的事务隔离级别来避免，不需要调换上下文和 while （代价是其他读写提交也会生效，慎重修改）

[HUNYXV]

```
pip install apscheduler
```

[jianhua]

搜索：flask + celery

[Aether]

@knightdf 一直在用 celery，没出过问题（好几年了）。你多研究下？

[knightdf]

@Aether 你回错人了？

[myCupOfTea]

apscheduler 有的时候会忽略任务,比如当前系统比较繁忙正好到了定时任务的时候

[fansfans]

@cz5424 我目前复现情况也听多的 技术在数据库中查询有数据 使用 sqlalchemy 依旧无法获取最新数据 。

[yagamil]

看着像脏读的问题. 每次用 sql 连接语句重连就不出现这个问题