如何在 Python 中使用 Flask 执行周期性任务 [英] How to perform periodic task with Flask in Python

问题描述

我一直在使用 Flask 为我的 k8055 USB接口板；相当标准的吸气剂和推杆，Flask 确实让我的生活变得更轻松.

I've been using Flask to provide a simple web API for my k8055 USB interface board; fairly standard getters and putters, and Flask really made my life a lot easier.

但我希望能够在乳清发生时将状态变化记录为/接近.

But I want to be able to register changes of state as / near when whey happen.

例如，如果我有一个按钮连接到板上，我可以轮询该特定端口的 api.但是如果我想让输出直接反映输出，无论是否有人在与 api 对话，我都会有这样的事情.

For instance, if I have a button connected to the board, I can poll the api for that particular port. But if I wanted to have the outputs directly reflect the outputs, whether or not someone was talking to the api, I would have something like this.

while True:
    board.read()
    board.digital_outputs = board.digital_inputs
    board.read()
    time.sleep(1)

并且每一秒，输出都会更新以匹配输入.

And every second, the outputs would be updated to match the inputs.

在Flask下有没有办法做这种事情?我之前在 Twisted 中做过类似的事情，但是 Flask 对这个特定的应用程序来说太方便了，暂时还不能放弃......

Is there any way to do this kind of thing under Flask? I've done similar things in Twisted before but Flask is too handy for this particular application to give up on it just yet...

谢谢.

推荐答案

您可以将 cron 用于简单的任务.

You could use cron for simple tasks.

为您的任务创建一个烧瓶视图.

Create a flask view for your task.

# a separate view for periodic task
@app.route('/task')
def task():
    board.read()
    board.digital_outputs = board.digital_inputs

然后使用 cron，定期从该 url 下载

Then using cron, download from that url periodically

# cron task to run each minute
0-59 * * * * run_task.sh

run_task.sh 内容在哪里

Where run_task.sh contents are

wget http://localhost/task

Cron 的运行频率不能超过一分钟一次.如果您需要更高的频率，(例如，每 5 秒 = 每分钟 12 次)，则必须在 tun_task.sh 中按以下方式进行

Cron is unable to run more frequently than once a minute. If you need higher frequency, (say, each 5 seconds = 12 times per minute), you must do it in tun_task.sh in the following way

# loop 12 times with a delay
for i in 1 2 3 4 5 6 7 8 9 10 11 12
do
    # download url in background for not to affect delay interval much
    wget -b http://localhost/task
    sleep 5s
done

这篇关于如何在 Python 中使用 Flask 执行周期性任务的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！