| 12345678910111213141516171819202122232425262728293031323334353637383940414243 |
- import logging
- import sys
- import traceback
- from datetime import datetime, timedelta
- from traceback import format_tb
- from pytz import utc
- from apscheduler.events import (
- JobExecutionEvent, EVENT_JOB_MISSED, EVENT_JOB_ERROR, EVENT_JOB_EXECUTED)
- async def run_coroutine_job(job, jobstore_alias, run_times, logger_name):
- """Coroutine version of run_job()."""
- events = []
- logger = logging.getLogger(logger_name)
- for run_time in run_times:
- # See if the job missed its run time window, and handle possible misfires accordingly
- if job.misfire_grace_time is not None:
- difference = datetime.now(utc) - run_time
- grace_time = timedelta(seconds=job.misfire_grace_time)
- if difference > grace_time:
- events.append(JobExecutionEvent(EVENT_JOB_MISSED, job.id, jobstore_alias,
- run_time))
- logger.warning('Run time of job "%s" was missed by %s', job, difference)
- continue
- logger.info('Running job "%s" (scheduled at %s)', job, run_time)
- try:
- retval = await job.func(*job.args, **job.kwargs)
- except BaseException:
- exc, tb = sys.exc_info()[1:]
- formatted_tb = ''.join(format_tb(tb))
- events.append(JobExecutionEvent(EVENT_JOB_ERROR, job.id, jobstore_alias, run_time,
- exception=exc, traceback=formatted_tb))
- logger.exception('Job "%s" raised an exception', job)
- traceback.clear_frames(tb)
- else:
- events.append(JobExecutionEvent(EVENT_JOB_EXECUTED, job.id, jobstore_alias, run_time,
- retval=retval))
- logger.info('Job "%s" executed successfully', job)
- return events
|
