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()
版权属于:
geteshi
作品采用:
《
署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0)
》许可协议授权
评论 (0)