0
点赞
收藏
分享

微信扫一扫

python apscheduler cron定时任务触发接口自动化巡检

janedaring 2022-02-22 阅读 111

python cron定时任务触发接口自动化巡检


定时任务触发方式有几种类型,日常的工作中,研发同学运用比较多的就是cron方式

查了一下APScheduler框架内支持多种定时任务方式
首先先安装apscheduler模块

代码如下:(在方法内注释了各种时间参数的定义与范围)

from apscheduler.schedulers.blocking import BlockingScheduler


class Timing:
    def __init__(self, start_date, end_date, hour=None):
        self.start_date = start_date
        self.end_date = end_date
        self.hour = hour

    def cron(self, job, *value_list):
        """cron格式 在特定时间周期性地触发"""
        # year (int 或 str) – 年,4位数字
        # month (int 或 str) – 月 (范围1-12)
        # day (int 或 str) – 日 (范围1-31)
        # week (int 或 str) – 周 (范围1-53)
        # day_of_week (int 或 str) – 周内第几天或者星期几 (范围0-6 或者 mon,tue,wed,thu,fri,sat,sun)
        # hour (int 或 str) – 时 (范围0-23)
        # minute (int 或 str) – 分 (范围0-59)
        # second (int 或 str) – 秒 (范围0-59)
        # start_date (datetime 或 str) – 最早开始日期(包含)
        # end_date (datetime 或 str) – 分 最晚结束时间(包含)
        # timezone (datetime.tzinfo 或str) – 指定时区
        scheduler = BlockingScheduler()
        scheduler.add_job(job, 'cron', start_date=self.start_date, end_date=self.end_date, hour=self.hour,
                          args=[*value_list])
        scheduler.start()

    def interval(self, job, *value_list):
        """interval格式 周期触发任务"""
        # weeks (int) - 间隔几周
        # days (int)  - 间隔几天
        # hours (int) - 间隔几小时
        # minutes (int) - 间隔几分钟
        # seconds (int) - 间隔多少秒
        # start_date (datetime 或 str) - 开始日期
        # end_date (datetime 或 str) - 结束日期
        # timezone (datetime.tzinfo 或str) - 时区
        scheduler = BlockingScheduler()
        # 在 2019-08-29 22:15:00至2019-08-29 22:17:00期间,每隔1分30秒 运行一次 job 方法
        scheduler.add_job(job, 'interval', minutes=1, seconds=30, start_date=self.start_date,
                          end_date=self.end_date, args=[*value_list])
        scheduler.start()

    @staticmethod
    def date(job, *value_list):
        """date格式 特定时间点触发"""
        # run_date (datetime 或 str) - 作业的运行日期或时间
        # timezone (datetime.tzinfo 或 str)  - 指定时区
        scheduler = BlockingScheduler()
        # 在 2019-8-30 01:00:01 运行一次 job 方法
        scheduler.add_job(job, 'date', run_date='2019-8-30 01:00:00', args=[*value_list])
        scheduler.start()

封装的方法不是很通用,后面会优化一下代码,但最起码现在是能用的,哈哈哈哈哈哈
思考了一下思路,巡检触发任务,然后触发钉钉,所以定时任务应该是在最上层
之前分享的钉钉封装的代码内底部继续完善一下

if __name__ == '__main__':
    file_list = ["test_shiyan.py", "MeetSpringFestival.py"]
    # run_py(file_list)
    case_list = ["test_case_01", "test_case_02"]
    # run_case(test_sample, case_list)
    dingDing_list = [2, case_list, test_sample]
    # run_dingDing(*dingDing_list)
    Timing('2022-02-15 00:00:00', '2022-02-16 00:00:00', '0-23').cron(run_dingDing, *dingDing_list)

就是我们上面写的这里能看到

def cron(self, job, *value_list):
        """cron格式 在特定时间周期性地触发"""
		scheduler.add_job(job, 'cron', start_date=self.start_date, end_date=self.end_date, hour=self.hour,
		                          args=[*value_list])

时间范围的填写我放在了Timing()初始化内,看着舒服一点
在运行Timing().cron()后就可以触发定时了,但是必须要开着电脑才可以,等后面开始研究平台,存储在服务器内就美吱吱了~

举报

相关推荐

0 条评论