Flask动态展示长时时间任务日志|极客教程

Flask动态展示长时时间任务日志

介绍

在现代的Web应用程序中，有时需要执行长时间运行的任务，这些任务可能需要花费几秒钟甚至几分钟的时间。对于用户来说，等待这么长时间是不理想的，并且无法获得即时的反馈。为了改进用户体验，我们可以使用Flask框架来动态展示长时间任务的日志。

本文将着重介绍如何使用Flask来实现动态展示长时间任务日志的功能，并提供相应的示例代码。

准备工作

在开始之前，我们需要准备以下工具和环境：

Python3
Flask框架
虚拟环境（可选）

安装Flask

要使用Flask框架，我们首先需要安装它。可以通过以下命令在命令行中安装Flask：

pip install Flask

如果你使用的是虚拟环境，请确保先激活虚拟环境。

创建Flask应用

在安装完Flask之后，我们可以创建一个简单的Flask应用来测试是否正常安装。创建一个名为app.py的文件，然后在其中输入以下代码：

from flask import Flask

app = Flask(__name__)

@app.route('/')
def hello():
    return 'Hello, Flask!'

if __name__ == '__main__':
    app.run()

保存并运行这个文件，然后在浏览器中打开http://localhost:5000，如果你看到一个显示Hello, Flask!的页面，那么说明Flask已经成功安装并可以正常运行。

动态展示长时间任务日志

现在我们开始解释如何使用Flask来动态展示长时间任务的日志。

创建长时间任务

首先，我们需要创建一个长时间任务，然后将其与Flask应用的路由函数关联起来。在本例中，我们将使用一个简单的示例任务来模拟长时间运行的过程。

在app.py文件中添加以下代码：

import time

def long_running_task():
    for i in range(10):
        print(f'Processing task {i}...')
        time.sleep(1)
    print('Task completed!')

@app.route('/task')
def run_task():
    long_running_task()
    return 'Task completed!'

在上述代码中，我们定义了一个long_running_task函数，它模拟了长时间运行的任务。在这个示例中，任务将打印出数字1到10，并在两个数字之间花费1秒的时间。最后，任务完成时打印一条消息。

我们还创建了一个名为run_task的路由函数，它将调用long_running_task函数并返回一个表示任务完成的消息。

使用Ajax轮询动态获取日志

为了实现动态展示任务日志的功能，我们将使用Ajax轮询来定期获取并更新日志信息。

在app.py文件中添加以下代码：

import subprocess

def long_running_task():
    for i in range(10):
        print(f'Processing task {i}...')
        time.sleep(1)
    print('Task completed!')

def run_subprocess_task():
    process = subprocess.Popen(['python', 'task.py'], stdout=subprocess.PIPE, universal_newlines=True)
    for stdout_line in iter(process.stdout.readline, ""):
        yield stdout_line.rstrip()
    process.stdout.close()
    return_code = process.wait()
    if return_code:
        raise subprocess.CalledProcessError(return_code, cmd)

@app.route('/task')
def run_task():
    #long_running_task()
    return run_subprocess_task()

@app.route('/log')
def get_log():
    def generate_log():
        for line in run_subprocess_task():
            yield line + '<br>'
    return app.response_class(generate_log(), mimetype='text/html')

@app.route('/')
def index():
    return """
    <html>
    <head>
        <script src="https://ajax.googleapis.com/ajax/libs/jquery/3.5.1/jquery.min.js"></script>
        <script>
            function updateLog() {
                 $.ajax({ url: '/log', cache: false, success: function(data) {$ ('#log').html(data);
                        setTimeout(updateLog, 1000);
                    }
                });
            }

            $(function() {
                setTimeout(updateLog, 1000);
            });
        </script>
    </head>
    <body>
        <h1>Long Running Task Log</h1>
        <div id="log"></div>
    </body>
    </html>
    """

在上述代码中，我们创建了一个名为run_subprocess_task的新函数，其中我们使用了subprocess模块来运行一个独立的Python脚本task.py。该脚本实现了长时间运行的任务，并将其输出传递回主应用程序。我们使用yield关键字将输出逐行传回，并在run_task函数中返回。

在app.py文件中，我们还添加了一个新的路由函数/log，用于生成日志输出。该函数是一个Python生成器，用于获取run_subprocess_task函数的输出，并将每行的日志以HTML格式返回。

最后，我们还修改了根路由函数，以便在页面加载时启动Ajax轮询，并在页面上动态更新日志。