Coverage for superdjango/contrib/scheduler/interfaces.py : 100%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Imports
3from .library import Job
5# Exports
7__all__ = (
8 "Registry",
9 "schedule",
10)
12# Classes
15class Registry(object):
16 """Register jobs to be executed on a schedule."""
18 def __init__(self, label="global"):
19 self.label = label
20 self._registry = dict()
22 def get_jobs(self, app_name=None):
23 """Get the jobs that have been registered.
25 :param app_name: Get the jobs only for the given app name.
26 :type app_name: str
28 :rtype: list[superdjango.contrib.scheduler.library.Job]
30 """
31 a = list()
32 for _app_name, jobs in self._registry.items():
33 if app_name and app_name != _app_name:
34 continue
36 a += jobs
38 return a
40 def register(self, app_name, callback, active=False, at=None, description=None, frequency=None, interval=None,
41 label=None, **kwargs):
42 """Register a job.
44 :param app_name: The name of the app.
45 :type app_name: str
47 :param callback: The callable to be executed.
49 :param active: Indicates the job is to be executed.
50 :type active: bool
52 :param at: The specific time at which job should run.
53 :type at: str
55 :param description: Optional, additional description of the job.
56 :type description: str
58 :param frequency: The frequency upon which the job runs.
59 :type frequency: str
61 :param interval: The interval upon which the job runs.
62 :type interval: int
64 :param label: The label of the job. Defaults to the name of the callback.
65 :type label: str
67 ``kwargs`` are passed as parameters to the callback.
69 """
70 if app_name not in self._registry:
71 self._registry[app_name] = list()
73 if not callable(callback):
74 raise TypeError("Can't register job for %s; it must be callable: %s" % (app_name, callback))
76 _label = label or callback.__name__.replace("_", " ")
78 if description is None and callback.__doc__:
79 description = callback.__doc__
81 job = Job(
82 _label,
83 active=active,
84 app_name=app_name,
85 at=at,
86 callback=callback,
87 description=description,
88 frequency=frequency,
89 interval=interval,
90 **kwargs
91 )
93 self._registry[app_name].append(job)
96schedule = Registry()