APScheduler - 作业未执行

我是新来APScheduler并在更大范围内实现它之前，测试它。APScheduler - 作业未执行

我已经创建了下面的代码，但不知何故，当我使用add_job与trigger='date'我的函数永远不会被调用。如果我使用trigger='interval'，然后一切工作正常。

我也试图与run_date参数去打球，没有运气。

什么可能是错的任何想法？

Apscheduler是版本3.0.3

提前感谢:)

def my_listener(event): 
    if event.exception: 
     print('The job crashed :(') 
    else: 
     print('The job worked :)') 

def test(): 
    print("{} ok ".format(datetime.now())) 


def myFunc(content, img): 
    print("{} - content={}|image{}".format(datetime.now(), content, img)) 
    myfile = open("scheduler.log", "a") 
    myfile.write("{} - content={}|image{}".format(datetime.now(), content, img)) 
    myfile.close() 

def main(): 
    jobstores = \ 
     { 
      'default': SQLAlchemyJobStore(url="postgresql+psycopg2://{}:{}@{}:{}/{}".format(db_user, db_password, db_host, db_port, db_database)) 
     } 

    executors = \ 
     { 
      'default': ThreadPoolExecutor(20), 
      'processpool': ProcessPoolExecutor(5) 
     } 

    job_defaults = \ 
     { 
      'coalesce': False, 
      'max_instances': 3 
     } 

    scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc) 
    scheduler.start() 
    CURR_DATE = datetime.strptime(datetime.strftime(datetime.now(), '%Y%m%d%H%M'), '%Y%m%d%H%M') 
    JOB_DATE = CURR_DATE + timedelta(minutes=1) 
    uid=uuid.uuid4() 
    newjob = scheduler.add_job(myFunc, 
           trigger='date', 
           args=['content data', 'image data'], 
           kwargs=None, 
           id=str(uid), 
           name='test' + str(uid), 
           misfire_grace_time=5, 
           coalesce=False, 
           max_instances=1, 
           next_run_time= JOB_DATE, 
           jobstore='default', 
           executor='default', 
           replace_existing=True) 

    print("Added - {}".format(newjob)) 
    scheduler.add_listener(my_listener, events.EVENT_JOB_EXECUTED | events.EVENT_JOB_ERROR) 
    scheduler.print_jobs() 
    while True: 
     sys.stdout.write('{}\n'.format(datetime.now())); sys.stdout.flush() 
     sleep(1) 

if __name__ == "__main__": 
    main()

来源

2015-09-25 koko

问题是，你正在使用 next_run_time = JOB_DATE，代替run_date = JOB_DATE，

另一选项声明你触发并把它当作参数传递给这样的添加工作：

trigger = DateTrigger(run_date=start_date) 
newjob = scheduler.add_job(myFunc, 
          trigger=trigger, 
          args=['content data', 'image data'], 
          kwargs=None, 
          id=str(uid), 
          name='test' + str(uid), 
          misfire_grace_time=5, 
          coalesce=False, 
          max_instances=1, 
          jobstore='default', 
          executor='default', 
          replace_existing=True)

与您的代码的另一个问题是，当你做你的调度timeaware您使用的是不知道的日期时间

CURR_DATE = datetime.strptime(datetime.strftime(datetime.now(), '%Y%m%d%H%M'), '%Y%m%d%H%M') 
JOB_DATE = CURR_DATE + timedelta(minutes=1) #this is unaware

尝试宣告你的约会对象是这样的：

import pytz 
import datetime 

job_date = datetime.datetime.now(pytz.UTC) + datetime.timedelta(minutes=1)

来源

2015-10-29 04:10:52

APScheduler - 作业未执行

回答

相关问题