Source code for pyams_scheduler.interfaces

#
# Copyright (c) 2008-2015 Thierry Florac <tflorac AT ulthar.net>
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#

__docformat__ = 'restructuredtext'

from zope.annotation.interfaces import IAttributeAnnotatable
from zope.container.constraints import containers, contains
from zope.interface import Attribute, Interface, implementer
from zope.interface.interfaces import IObjectEvent, ObjectEvent
from zope.schema import Bool, Bytes, Choice, Datetime, Float, Int, List, Object, Text, TextLine

from pyams_zmq.interfaces import IZMQProcess

from pyams_scheduler import _


#
# Scheduler events
#

[docs]class IBeforeRunJobEvent(IObjectEvent): """Interface for events notified before a job is run"""
[docs]@implementer(IBeforeRunJobEvent) class BeforeRunJobEvent(ObjectEvent): """Before run job event"""
[docs]class IAfterRunJobEvent(IObjectEvent): """Interface for events notified after a job is run""" status = Attribute("Job execution status")
[docs]@implementer(IAfterRunJobEvent) class AfterRunJobEvent(ObjectEvent): """After run job event""" def __init__(self, object, status): super(AfterRunJobEvent, self).__init__(object) self.status = status
# # Task history item interface #
[docs]class ITaskHistoryContainer(Interface): """Task history container interface""" contains('pyams_scheduler.interfaces.ITaskHistory')
[docs]class ITaskHistory(Interface): """Scheduler task history item interface""" containers(ITaskHistoryContainer) date = Datetime(title="Execution date", required=True) status = Choice(title="Execution status", values=('OK', 'Warning', 'Error', 'Empty')) report = Text(title="Execution report", required=True)
# # Scheduler interface # SCHEDULER_NAME = 'Tasks scheduler' SCHEDULER_STARTER_KEY = 'pyams_scheduler.start_handler' SCHEDULER_HANDLER_KEY = 'pyams_scheduler.tcp_handler' SCHEDULER_AUTH_KEY = 'pyams_scheduler.allow_auth' SCHEDULER_CLIENTS_KEY = 'pyams_scheduler.allow_clients' SCHEDULER_JOBSTORE_KEY = 'pyams_scheduler.jobs'
[docs]class ISchedulerProcess(IZMQProcess): """Scheduler process marker interface"""
[docs]class ISchedulerHandler(Interface): """Scheduler manager marker interface"""
[docs]class IScheduler(IAttributeAnnotatable): """Scheduler interface""" contains('pyams_scheduler.interfaces.ITask') zodb_name = Choice(title=_("ZODB connection name"), description=_("Name of ZODB defining scheduler connection"), required=False, default='', vocabulary="PyAMS ZODB connections") report_mailer = Choice(title=_("Reports mailer"), description=_("Mail delivery utility used to send mails"), required=False, vocabulary='PyAMS mailer utilities') report_source = TextLine(title=_("Reports source"), description=_("Mail address from which reports will be sent"), required=False) internal_id = Attribute("Internal ID") def get_socket(self): """Get ZMQ socket matching scheduler utility""" def get_task(self, task_id): """Get task matching given task ID""" def get_jobs(self): """Get text output of running jobs""" tasks = List(title=_("Scheduler tasks"), description=_("List of tasks assigned to this scheduler"), required=False) history = List(title=_("History"), description=_("Task history"), value_type=Object(schema=ITaskHistory), readonly=True)
# # Scheduler task interface #
[docs]class IJobInfo(Interface): """Job interface""" id = TextLine(title="Job ID") next_run_time = Float(title="Job next run time") job_state = Bytes(title="Job state")
[docs]class ITaskInfo(Interface): """Scheduler task interface""" containers(IScheduler) name = TextLine(title=_("Task name"), description=_("Descriptive name given to this task"), required=True) schedule_mode = Choice(title=_("Scheduling mode"), description=_("Scheduling mode defines how task will be scheduled"), vocabulary="PyAMS scheduling modes", required=True) report_target = TextLine(title=_("Reports target(s)"), description=_("Mail address(es) to which execution reports will be sent; you can enter " "several addresses separated by semicolons"), required=False) errors_target = TextLine(title=_("Errors reports target(s)"), description=_("Mail address(es) to which error reports will be sent; you can enter " "several addresses separated by semicolons; keep empty to use normal " "reports target"), required=False) report_errors_only = Bool(title=_("Only report errors?"), description=_("If 'Yes', only error reports will be sent to given errors target"), required=True, default=False) send_empty_reports = Bool(title=_("Send empty reports?"), description=_("If 'No', empty reports will not be sent by mail"), required=True, default=False) keep_empty_reports = Bool(title=_("Keep empty reports history?"), description=_("If 'Yes', empty reports will be kept in task history"), required=True, default=False) history_duration = Int(title=_("History duration"), description=_("Number of days during which execution reports are kept in history; enter " "0 to remove limit"), required=False) history_length = Int(title=_("History max length"), description=_("Number of execution reports to keep in history; enter 0 to remove limit"), required=False)
[docs]class ITask(ITaskInfo, IAttributeAnnotatable): """Complete task interface""" settings_view_name = TextLine(title=_("Settings view name"), default='settings.html', required=False) history = List(title=_("History"), description=_("Task history"), value_type=Object(schema=ITaskHistory)) runnable = Attribute("Is the task runnable ?") internal_id = Attribute("Internal ID") def get_trigger(self, registry): """Get scheduler job trigger""" def get_scheduling_info(self, registry): """Get scheduling info""" def run(self, report, **kw): """Launch job execution""" def store_report(self, report, status): """Store execution report in task's history and send it by mail""" def send_report(self, report, status, target=None): """Store execution report in task's history and send it by mail""" def reset(self): """Re-schedule job execution""" def launch(self): """Ask task for immediate execution"""
# # Task scheduling modes interfaces #
[docs]class ITaskSchedulingMode(Interface): """Scheduler task scheduling mode""" marker_interface = Attribute("Class name of scheduling mode marker interface") schema = Attribute("Class name of scheduling mode info interface") def get_trigger(self, task): """Get trigger for the given task""" def schedule(self, task, scheduler): """Add given task to the scheduler"""
[docs]class ITaskSchedulingMarker(Interface): """Base interface for task scheduling mode markers"""
[docs]class IBaseTaskScheduling(Interface): """Base interface for task scheduling info""" active = Bool(title=_("Active task"), description=_("You can disable a task by selecting 'No'"), required=True, default=False) start_date = Datetime(title=_("First execution date"), description=_("Date from which scheduling should start"), required=False)
# Scheduler cron-style tasks interfaces SCHEDULER_TASK_CRON_INFO = 'pyams_scheduler.trigger.cron'
[docs]class ICronTask(Interface): """Cron-style task marker interface"""
[docs]class ICronTaskScheduling(IBaseTaskScheduling): """Base interface for cron-style scheduled tasks""" end_date = Datetime(title=_("Last execution date"), description=_("Date past which scheduling should end"), required=False) year = TextLine(title=_("Years"), description=_("Years for which to schedule the job"), required=False, default='*') month = TextLine(title=_("Months"), description=_("Months (1-12) for which to schedule the job"), required=False, default='*') day = TextLine(title=_("Month days"), description=_("Days (1-31) for which to schedule the job"), required=False, default='*') week = TextLine(title=_("Weeks"), description=_("Year weeks (1-53) for which to schedule the job"), required=False, default='*') day_of_week = TextLine(title=_("Week days"), description=_("Week days (0-6, with 0 as monday) for which to schedule the job"), required=False, default='*') hour = TextLine(title=_("Hours"), description=_("Hours (0-23) for which to schedule the job"), required=False, default='*') minute = TextLine(title=_("Minutes"), description=_("Minutes (0-59) for which to schedule the job"), required=False, default='*') second = TextLine(title=_("Seconds"), description=_("Seconds (0-59) for which to schedule the job"), required=False, default='0')
# Scheduler date-style tasks interface SCHEDULER_TASK_DATE_INFO = 'pyams_scheduler.trigger.date'
[docs]class IDateTask(Interface): """Date-style task marker interface"""
[docs]class IDateTaskScheduling(IBaseTaskScheduling): """Base interface for date-style scheduled tasks""" start_date = Datetime(title=_("Execution date"), description=_("Date on which execution should start"), required=True)
# Scheduler loop-style tasks interface SCHEDULER_TASK_LOOP_INFO = 'pyams_scheduler.trigger.loop'
[docs]class ILoopTask(Interface): """Loop-style task marker interface"""
[docs]class ILoopTaskScheduling(IBaseTaskScheduling): """Base interface for loop-style scheduled tasks""" end_date = Datetime(title=_("Last execution date"), description=_("Date past which scheduling should end"), required=False) weeks = Int(title=_("Weeks interval"), description=_("Number of weeks between executions"), required=True, default=0) days = Int(title=_("Days interval"), description=_("Number of days between executions"), required=True, default=0) hours = Int(title=_("Hours interval"), description=_("Number of hours between executions"), required=True, default=0) minutes = Int(title=_("Minutes interval"), description=_("Number of minutes between executions"), required=True, default=1) seconds = Int(title=_("Seconds interval"), description=_("Number of seconds between executions"), required=True, default=0)