# Imports
from .library import Job
# Exports
__all__ = (
"Registry",
"schedule",
)
# Classes
[docs]class Registry(object):
"""Register jobs to be executed on a schedule."""
[docs] def __init__(self, label="global"):
self.label = label
self._registry = dict()
[docs] def get_jobs(self, app_name=None):
"""Get the jobs that have been registered.
:param app_name: Get the jobs only for the given app name.
:type app_name: str
:rtype: list[superdjango.contrib.scheduler.library.Job]
"""
a = list()
for _app_name, jobs in self._registry.items():
if app_name and app_name != _app_name:
continue
a += jobs
return a
[docs] def register(self, app_name, callback, active=False, at=None, description=None, frequency=None, interval=None,
label=None, **kwargs):
"""Register a job.
:param app_name: The name of the app.
:type app_name: str
:param callback: The callable to be executed.
:param active: Indicates the job is to be executed.
:type active: bool
:param at: The specific time at which job should run.
:type at: str
:param description: Optional, additional description of the job.
:type description: str
:param frequency: The frequency upon which the job runs.
:type frequency: str
:param interval: The interval upon which the job runs.
:type interval: int
:param label: The label of the job. Defaults to the name of the callback.
:type label: str
``kwargs`` are passed as parameters to the callback.
"""
if app_name not in self._registry:
self._registry[app_name] = list()
if not callable(callback):
raise TypeError("Can't register job for %s; it must be callable: %s" % (app_name, callback))
_label = label or callback.__name__.replace("_", " ")
if description is None and callback.__doc__:
description = callback.__doc__
job = Job(
_label,
active=active,
app_name=app_name,
at=at,
callback=callback,
description=description,
frequency=frequency,
interval=interval,
**kwargs
)
self._registry[app_name].append(job)
schedule = Registry()