python 通过 flask-aphotoshopcheduler 实现定期执行任务

Flask-APScheduler 介绍

Flask-APScheduler 是基于 APScheduler 库开发的 Flask 拓展库。APScheduler 的全称是Advanced Python Scheduler 。本文主要通过 APScheduler 中的 cron触发器来实现定时启动任务的目的。

安装 flask_apscheduler 库

pip install flask_apscheduler

导入时报错

1.  
   error: subprocess-exited-with-error
2.  
    
3.  
   × pip subprocess to install build dependencies did not run successfully.
4.  
   │ exit code: 2
5.  
   ╰─> See above for output.
6.  
    
7.  
   note: This error originates from a subprocess, and is likely not a problem with pip.

在网上查阅相关资料，使用如下方式成功安装 flask-apscheduler。但是运行程序时，仍然识别不到flask-apscheduler 库

1.  
   pip install --upgrade setuptools
2.  
   pip install flask-apscheduler

基本结构

1.  
   from flask import Flask
2.  
   from flask_apscheduler import APScheduler
3.  
    
4.  
    
5.  
   class Config():
6.  
   JOBS = [
7.  
   {
8.  
   'id': 'job1',
9.  
   'func': 'test:test', # 第一个test为函数所在py文件名
10.  
    'args': (), # 函数不需要参数时，可以设置为空元组，也可以直接省略
11.  
    'trigger': 'cron', # 使用cron触发器
12.  
    'day': '*', # * 表示每一天
13.  
    'hour': '10',
14.  
    'minute': '0',
15.  
    'second': '0'
16.  
     
17.  
    }
18.  
    ]
19.  
    SCHEDULER_API_ENABLED = True
20.  
     
21.  
     
22.  
    def test():
23.  
    print('success!')
24.  
     
25.  
     
26.  
    if __name__ == '__main__':
27.  
    app = Flask(__name__)
28.  
     
29.  
    app.config.from_object(Config())
30.  
     
31.  
    scheduler = APScheduler()
32.  
    scheduler.init_app(app) # 将调度器对象与Flask应用程序实例(app)相关联
33.  
    scheduler.start()
34.  
     
35.  
    app.run(host='0.0.0.0', port=8000, debug=True)
36.  
     

        其中，SCHEDULER_API_ENABLED = True 用来启用 Flask-APScheduler 库的 API 功能，设置为 True 后，可以使用 API 来查看和管理已经调度的任务。例如，可以通过对 http://127.0.0.1:8000/scheduler/jobs 发送 GET 请求，来获取所有已添加的 Job 信息。（如果不需要 API 功能，可以设置为 False ，以提高安全性和性能）

前端返回结果如下：

1.  
   [
2.  
   {
3.  
   "id": "job1",
4.  
   "name": "job1",
5.  
   "func": "test:test",
6.  
   "args": [],
7.  
   "kwargs": {},
8.  
   "trigger": "cron",
9.  
   "day": "*",
10.  
    "hour": "10",
11.  
    "minute": "0",
12.  
    "second": "0",
13.  
    "misfire_grace_time": 1,
14.  
    "max_instances": 1,
15.  
    "next_run_time": "2023-05-19T10:00:00 08:00"
16.  
    }
17.  
    ]

        app.config.from_object() 用于从一个 python 对象中加载配置变量将其应用于 Flask 应用程序。在这段代码中，通过 app.config['JOBS'] 可以获取到定义的调度任务列表，而通过 app.config['SCHEDULER_API_ENABLED'] 可以获取到 API 是否被启用。其他 Flask 扩展也可以通过这种方式使用配置变量，例如 Flask-SQLAlchemy 扩展需要从 SQLALCHEMY_DATABASE_URI 变量中获取数据库连接字符串。

应用实例

需求：每天定时检查文件夹大小，并进行管理

1.  
   import os
2.  
   import time
3.  
    
4.  
   from flask import Flask
5.  
   from flask_apscheduler import APScheduler
6.  
    
7.  
    
8.  
   def folder_size(path):
9.  
   total = 0
10.  
    for entry in os.scandir(path):
11.  
    if entry.is_file():
12.  
    total = entry.stat().st_size
13.  
    elif entry.is_dir():
14.  
    total = folder_size(entry.path)
15.  
    return total
16.  
     
17.  
     
18.  
    def mng_folder_size():
19.  
    for f in os.listdir('/home/user/video'):
20.  
    if (folder_size('/home/user/video' f)) / 1024 * 1024 * 1024 >= 5:
21.  
    current_time = time.time()
22.  
    for i in os.listdir('/home/user/video/' f):
23.  
    creation_time = os.path.getmtime('/home/user/video/' f '/' i)
24.  
    if (current_time - creation_time) // (24 * 60 * 60) >= 7:
25.  
    os.unlink('/home/user/video/' f '/' i)
26.  
    print(f'{i} removed.')
27.  
     
28.  
     
29.  
    class Config(object):
30.  
    JOBS = [
31.  
    {
32.  
    'id': 'job1',
33.  
    'func': mng_folder_size,
34.  
    'trigger': 'cron',
35.  
    'day': '*',
36.  
    'hour': '0',
37.  
    'minute': '0',
38.  
    'second': '0'
39.  
    },
40.  
    {
41.  
    'id': 'job2',
42.  
    'func': mng_folder_size,
43.  
    'trigger': 'cron',
44.  
    'day': '*',
45.  
    'hour': '12',
46.  
    'minute': '0',
47.  
    'second': '0'
48.  
    }
49.  
    ]
50.  
    SCHEDULER_API_ENABLED = True
51.  
     
52.  
     
53.  
    if __name__ == '__main__':
54.  
    app = Flask(__name__) # 也可以写在上边
55.  
     
56.  
    app.config.from_object(Config())
57.  
     
58.  
    scheduler = APScheduler()
59.  
    scheduler.init_app(app)
60.  
    scheduler.start()
61.  
     
62.  
    app.run(host='0.0.0.0', port=5055, debug=True)

        folder_size() 用于计算文件夹大小。它使用递归来遍历整个目录树，判断是文件还是目录。如果是文件，直接将其大小计入 total ；如果是目录，则递归调用本函数以继续遍历其中所有的文件和子目录，并计算大小。

        os.scandir() 用于遍历指定目录下的文件和子目录。与 os.listdir() 不同，os.scandir() 返回一个迭代器对象，可以逐个访问每个目录项，并且提供了对象的一些属性和方法：

• name : 目录项的名称（str）
• path : 目录项的路径（str）
• is_file() : 判断目录项是否为普通文件
• is_dir() : 判断目录项是否为目录
• is_symlink() : 判断目录项是否为符号链接
• stat() : 获取目录项的状态信息

遇到的问题

1.  
   D:\ProgramData\Anaconda3\lib\site-packages\apscheduler\util.py:428: PytzUsageWarning: The localize method is no longer necessary, as this time zone supports the fold attribute (PEP 495). For more details on migrating to a PEP 495-compliant implementation, see https://pytz-deprecation-shim.readthedocs.io/en/latest/migration.html
2.  
   return tzinfo.localize(dt)

在Config类中加入 SCHEDULER_TIMEZONE = 'Asia/Shanghai'  即可解决时区问题。

这篇好文章是转载于：学新通技术网