job.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302
  1. from inspect import ismethod, isclass
  2. from uuid import uuid4
  3. import six
  4. from apscheduler.triggers.base import BaseTrigger
  5. from apscheduler.util import (
  6. ref_to_obj, obj_to_ref, datetime_repr, repr_escape, get_callable_name, check_callable_args,
  7. convert_to_datetime)
  8. try:
  9. from collections.abc import Iterable, Mapping
  10. except ImportError:
  11. from collections import Iterable, Mapping
  12. class Job(object):
  13. """
  14. Contains the options given when scheduling callables and its current schedule and other state.
  15. This class should never be instantiated by the user.
  16. :var str id: the unique identifier of this job
  17. :var str name: the description of this job
  18. :var func: the callable to execute
  19. :var tuple|list args: positional arguments to the callable
  20. :var dict kwargs: keyword arguments to the callable
  21. :var bool coalesce: whether to only run the job once when several run times are due
  22. :var trigger: the trigger object that controls the schedule of this job
  23. :var str executor: the name of the executor that will run this job
  24. :var int misfire_grace_time: the time (in seconds) how much this job's execution is allowed to
  25. be late (``None`` means "allow the job to run no matter how late it is")
  26. :var int max_instances: the maximum number of concurrently executing instances allowed for this
  27. job
  28. :var datetime.datetime next_run_time: the next scheduled run time of this job
  29. .. note::
  30. The ``misfire_grace_time`` has some non-obvious effects on job execution. See the
  31. :ref:`missed-job-executions` section in the documentation for an in-depth explanation.
  32. """
  33. __slots__ = ('_scheduler', '_jobstore_alias', 'id', 'trigger', 'executor', 'func', 'func_ref',
  34. 'args', 'kwargs', 'name', 'misfire_grace_time', 'coalesce', 'max_instances',
  35. 'next_run_time', '__weakref__')
  36. def __init__(self, scheduler, id=None, **kwargs):
  37. super(Job, self).__init__()
  38. self._scheduler = scheduler
  39. self._jobstore_alias = None
  40. self._modify(id=id or uuid4().hex, **kwargs)
  41. def modify(self, **changes):
  42. """
  43. Makes the given changes to this job and saves it in the associated job store.
  44. Accepted keyword arguments are the same as the variables on this class.
  45. .. seealso:: :meth:`~apscheduler.schedulers.base.BaseScheduler.modify_job`
  46. :return Job: this job instance
  47. """
  48. self._scheduler.modify_job(self.id, self._jobstore_alias, **changes)
  49. return self
  50. def reschedule(self, trigger, **trigger_args):
  51. """
  52. Shortcut for switching the trigger on this job.
  53. .. seealso:: :meth:`~apscheduler.schedulers.base.BaseScheduler.reschedule_job`
  54. :return Job: this job instance
  55. """
  56. self._scheduler.reschedule_job(self.id, self._jobstore_alias, trigger, **trigger_args)
  57. return self
  58. def pause(self):
  59. """
  60. Temporarily suspend the execution of this job.
  61. .. seealso:: :meth:`~apscheduler.schedulers.base.BaseScheduler.pause_job`
  62. :return Job: this job instance
  63. """
  64. self._scheduler.pause_job(self.id, self._jobstore_alias)
  65. return self
  66. def resume(self):
  67. """
  68. Resume the schedule of this job if previously paused.
  69. .. seealso:: :meth:`~apscheduler.schedulers.base.BaseScheduler.resume_job`
  70. :return Job: this job instance
  71. """
  72. self._scheduler.resume_job(self.id, self._jobstore_alias)
  73. return self
  74. def remove(self):
  75. """
  76. Unschedules this job and removes it from its associated job store.
  77. .. seealso:: :meth:`~apscheduler.schedulers.base.BaseScheduler.remove_job`
  78. """
  79. self._scheduler.remove_job(self.id, self._jobstore_alias)
  80. @property
  81. def pending(self):
  82. """
  83. Returns ``True`` if the referenced job is still waiting to be added to its designated job
  84. store.
  85. """
  86. return self._jobstore_alias is None
  87. #
  88. # Private API
  89. #
  90. def _get_run_times(self, now):
  91. """
  92. Computes the scheduled run times between ``next_run_time`` and ``now`` (inclusive).
  93. :type now: datetime.datetime
  94. :rtype: list[datetime.datetime]
  95. """
  96. run_times = []
  97. next_run_time = self.next_run_time
  98. while next_run_time and next_run_time <= now:
  99. run_times.append(next_run_time)
  100. next_run_time = self.trigger.get_next_fire_time(next_run_time, now)
  101. return run_times
  102. def _modify(self, **changes):
  103. """
  104. Validates the changes to the Job and makes the modifications if and only if all of them
  105. validate.
  106. """
  107. approved = {}
  108. if 'id' in changes:
  109. value = changes.pop('id')
  110. if not isinstance(value, six.string_types):
  111. raise TypeError("id must be a nonempty string")
  112. if hasattr(self, 'id'):
  113. raise ValueError('The job ID may not be changed')
  114. approved['id'] = value
  115. if 'func' in changes or 'args' in changes or 'kwargs' in changes:
  116. func = changes.pop('func') if 'func' in changes else self.func
  117. args = changes.pop('args') if 'args' in changes else self.args
  118. kwargs = changes.pop('kwargs') if 'kwargs' in changes else self.kwargs
  119. if isinstance(func, six.string_types):
  120. func_ref = func
  121. func = ref_to_obj(func)
  122. elif callable(func):
  123. try:
  124. func_ref = obj_to_ref(func)
  125. except ValueError:
  126. # If this happens, this Job won't be serializable
  127. func_ref = None
  128. else:
  129. raise TypeError('func must be a callable or a textual reference to one')
  130. if not hasattr(self, 'name') and changes.get('name', None) is None:
  131. changes['name'] = get_callable_name(func)
  132. if isinstance(args, six.string_types) or not isinstance(args, Iterable):
  133. raise TypeError('args must be a non-string iterable')
  134. if isinstance(kwargs, six.string_types) or not isinstance(kwargs, Mapping):
  135. raise TypeError('kwargs must be a dict-like object')
  136. check_callable_args(func, args, kwargs)
  137. approved['func'] = func
  138. approved['func_ref'] = func_ref
  139. approved['args'] = args
  140. approved['kwargs'] = kwargs
  141. if 'name' in changes:
  142. value = changes.pop('name')
  143. if not value or not isinstance(value, six.string_types):
  144. raise TypeError("name must be a nonempty string")
  145. approved['name'] = value
  146. if 'misfire_grace_time' in changes:
  147. value = changes.pop('misfire_grace_time')
  148. if value is not None and (not isinstance(value, six.integer_types) or value <= 0):
  149. raise TypeError('misfire_grace_time must be either None or a positive integer')
  150. approved['misfire_grace_time'] = value
  151. if 'coalesce' in changes:
  152. value = bool(changes.pop('coalesce'))
  153. approved['coalesce'] = value
  154. if 'max_instances' in changes:
  155. value = changes.pop('max_instances')
  156. if not isinstance(value, six.integer_types) or value <= 0:
  157. raise TypeError('max_instances must be a positive integer')
  158. approved['max_instances'] = value
  159. if 'trigger' in changes:
  160. trigger = changes.pop('trigger')
  161. if not isinstance(trigger, BaseTrigger):
  162. raise TypeError('Expected a trigger instance, got %s instead' %
  163. trigger.__class__.__name__)
  164. approved['trigger'] = trigger
  165. if 'executor' in changes:
  166. value = changes.pop('executor')
  167. if not isinstance(value, six.string_types):
  168. raise TypeError('executor must be a string')
  169. approved['executor'] = value
  170. if 'next_run_time' in changes:
  171. value = changes.pop('next_run_time')
  172. approved['next_run_time'] = convert_to_datetime(value, self._scheduler.timezone,
  173. 'next_run_time')
  174. if changes:
  175. raise AttributeError('The following are not modifiable attributes of Job: %s' %
  176. ', '.join(changes))
  177. for key, value in six.iteritems(approved):
  178. setattr(self, key, value)
  179. def __getstate__(self):
  180. # Don't allow this Job to be serialized if the function reference could not be determined
  181. if not self.func_ref:
  182. raise ValueError(
  183. 'This Job cannot be serialized since the reference to its callable (%r) could not '
  184. 'be determined. Consider giving a textual reference (module:function name) '
  185. 'instead.' % (self.func,))
  186. # Instance methods cannot survive serialization as-is, so store the "self" argument
  187. # explicitly
  188. func = self.func
  189. if ismethod(func) and not isclass(func.__self__) and obj_to_ref(func) == self.func_ref:
  190. args = (func.__self__,) + tuple(self.args)
  191. else:
  192. args = self.args
  193. return {
  194. 'version': 1,
  195. 'id': self.id,
  196. 'func': self.func_ref,
  197. 'trigger': self.trigger,
  198. 'executor': self.executor,
  199. 'args': args,
  200. 'kwargs': self.kwargs,
  201. 'name': self.name,
  202. 'misfire_grace_time': self.misfire_grace_time,
  203. 'coalesce': self.coalesce,
  204. 'max_instances': self.max_instances,
  205. 'next_run_time': self.next_run_time
  206. }
  207. def __setstate__(self, state):
  208. if state.get('version', 1) > 1:
  209. raise ValueError('Job has version %s, but only version 1 can be handled' %
  210. state['version'])
  211. self.id = state['id']
  212. self.func_ref = state['func']
  213. self.func = ref_to_obj(self.func_ref)
  214. self.trigger = state['trigger']
  215. self.executor = state['executor']
  216. self.args = state['args']
  217. self.kwargs = state['kwargs']
  218. self.name = state['name']
  219. self.misfire_grace_time = state['misfire_grace_time']
  220. self.coalesce = state['coalesce']
  221. self.max_instances = state['max_instances']
  222. self.next_run_time = state['next_run_time']
  223. def __eq__(self, other):
  224. if isinstance(other, Job):
  225. return self.id == other.id
  226. return NotImplemented
  227. def __repr__(self):
  228. return '<Job (id=%s name=%s)>' % (repr_escape(self.id), repr_escape(self.name))
  229. def __str__(self):
  230. return repr_escape(self.__unicode__())
  231. def __unicode__(self):
  232. if hasattr(self, 'next_run_time'):
  233. status = ('next run at: ' + datetime_repr(self.next_run_time) if
  234. self.next_run_time else 'paused')
  235. else:
  236. status = 'pending'
  237. return u'%s (trigger: %s, %s)' % (self.name, self.trigger, status)