使用 Flask 和 Python 定时刷新 CSV 文件-Python教程-PHP中文网

使用 Flask 和 Python 定时刷新 CSV 文件

霞舞

发布： 2025-08-24 23:42:30

原创

853人浏览过

使用 flask 和 python 定时刷新 csv 文件

本文旨在指导开发者如何使用 Flask 框架和 Python 实现定时刷新 CSV 文件的功能，以便在 Web 应用中展示最新的数据。我们将探讨使用 Celery、APScheduler 或简单的 cron 任务来自动化数据抓取和更新 CSV 文件的过程，并讨论如何处理文件锁定的问题，确保数据的一致性。

在构建 Web 应用时，经常需要从外部数据源获取信息，并将其展示在网页上。如果数据源是 CSV 文件，并且需要定期更新，就需要一种机制来自动刷新 CSV 文件。本文将介绍几种使用 Flask 和 Python 实现此功能的方法。

1. 使用 Celery

Celery 是一个分布式任务队列，可以用于异步执行任务。它非常适合用于定时刷新 CSV 文件，因为它可以在后台运行，而不会阻塞 Web 应用的主线程。

立即学习“Python免费学习笔记（深入）”；

安装 Celery:
```
pip install celery redis
```
登录后复制
这里使用了 Redis 作为 Celery 的消息代理。你也可以使用其他的消息代理，如 RabbitMQ。

创建 Celery 任务:

from celery import Celery
import pandas as pd

celery_app = Celery('my_app', broker='redis://localhost:6379/0')

@celery_app.task
def update_csv():
    """
    This task scrapes data and updates the CSV file.
    Replace this with your actual scraping and CSV writing logic.
    """
    # Your scraping logic here
    data = {'col1': [1, 2], 'col2': [3, 4]}  # Example data
    df = pd.DataFrame(data)
    df.to_csv('data.csv', index=False)
    print("CSV file updated!")

@celery_app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    sender.add_periodic_task(600.0, update_csv.s(), name='update csv every 10 minutes')

登录后复制

这段代码定义了一个 Celery 任务 update_csv，它负责抓取数据并更新 CSV 文件。setup_periodic_tasks 函数使用 add_periodic_task 方法来定期执行 update_csv 任务，这里设置为每 600 秒（10 分钟）执行一次。

启动 Celery Worker:
```
celery -A your_module_name worker -l info
```
登录后复制
将 your_module_name 替换为包含 Celery 应用的 Python 模块的名称。

2. 使用 APScheduler

APScheduler 是一个 Python 库，用于在后台调度任务。它比 Celery 更轻量级，更易于设置，但不如 Celery 强大，不适合大规模的分布式任务。

安装 APScheduler:
```
pip install APScheduler
```
登录后复制

创建调度器并添加任务:

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler
import pandas as pd

app = Flask(__name__)

def update_csv():
    """
    This task scrapes data and updates the CSV file.
    Replace this with your actual scraping and CSV writing logic.
    """
    # Your scraping logic here
    data = {'col1': [5, 6], 'col2': [7, 8]}  # Example data
    df = pd.DataFrame(data)
    df.to_csv('data.csv', index=False)
    print("CSV file updated!")

scheduler = BackgroundScheduler()
scheduler.add_job(func=update_csv, trigger="interval", minutes=10)
scheduler.start()

@app.route('/')
def index():
    return "App is running!"

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

登录后复制

这段代码创建了一个 BackgroundScheduler 实例，并使用 add_job 方法添加了一个任务 update_csv，该任务将每 10 分钟执行一次。

华文笔杆

写材料用华文，华文笔杆帮你搞定公文写作

491

查看详情

3. 使用 Cron 任务

Cron 是一个 Linux 系统下的定时任务工具。你可以使用它来定期执行 Python 脚本，该脚本负责刷新 CSV 文件。

创建 Python 脚本:

import pandas as pd

def update_csv():
    """
    This task scrapes data and updates the CSV file.
    Replace this with your actual scraping and CSV writing logic.
    """
    # Your scraping logic here
    data = {'col1': [9, 10], 'col2': [11, 12]}  # Example data
    df = pd.DataFrame(data)
    df.to_csv('data.csv', index=False)
    print("CSV file updated!")

if __name__ == '__main__':
    update_csv()

登录后复制

将此脚本保存为 update_csv.py。

编辑 Cron 表达式:

在终端中输入 crontab -e，打开 Cron 表达式编辑器。
添加 Cron 任务:

在 Cron 表达式编辑器中添加以下行：
```
*/10 * * * * python /path/to/update_csv.py
```
登录后复制
将 /path/to/update_csv.py 替换为 update_csv.py 脚本的实际路径。这个 Cron 表达式表示每 10 分钟执行一次 update_csv.py 脚本。

处理文件锁定

当多个进程同时尝试写入同一个 CSV 文件时，可能会发生文件锁定问题，导致数据损坏。为了避免这种情况，可以使用以下方法：

文件锁定库: 使用 fcntl 或 filelock 等库来显式地锁定文件，确保只有一个进程可以写入文件。
使用数据库: 将数据存储在数据库中，而不是 CSV 文件中。数据库系统通常提供内置的并发控制机制，可以避免文件锁定问题。由于你在问题中提到了 SQLAlchemy，这是一个推荐的方案。
临时文件: 先将数据写入到一个临时文件，然后使用原子操作（例如 os.rename）将临时文件重命名为最终的 CSV 文件。

注意事项和总结

选择哪种方法取决于你的具体需求。Celery 适合于需要分布式任务队列的场景，APScheduler 适合于简单的定时任务，Cron 任务适合于在 Linux 系统下运行的脚本。
务必处理文件锁定问题，以确保数据的一致性。
在生产环境中，建议使用更健壮的错误处理机制，例如日志记录和异常处理。
对于频繁更新的数据，使用数据库可能比使用 CSV 文件更有效率。

通过本文的介绍，你应该能够使用 Flask 和 Python 实现定时刷新 CSV 文件的功能，并将其应用于你的 Web 应用中。

以上就是使用 Flask 和 Python 定时刷新 CSV 文件的详细内容，更多请关注php中文网其它相关文章！