Python flask框架定时任务apscheduler应用介绍_python_

class Config(object): // 配置项 JOBS = [ { 'id': 'job1', 'func': 'run:add', 'args': (1, 2), 'trigger': 'interval', 'seconds': 3 } ] SCHEDULER_API_ENABLED = True def add(a, b): print(a+b)

def _load_api(self): """ Add the routes for the scheduler API. """ self._add_url_route('get_scheduler_info', '', api.get_scheduler_info, 'GET') self._add_url_route('add_job', '/jobs', api.add_job, 'POST') self._add_url_route('get_job', '/jobs/', api.get_job, 'GET') self._add_url_route('get_jobs', '/jobs', api.get_jobs, 'GET') self._add_url_route('delete_job', '/jobs/', api.delete_job, 'DELETE') self._add_url_route('update_job', '/jobs/', api.update_job, 'PATCH') self._add_url_route('pause_job', '/jobs//pause', api.pause_job, 'POST') self._add_url_route('resume_job', '/jobs//resume', api.resume_job, 'POST') self._add_url_route('run_job', '/jobs//run', api.run_job, 'POST')

class Config(object): JOBS = [ { 'id': 'job1', 'func': 'scheduler:task', 'args': (1, 2), 'trigger': 'cron', 'day': '*', 'hour': '13', 'minute': '16', 'second': '20' } ] SCHEDULER_API_ENABLED = True def task(a, b): print(str(datetime.datetime.now()) + ' execute task ' + '{}+{}={}'.format(a, b, a + b))

from flask import Flask from flask_apscheduler import APScheduler import datetime class Config(object): SCHEDULER_API_ENABLED = True scheduler = APScheduler() # interval examples @scheduler.task('interval', id='do_job_1', seconds=30, misfire_grace_time=900) def job1(): print(str(datetime.datetime.now()) + ' Job 1 executed')