Skip to content
Branch: master
Find file Copy path
615 lines (513 sloc) 19.7 KB
Python job scheduling for humans.
An in-process scheduler for periodic jobs that uses the builder pattern
for configuration. Schedule lets you run Python functions (or any other
callable) periodically at pre-determined intervals using a simple,
human-friendly syntax.
Inspired by Addam Wiggins' article "Rethinking Cron" [1] and the
"clockwork" Ruby module [2][3].
- A simple to use API for scheduling jobs.
- Very lightweight and no external dependencies.
- Excellent test coverage.
- Tested on Python 2.7, 3.5 and 3.6
>>> import schedule
>>> import time
>>> def job(message='stuff'):
>>> print("I'm working on:", message)
>>> schedule.every(10)
>>> schedule.every(5).to(10)
>>> schedule.every(), message='things')
>>> schedule.every()"10:30").do(job)
>>> while True:
>>> schedule.run_pending()
>>> time.sleep(1)
import collections
import datetime
import functools
import logging
import random
import re
import time
logger = logging.getLogger('schedule')
class ScheduleError(Exception):
"""Base schedule exception"""
class ScheduleValueError(ScheduleError):
"""Base schedule value error"""
class IntervalError(ScheduleValueError):
"""An improper interval was used"""
class CancelJob(object):
Can be returned from a job to unschedule itself.
class Scheduler(object):
Objects instantiated by the :class:`Scheduler <Scheduler>` are
factories to create jobs, keep record of scheduled jobs and
handle their execution.
def __init__(self): = []
def run_pending(self):
Run all jobs that are scheduled to run.
Please note that it is *intended behavior that run_pending()
does not run missed jobs*. For example, if you've registered a job
that should run every minute and you only call run_pending()
in one hour increments then your job won't be run 60 times in
between but only once.
runnable_jobs = (job for job in if job.should_run)
for job in sorted(runnable_jobs):
def run_all(self, delay_seconds=0):
Run all jobs regardless if they are scheduled to run or not.
A delay of `delay` seconds is added between each job. This helps
distribute system load generated by the jobs more evenly
over time.
:param delay_seconds: A delay added between every executed job
"""'Running *all* %i jobs with %is delay inbetween',
len(, delay_seconds)
for job in[:]:
def clear(self, tag=None):
Deletes scheduled jobs marked with the given tag, or all jobs
if tag is omitted.
:param tag: An identifier used to identify a subset of
jobs to delete
if tag is None:
else:[:] = (job for job in if tag not in job.tags)
def cancel_job(self, job):
Delete a scheduled job.
:param job: The job to be unscheduled
except ValueError:
def every(self, interval=1):
Schedule a new periodic job.
:param interval: A quantity of a certain time unit
:return: An unconfigured :class:`Job <Job>`
job = Job(interval, self)
return job
def _run_job(self, job):
ret =
if isinstance(ret, CancelJob) or ret is CancelJob:
def next_run(self):
Datetime when the next job should run.
:return: A :class:`~datetime.datetime` object
if not
return None
return min(
def idle_seconds(self):
:return: Number of seconds until
:meth:`next_run <Scheduler.next_run>`.
return (self.next_run -
class Job(object):
A periodic job as used by :class:`Scheduler`.
:param interval: A quantity of a certain time unit
:param scheduler: The :class:`Scheduler <Scheduler>` instance that
this job will register itself with once it has
been fully configured in :meth:``.
Every job runs at a given fixed time interval that is defined by:
* a :meth:`time unit <Job.second>`
* a quantity of `time units` defined by `interval`
A job is usually created and returned by :meth:`Scheduler.every`
method, which also defines its `interval`.
def __init__(self, interval, scheduler=None):
self.interval = interval # pause interval * unit between runs
self.latest = None # upper limit to the interval
self.job_func = None # the job job_func to run
self.unit = None # time units, e.g. 'minutes', 'hours', ...
self.at_time = None # optional time at which this job runs
self.last_run = None # datetime of the last run
self.next_run = None # datetime of the next run
self.period = None # timedelta between runs, only valid for
self.start_day = None # Specific day of the week to start on
self.tags = set() # unique set of tags for the job
self.scheduler = scheduler # scheduler to register with
def __lt__(self, other):
PeriodicJobs are sortable based on the scheduled time they
run next.
return self.next_run < other.next_run
def __str__(self):
return (
"Job(interval={}, "
"unit={}, "
"do={}, "
"args={}, "
def __repr__(self):
def format_time(t):
return t.strftime('%Y-%m-%d %H:%M:%S') if t else '[never]'
def is_repr(j):
return not isinstance(j, Job)
timestats = '(last run: %s, next run: %s)' % (
format_time(self.last_run), format_time(self.next_run))
if hasattr(self.job_func, '__name__'):
job_func_name = self.job_func.__name__
job_func_name = repr(self.job_func)
args = [repr(x) if is_repr(x) else str(x) for x in self.job_func.args]
kwargs = ['%s=%s' % (k, repr(v))
for k, v in self.job_func.keywords.items()]
call_repr = job_func_name + '(' + ', '.join(args + kwargs) + ')'
if self.at_time is not None:
return 'Every %s %s at %s do %s %s' % (
self.unit[:-1] if self.interval == 1 else self.unit,
self.at_time, call_repr, timestats)
fmt = (
'Every %(interval)s ' +
('to %(latest)s ' if self.latest is not None else '') +
'%(unit)s do %(call_repr)s %(timestats)s'
return fmt % dict(
unit=(self.unit[:-1] if self.interval == 1 else self.unit),
def second(self):
if self.interval != 1:
raise IntervalError('Use seconds instead of second')
return self.seconds
def seconds(self):
self.unit = 'seconds'
return self
def minute(self):
if self.interval != 1:
raise IntervalError('Use minutes instead of minute')
return self.minutes
def minutes(self):
self.unit = 'minutes'
return self
def hour(self):
if self.interval != 1:
raise IntervalError('Use hours instead of hour')
return self.hours
def hours(self):
self.unit = 'hours'
return self
def day(self):
if self.interval != 1:
raise IntervalError('Use days instead of day')
return self.days
def days(self):
self.unit = 'days'
return self
def week(self):
if self.interval != 1:
raise IntervalError('Use weeks instead of week')
return self.weeks
def weeks(self):
self.unit = 'weeks'
return self
def monday(self):
if self.interval != 1:
raise IntervalError('Use mondays instead of monday')
self.start_day = 'monday'
return self.weeks
def tuesday(self):
if self.interval != 1:
raise IntervalError('Use tuesdays instead of tuesday')
self.start_day = 'tuesday'
return self.weeks
def wednesday(self):
if self.interval != 1:
raise IntervalError('Use wednesdays instead of wednesday')
self.start_day = 'wednesday'
return self.weeks
def thursday(self):
if self.interval != 1:
raise IntervalError('Use thursdays instead of thursday')
self.start_day = 'thursday'
return self.weeks
def friday(self):
if self.interval != 1:
raise IntervalError('Use fridays instead of friday')
self.start_day = 'friday'
return self.weeks
def saturday(self):
if self.interval != 1:
raise IntervalError('Use saturdays instead of saturday')
self.start_day = 'saturday'
return self.weeks
def sunday(self):
if self.interval != 1:
raise IntervalError('Use sundays instead of sunday')
self.start_day = 'sunday'
return self.weeks
def tag(self, *tags):
Tags the job with one or more unique indentifiers.
Tags must be hashable. Duplicate tags are discarded.
:param tags: A unique list of ``Hashable`` tags.
:return: The invoked job instance
if not all(isinstance(tag, collections.Hashable) for tag in tags):
raise TypeError('Tags must be hashable')
return self
def at(self, time_str):
Specify a particular time that the job should be run at.
:param time_str: A string in one of the following formats: `HH:MM:SS`,
`HH:MM`,`:MM`, `:SS`. The format must make sense given how often
the job is repeating; for example, a job that repeats every minute
should not be given a string in the form `HH:MM:SS`. The difference
between `:MM` and `:SS` is inferred from the selected time-unit
(e.g. `every()':30')` vs. `every()':30')`).
:return: The invoked job instance
if (self.unit not in ('days', 'hours', 'minutes')
and not self.start_day):
raise ScheduleValueError('Invalid unit')
if not isinstance(time_str, str):
raise TypeError('at() should be passed a string')
if self.unit == 'days' or self.start_day:
if not re.match(r'^([0-2]\d:)?[0-5]\d:[0-5]\d$', time_str):
raise ScheduleValueError('Invalid time format')
if self.unit == 'hours':
if not re.match(r'^([0-5]\d)?:[0-5]\d$', time_str):
raise ScheduleValueError(('Invalid time format for'
' an hourly job'))
if self.unit == 'minutes':
if not re.match(r'^:[0-5]\d$', time_str):
raise ScheduleValueError(('Invalid time format for'
' a minutely job'))
time_values = time_str.split(':')
if len(time_values) == 3:
hour, minute, second = time_values
elif len(time_values) == 2 and self.unit == 'minutes':
hour = 0
minute = 0
_, second = time_values
hour, minute = time_values
second = 0
if self.unit == 'days' or self.start_day:
hour = int(hour)
if not (0 <= hour <= 23):
raise ScheduleValueError('Invalid number of hours')
elif self.unit == 'hours':
hour = 0
elif self.unit == 'minutes':
hour = 0
minute = 0
minute = int(minute)
second = int(second)
self.at_time = datetime.time(hour, minute, second)
return self
def to(self, latest):
Schedule the job to run at an irregular (randomized) interval.
The job's interval will randomly vary from the value given
to `every` to `latest`. The range defined is inclusive on
both ends. For example, `every(A).to(B).seconds` executes
the job function every N seconds such that A <= N <= B.
:param latest: Maximum interval between randomized job runs
:return: The invoked job instance
self.latest = latest
return self
def do(self, job_func, *args, **kwargs):
Specifies the job_func that should be called every time the
job runs.
Any additional arguments are passed on to job_func when
the job runs.
:param job_func: The function to be scheduled
:return: The invoked job instance
self.job_func = functools.partial(job_func, *args, **kwargs)
functools.update_wrapper(self.job_func, job_func)
except AttributeError:
# job_funcs already wrapped by functools.partial won't have
# __name__, __module__ or __doc__ and the update_wrapper()
# call will fail.
return self
def should_run(self):
:return: ``True`` if the job should be run now.
return >= self.next_run
def run(self):
Run the job and immediately reschedule it.
:return: The return value returned by the `job_func`
"""'Running job %s', self)
ret = self.job_func()
self.last_run =
return ret
def _schedule_next_run(self):
Compute the instant when this job should run next.
if self.unit not in ('seconds', 'minutes', 'hours', 'days', 'weeks'):
raise ScheduleValueError('Invalid unit')
if self.latest is not None:
if not (self.latest >= self.interval):
raise ScheduleError('`latest` is greater than `interval`')
interval = random.randint(self.interval, self.latest)
interval = self.interval
self.period = datetime.timedelta(**{self.unit: interval})
self.next_run = + self.period
if self.start_day is not None:
if self.unit != 'weeks':
raise ScheduleValueError('`unit` should be \'weeks\'')
weekdays = (
if self.start_day not in weekdays:
raise ScheduleValueError('Invalid start day')
weekday = weekdays.index(self.start_day)
days_ahead = weekday - self.next_run.weekday()
if days_ahead <= 0: # Target day already happened this week
days_ahead += 7
self.next_run += datetime.timedelta(days_ahead) - self.period
if self.at_time is not None:
if (self.unit not in ('days', 'hours', 'minutes')
and self.start_day is None):
raise ScheduleValueError(('Invalid unit without'
' specifying start day'))
kwargs = {
'second': self.at_time.second,
'microsecond': 0
if self.unit == 'days' or self.start_day is not None:
kwargs['hour'] = self.at_time.hour
if self.unit in ['days', 'hours'] or self.start_day is not None:
kwargs['minute'] = self.at_time.minute
self.next_run = self.next_run.replace(**kwargs)
# If we are running for the first time, make sure we run
# at the specified time *today* (or *this hour*) as well
if not self.last_run:
now =
if (self.unit == 'days' and self.at_time > now.time() and
self.interval == 1):
self.next_run = self.next_run - datetime.timedelta(days=1)
elif self.unit == 'hours' \
and self.at_time.minute > now.minute \
or (self.at_time.minute == now.minute
and self.at_time.second > now.second):
self.next_run = self.next_run - datetime.timedelta(hours=1)
elif self.unit == 'minutes' \
and self.at_time.second > now.second:
self.next_run = self.next_run - \
if self.start_day is not None and self.at_time is not None:
# Let's see if we will still make that time we specified today
if (self.next_run - >= 7:
self.next_run -= self.period
# The following methods are shortcuts for not having to
# create a Scheduler instance:
#: Default :class:`Scheduler <Scheduler>` object
default_scheduler = Scheduler()
#: Default :class:`Jobs <Job>` list
jobs = # todo: should this be a copy, e.g. jobs()?
def every(interval=1):
"""Calls :meth:`every <Scheduler.every>` on the
:data:`default scheduler instance <default_scheduler>`.
return default_scheduler.every(interval)
def run_pending():
"""Calls :meth:`run_pending <Scheduler.run_pending>` on the
:data:`default scheduler instance <default_scheduler>`.
def run_all(delay_seconds=0):
"""Calls :meth:`run_all <Scheduler.run_all>` on the
:data:`default scheduler instance <default_scheduler>`.
def clear(tag=None):
"""Calls :meth:`clear <Scheduler.clear>` on the
:data:`default scheduler instance <default_scheduler>`.
def cancel_job(job):
"""Calls :meth:`cancel_job <Scheduler.cancel_job>` on the
:data:`default scheduler instance <default_scheduler>`.
def next_run():
"""Calls :meth:`next_run <Scheduler.next_run>` on the
:data:`default scheduler instance <default_scheduler>`.
return default_scheduler.next_run
def idle_seconds():
"""Calls :meth:`idle_seconds <Scheduler.idle_seconds>` on the
:data:`default scheduler instance <default_scheduler>`.
return default_scheduler.idle_seconds
You can’t perform that action at this time.