# pylint: disable = too-many-lines
"""SQLAlchemy-powered model definitions for jobs."""
from os import makedirs
from os.path import join, exists
from shutil import copy2
from time import time
from json import dumps
from collections import OrderedDict
from re import sub as re_sub
from sqlalchemy import Column, ForeignKey, String, Text, Enum, Boolean
from sqlalchemy import Float, Integer, PickleType, DateTime
from sqlalchemy.orm import relationship
from lxml import etree
import colander
from chrysalio.lib.i18n import view_i18n_labels, edit_i18n_labels
from chrysalio.lib.i18n import schema_i18n_labels, defaults_i18n_labels
from chrysalio.lib.i18n import record_format_i18n
from chrysalio.lib.utils import make_id, deltatime_label, age
from chrysalio.lib.xml import i18n_xml_text, db2xml_i18n_labels
from chrysalio.helpers.literal import Literal
from chrysalio.helpers.builder import Builder
from chrysalio.models import ID_LEN, NAME_LEN, LABEL_LEN, DESCRIPTION_LEN
from chrysalio.models import MODULE_LEN, VALUE_LEN, EMAIL_LEN
from chrysalio.models import DBDeclarativeClass
from chrysalio.models.dbbase import DBBaseClass
from chrysalio.models.dbuser import DBUser
from chrysalio.models.dbgroup import DBGroup
from ..lib.i18n import _
from ..relaxng import RELAXNG_CIOSERVICE
SERVICE_LEN = 128
JOB_EVERY_UNIT = {'day': _('day'), 'hour': _('hour'), 'minute': _('minute')}
JOB_DEFAULT_TTL = 300
JOB_DEFAULT_TTL_LOG = 604800
JOB_LOG_STATUS_LABELS = OrderedDict(( # yapf: disable
('error', _('Error')), ('warning', _('Warning')),
('info', _('Information')), ('trace', _('Trace')),
('running', _('In progress'))))
JOB_ACCESSES = {'free': _('free'), 'restricted': _('restricted')}
THREADED = {'no': _('no'), 'yes': _('yes'), 'monitored': _('monitored')}
# =============================================================================
[docs]
class DBJob(DBDeclarativeClass, DBBaseClass):
"""SQLAlchemy-powered job class."""
# pylint: disable = too-many-instance-attributes
suffix = 'ciojob'
attachments_dir = 'Jobs'
_settings_tabs = (
_('Information'), _('Execution'), _('Values'), _('Authorized users'),
_('Authorized groups'))
__tablename__ = 'srv_jobs'
__table_args__ = {'mysql_engine': 'InnoDB'}
job_id = Column(String(ID_LEN), primary_key=True)
i18n_label = Column(Text(), nullable=False)
i18n_description = Column(PickleType(1))
attachments_key = Column(String(ID_LEN + 20))
picture = Column(String(ID_LEN + 4))
icon = Column(String(ID_LEN + 4))
threaded = Column(Enum(*THREADED.keys(), name='threaded'), default='no')
ttl = Column(Integer, default=JOB_DEFAULT_TTL)
ttl_log = Column(Integer, default=JOB_DEFAULT_TTL_LOG)
service = Column(String(SERVICE_LEN), nullable=False)
context = Column(String(NAME_LEN))
access = Column(
Enum(*JOB_ACCESSES.keys(), name='job_access_enum'), default='free')
priority = Column(Integer, default=0)
every = Column(Integer, default=0)
every_unit = Column(
Enum(*JOB_EVERY_UNIT.keys(), name='every_enum'), default='minute')
every_at = Column(Integer, default=0)
stopped = Column(Boolean(name='stopped'), default=False)
last_trigger = Column(DateTime)
last_execution = Column(DateTime)
areas = relationship('DBJobArea', cascade='all, delete')
values = relationship('DBJobValue', cascade='all, delete')
users = relationship('DBJobUser', cascade='all, delete')
groups = relationship('DBJobGroup', cascade='all, delete')
logs = relationship('DBJobLog', cascade='all, delete')
# -------------------------------------------------------------------------
[docs]
@classmethod
def xml2db(cls, dbsession, job_elt, error_if_exists=True, kwargs=None):
"""Load a job from a XML element.
:type dbsession: sqlalchemy.orm.session.Session
:param dbsession:
SQLAlchemy session.
:type job_elt: lxml.etree.Element
:param job_elt:
Job XML element.
:param bool error_if_exists: (default=True)
It returns an error if job already exists.
:param dict kwargs: (optional)
Dictionary of keyword arguments.
:rtype: :class:`pyramid.i18n.TranslationString` or ``None``
:return:
Error message or ``None``.
"""
# pylint: disable = unused-argument
job_id = make_id(job_elt.get('id'), 'token', ID_LEN)
dbjob = dbsession.query(cls).filter_by(job_id=job_id).first()
if dbjob is not None:
if error_if_exists:
return _('Job "${j}" already exists.', {'j': job_id})
return None
# Create job
record = cls.record_from_xml(job_id, job_elt)
error = cls.record_format(record)
if error:
return error
dbjob = cls(**record)
dbsession.add(dbjob)
# Add areas
dbsession.flush()
namespace = RELAXNG_CIOSERVICE['namespace']
for elt in job_elt.findall(
'{{{0}}}areas/{{{0}}}area'.format(namespace)):
dbjob.areas.append(DBJobArea(area_id=elt.text.strip()))
# Add values
for elt in job_elt.findall(
'{{{0}}}values/{{{0}}}value'.format(namespace)):
dbjob.values.append(
DBJobValue(
variable=elt.get('variable')[:ID_LEN],
value='' if elt.text is None else elt.text[:VALUE_LEN]))
# Add users
kwargs = {} if kwargs is None else kwargs
refs = kwargs.get('users', {})
done = set()
for elt in job_elt.findall(
'{{{0}}}users/{{{0}}}user'.format(namespace)):
user_id = refs.get(elt.text)
if user_id is not None and user_id not in done:
done.add(user_id)
dbjob.users.append(DBJobUser(user_id=user_id))
# Add groups
refs = kwargs.get('groups', ())
done = set()
for elt in job_elt.findall(
'{{{0}}}groups/{{{0}}}group'.format(namespace)):
group_id = elt.text
if group_id and group_id in refs and group_id not in done:
done.add(group_id)
dbjob.groups.append(DBJobGroup(group_id=group_id))
# Add logs
for elt in job_elt.findall('{{{0}}}logs/{{{0}}}log'.format(namespace)):
dbjob.logs.append(
DBJobLog(
timestamp=float(elt.get('time')),
caller=elt.get('caller', ''),
build_id=elt.get('build'),
status=elt.get('status'),
text=re_sub(' *\n *', '\n', elt.text.strip()),
domain=elt.get('domain')))
return None
# -------------------------------------------------------------------------
[docs]
@classmethod
def record_from_xml(cls, job_id, job_elt):
"""Convert an user job XML element into a dictionary.
:param str job_id:
User job ID.
:type job_elt: lxml.etree.Element
:param job_elt:
Job XML element.
:rtype: dict
"""
namespace = RELAXNG_CIOSERVICE['namespace']
attachments_elt = job_elt.find('{{{0}}}attachments'.format(namespace))
service_elt = job_elt.find('{{{0}}}service'.format(namespace))
every_elt = job_elt.find('{{{0}}}every'.format(namespace))
return { # yapf: disable
'job_id': job_id,
'i18n_label': dumps(i18n_xml_text(
job_elt, 'ns0:label', {'ns0': namespace}), ensure_ascii=False),
'i18n_description': i18n_xml_text(
job_elt, 'ns0:description', {'ns0': namespace}),
'attachments_key':
attachments_elt is not None and attachments_elt.get(
'key') or None,
'picture':
attachments_elt is not None and attachments_elt.findtext(
'{{{0}}}picture'.format(namespace)) or None,
'icon': attachments_elt is not None and attachments_elt.findtext(
'{{{0}}}icon'.format(namespace)) or None,
'service': service_elt.text.strip(),
'context': service_elt.get('context'),
'threaded': service_elt.get('threaded', 'no').replace(
'true', 'yes'),
'ttl': service_elt.get('ttl', JOB_DEFAULT_TTL),
'ttl_log': service_elt.get('ttl-log', JOB_DEFAULT_TTL_LOG),
'access': job_elt.findtext('{{{0}}}access'.format(namespace)),
'priority': int(job_elt.findtext(
'{{{0}}}priority'.format(namespace)) or '0'),
'every': every_elt is not None and int(every_elt.text or '1') or 0,
'every_unit': every_elt is not None and every_elt.get(
'unit') or None,
'every_at': every_elt is not None and every_elt.get(
'at', '0') or None,
'stopped': every_elt is not None and every_elt.get(
'stopped') in ('true', '1')}
# -------------------------------------------------------------------------
# -------------------------------------------------------------------------
[docs]
def db2xml(self, dbsession):
"""Serialize a job to a XML representation.
:type dbsession: sqlalchemy.orm.session.Session
:param dbsession:
SQLAlchemy session.
:rtype: lxml.etree.Element
"""
# pylint: disable = too-many-branches
job_elt = etree.Element('job')
job_elt.set('id', self.job_id)
# Labels and descriptions
db2xml_i18n_labels(self, job_elt, 3)
# Attachments
if self.attachments_key:
elt = etree.SubElement(
job_elt, 'attachments', key=self.attachments_key)
if self.picture:
etree.SubElement(elt, 'picture').text = self.picture
if self.icon:
etree.SubElement(elt, 'icon').text = self.icon
# Service
elt = etree.SubElement(job_elt, 'service')
elt.text = self.service
if self.context:
elt.set('context', self.context)
if self.threaded != 'no':
elt.set('threaded', self.threaded)
if self.ttl != JOB_DEFAULT_TTL:
elt.set('ttl', str(self.ttl))
if self.ttl_log != JOB_DEFAULT_TTL_LOG:
elt.set('ttl-log', str(self.ttl_log))
# Access
if self.access != 'free':
etree.SubElement(job_elt, 'access').text = self.access
# Priority
if self.priority:
etree.SubElement(job_elt, 'priority').text = str(self.priority)
# Areas
if self.areas:
elt = etree.SubElement(job_elt, 'areas')
for dbarea in self.areas:
etree.SubElement(elt, 'area').text = dbarea.area_id
# Every
if self.every:
elt = etree.SubElement(job_elt, 'every', unit=self.every_unit)
elt.text = str(self.every)
if self.every_at:
elt.set('at', '{0:0>2}'.format(self.every_at))
if self.stopped:
elt.set('stopped', 'true')
self.db2xml_extra(dbsession, job_elt)
return job_elt
# -------------------------------------------------------------------------
# -------------------------------------------------------------------------
[docs]
def attachments2directory(self, attachments, directory):
"""Copy from attachments directory the file corresponding to the job.
See: meth:`.models.DBBaseClass.attachments2directory`
"""
if not self.attachments_key or (not self.picture and not self.icon):
return
picture = join(
attachments, self.attachments_dir, self.attachments_key,
self.picture) if self.picture else None
icon = join(
attachments, self.attachments_dir, self.attachments_key,
self.icon) if self.icon else None
if (not picture or not exists(picture)) and (not icon
or not exists(icon)):
return
target = join(directory, self.attachments_dir, self.attachments_key)
if not exists(target):
makedirs(target)
if picture and exists(picture):
copy2(picture, target)
if icon and exists(icon):
copy2(icon, target)
# -------------------------------------------------------------------------
[docs]
def tab4view(self, request, tab_index, form, user_filter, user_paging):
"""Generate the tab content of a job.
:type request: pyramid.request.Request
:param request:
Current request.
:param int index:
Index of the tab.
:type form: .lib.form.Form
:param form:
Current form object.
:type user_filter: .lib.filter.Filter
:param user_filter:
Filter for users.
:type user_paging: .lib.paging.Paging
:param user_paging:
Paging for job users.
:rtype: chrysalio.helpers.literal.Literal
"""
if tab_index == 0:
return self._tab4view_information(request, form)
if tab_index == 1:
return self._tab4view_execution(request, form)
if tab_index == 2:
return self._tab4view_values(request, form)
if tab_index == 3:
return self._tab4view_users(
request, form, user_filter, user_paging)
if tab_index == 4:
return self._tab4view_groups(request)
return ''
# -------------------------------------------------------------------------
def _tab4view_information(self, request, form):
"""Generate the information tab.
:type request: pyramid.request.Request
:param request:
Current request.
:type form: .lib.form.Form
:param form:
Current form object.
:rtype: chrysalio.helpers.literal.Literal
"""
lang = request.locale_name
translate = request.localizer.translate
service = request.registry['services'].get(self.service) \
if 'services' in request.registry else None
label = service.label if service is not None else _('Unknown service')
html = form.grid_item(
translate(_('Identifier:')), self.job_id, clear=True)
html += form.grid_item(
translate(_('Service:')),
translate(label),
title=self.service,
clear=True)
html += form.grid_item(
translate(_('Service context:')),
self.context if service is not None else None,
clear=True)
html += view_i18n_labels(request, form, self)
html += form.grid_item(
translate(_('In a thread:')),
translate(THREADED[self.threaded]),
clear=True)
html += form.grid_item(
translate(_('Time To Live:')),
deltatime_label(self.ttl, lang=lang),
clear=True)
html += form.grid_item(
translate(_('Log Time To Live:')),
deltatime_label(self.ttl_log, lang=lang),
clear=True)
html += form.grid_item(
translate(_('Access:')),
translate(JOB_ACCESSES[self.access]),
clear=True)
html += form.grid_item(
translate(_('Priority:')),
self.priority and str(self.priority),
clear=True)
html += Builder().h3(translate(_('Application areas')))
if self.areas:
labels = {'cioexecute': _('Scheduled jobs')}
for module in request.registry.get('modules', {}).values():
labels.update(module.areas)
for dbarea in self.areas:
html += Builder().div(
'✔ {0}'.format(
translate(labels.get(dbarea.area_id, dbarea.area_id))),
title=dbarea.area_id)
else:
html += Builder().div(translate(_('Available everywhere.')))
return html
# -------------------------------------------------------------------------
def _tab4view_execution(self, request, form):
"""Generate the execution tab.
:type request: pyramid.request.Request
:param request:
Current request.
:type form: .lib.form.Form
:param form:
Current form object.
:rtype: chrysalio.helpers.literal.Literal
"""
lang = request.locale_name
translate = request.localizer.translate
if not self.every:
every = _('manual')
elif self.every_unit == 'day':
every = _(
'${e} at ${a}', {
'e': deltatime_label(days=self.every, lang=lang),
'a': deltatime_label(minutes=self.every_at, lang=lang)
})
elif self.every_unit == 'hour':
every = _(
'${e} at ${a}', {
'e': deltatime_label(hours=self.every, lang=lang),
'a': deltatime_label(minutes=self.every_at, lang=lang)
})
else:
every = deltatime_label(minutes=self.every, lang=lang)
html = form.grid_item(
translate(
_('Execution every:') if self.every else _('Execution:')),
translate(every),
clear=True)
if self.last_trigger:
html += form.grid_item(
translate(_('Last trigger:')),
translate(age(self.last_trigger)),
title=self.last_trigger.isoformat(' ').partition('.')[0],
clear=True)
if self.last_execution:
html += form.grid_item(
translate(_('Last execution:')),
translate(age(self.last_execution)),
title=self.last_execution.isoformat(' ').partition('.')[0],
clear=True)
html += form.grid_item(
translate(_('Stopped:')),
self.stopped and translate(_('yes')),
clear=True)
return html
# -------------------------------------------------------------------------
def _tab4view_values(self, request, form):
"""Generate the value tab.
:type request: pyramid.request.Request
:param request:
Current request.
:type form: .lib.form.Form
:param form:
Current form object.
:rtype: chrysalio.helpers.literal.Literal
"""
translate = request.localizer.translate
values = {k.variable: k.value for k in self.values}
service = request.registry['services'].get(self.service) \
if 'services' in request.registry else None
html = '' if service is None else service.values_tabview(
request, self.context, form, values)
if not html:
html = translate(_('This job does not define variables.'))
return html
# -------------------------------------------------------------------------
def _tab4view_users(self, request, form, user_filter, user_paging):
"""Generate the users tab.
:type request: pyramid.request.Request
:param request:
Current request.
:type form: .lib.form.Form
:param form:
Current form object.
:type user_filter: .lib.filter.Filter
:param user_filter:
Filter for users.
:type user_paging: .lib.paging.Paging
:param user_paging:
Paging for warehouse users.
"""
translate = request.localizer.translate
if self.access == 'free':
return translate(_('Access to this job is not restricted.'))
if not self.users:
return translate(_('No user is authorized.'))
html = DBUser.paging_filter(request, form, user_filter, user_paging)
html += self._user_thead(request, user_paging)
for dbuser in user_paging:
html += '<tr>'\
'<th><a href="{user_view}">{login}</a></th>'\
'<td class="cioOptional">{fname}</td><td>{lname}</td>'\
'</tr>\n'.format(
user_view=request.route_path(
'user_view', user_id=dbuser.user_id),
login=dbuser.login,
fname=dbuser.first_name or '',
lname=dbuser.last_name)
html += '</tbody>\n</table>\n'
return Literal(html)
# -------------------------------------------------------------------------
def _tab4view_groups(self, request):
"""Generate the users tab.
:type request: pyramid.request.Request
:param request:
Current request.
"""
translate = request.localizer.translate
if self.access == 'free':
return translate(_('Access to this job is not restricted.'))
if not self.groups:
return translate(_('No group is authorized.'))
dbgroups = request.dbsession.query(DBGroup).filter(
DBGroup.group_id.in_([k.group_id
for k in self.groups])).order_by('group_id')
html = '<table>\n<thead>\n'\
'<tr><th>{label}</th><th>{description}</th></tr>\n'\
'</thead>\n<tbody>\n'.format(
label=translate(_('Label')),
description=translate(_('Description')))
for dbgroup in dbgroups:
html += \
'<tr><th><a href="{group_view}">{label}</a></td>'\
'<td>{description}</td></tr>\n'.format(
group_view=request.route_path(
'group_view', group_id=dbgroup.group_id),
label=dbgroup.label(request),
description=dbgroup.description(request))
html += '</tbody>\n</table>\n'
return Literal(html)
# -------------------------------------------------------------------------
[docs]
@classmethod
def settings_schema(
cls, request, defaults, groups, dbjob=None, relax=False):
"""Return a Colander schema to edit a job.
:type request: pyramid.request.Request
:param request:
Current request.
:param dict defaults:
Default values for the form set by the user paging object.
:param dict groups:
A dictionary such as ``{group_id: (label, description),...}``.
:type dbjob: DBJob
:param dbjob: (optional)
Current scheduled job SqlAlchemy object.
:param bool relax: (default=False)
If ``True``, ``required`` is ignored.
:rtype: tuple
:return:
A tuple such as ``(schema, defaults)``.
"""
# Informations
# yapf: disable
schema = colander.SchemaNode(colander.Mapping())
if dbjob is None:
schema.add(colander.SchemaNode(
colander.String(), name='job_id',
validator=colander.All(
colander.Regex(r'^[a-z][a-z0-9_]+$'),
colander.Length(min=2, max=ID_LEN))))
schema.add(colander.SchemaNode(
colander.String(), name='service',
validator=colander.Length(max=SERVICE_LEN)))
schema.add(colander.SchemaNode(
colander.String(), name='context',
validator=colander.Length(max=NAME_LEN), missing=False))
schema_i18n_labels(request, schema, LABEL_LEN, DESCRIPTION_LEN)
schema.add(colander.SchemaNode(
colander.String(), name='threaded',
validator=colander.OneOf(THREADED.keys()), missing='no'))
schema.add(colander.SchemaNode(
colander.Integer(), name='ttl', missing=JOB_DEFAULT_TTL))
schema.add(colander.SchemaNode(
colander.Integer(), name='ttl_log', missing=JOB_DEFAULT_TTL_LOG))
schema.add(colander.SchemaNode(
colander.String(), name='access',
validator=colander.OneOf(JOB_ACCESSES.keys()), missing='free'))
schema.add(colander.SchemaNode(
colander.Integer(), name='priority', missing=0))
schema.add(colander.SchemaNode(
colander.Integer(), name='every', missing=1))
schema.add(colander.SchemaNode(
colander.String(), name='every_unit',
validator=colander.OneOf(JOB_EVERY_UNIT.keys()), missing='minute'))
schema.add(colander.SchemaNode(
colander.String(), name='every_at', missing='0'))
schema.add(colander.SchemaNode(
colander.Boolean(), name='stopped', missing=False))
# Areas
schema.add(colander.SchemaNode(
colander.Boolean(), name='area:all', missing=False))
schema.add(colander.SchemaNode(
colander.Boolean(), name='area:cioexecute', missing=False))
for module in request.registry.get('modules', {}).values():
for area_id in module.areas:
schema.add(colander.SchemaNode(
colander.Boolean(), name='area:{0}'.format(area_id),
missing=False))
# Groups
for group_id in groups:
schema.add(colander.SchemaNode(
colander.Boolean(), name='grp:{0}'.format(group_id),
missing=False))
# Defaults
if dbjob is None:
defaults.update({
'ttl': JOB_DEFAULT_TTL, 'ttl_log': JOB_DEFAULT_TTL_LOG,
'every': 6, 'every_unit': 'minute', 'access': 'free',
'stopped': True, 'area:all': True})
else:
defaults.update(defaults_i18n_labels(dbjob))
service = request.registry['services'].get(dbjob.service) \
if 'services' in request.registry else None
if dbjob.areas:
for dbarea in dbjob.areas:
defaults['area:{0}'.format(dbarea.area_id)] = True
else:
defaults['area:all'] = True
for dbuser in dbjob.users:
defaults['usr:{0}'.format(dbuser.user_id)] = True
for dbgroup in dbjob.groups:
defaults['grp:{0}'.format(dbgroup.group_id)] = True
if service is not None:
service.values_schema(schema, defaults, dbjob, relax=relax)
# yapf: enable
return schema, defaults
# -------------------------------------------------------------------------
[docs]
@classmethod
def tab4edit(
cls,
request,
tab_index,
form,
user_filter,
user_paging,
groups,
dbjob=None):
"""Generate the tab content of user job for edition.
:type request: pyramid.request.Request
:param request:
Current request.
:param int tab_index:
Index of the tab.
:type form: .lib.form.Form
:param form:
Current form object.
:type user_filter: .lib.filter.Filter
:param user_filter:
Filter for users.
:type user_paging: .lib.paging.Paging
:param user_paging:
Paging for all users.
:param dict groups:
A dictionary such as ``{group_id: (label, description),...}``.
:type dbjob: DBJob
:param dbjob: (optional)
Current user job SqlAlchemy object.
:rtype: chrysalio.helpers.literal.Literal
"""
# pylint: disable = too-many-arguments, too-many-positional-arguments
if tab_index == 0:
return cls._tab4edit_information(request, form, dbjob)
if tab_index == 1:
return cls._tab4edit_execution(request, form)
if tab_index == 2:
return cls._tab4edit_values(request, form, dbjob)
if tab_index == 3:
return cls._tab4edit_users(
request, form, user_filter, user_paging, dbjob)
if tab_index == 4:
return cls._tab4edit_groups(request, form, groups, dbjob)
return ''
# -------------------------------------------------------------------------
@classmethod
def _tab4edit_information(cls, request, form, dbjob):
"""Generate the information tab for edition.
:type request: pyramid.request.Request
:param request:
Current request.
:type form: .lib.form.Form
:param form:
Current form object.
:type dbjob: DBJob
:param dbjob:
Current user job SqlAlchemy object.
:rtype: chrysalio.helpers.literal.Literal
"""
translate = request.localizer.translate
if dbjob is None:
service_labels = dict(
(k, request.registry['services'][k].label)
for k in request.registry.get('services', ''))
html = form.grid_text(
'job_id',
translate(_('Identifier:')),
required=True,
maxlength=ID_LEN,
clear=True)
html += form.grid_select(
'service',
translate(_('Service:')),
[('', ' ')] + list(service_labels.items()),
required=True,
clear=True)
html += form.grid_text(
'context',
translate(_('Service context:')),
maxlength=NAME_LEN,
clear=True)
else:
service = request.registry['services'].get(dbjob.service) \
if 'services' in request.registry else None
html = form.grid_item(
translate(_('Identifier:')), dbjob.job_id, clear=True)
html += form.grid_item(
translate(_('Service:')),
translate(
service.
label if service is not None else _('Unknown service')),
title=dbjob.service,
clear=True)
html += form.grid_item(
translate(_('Service context:')),
dbjob.context if service is not None else None,
clear=True)
html += edit_i18n_labels(request, form, LABEL_LEN, DESCRIPTION_LEN)
html += form.grid_select(
'threaded',
translate(_('In a thread:')),
THREADED.items(),
clear=True)
html += form.grid_text(
'ttl',
translate(_('Time To Live:')),
required=True,
maxlength=6,
hint=translate(_('In seconds.')),
clear=True)
html += form.grid_text(
'ttl_log',
translate(_('Log Time To Live:')),
required=True,
maxlength=8,
hint=translate(_('In seconds (ex.: 604800 = 1 week)')),
clear=True)
html += form.grid_select(
'access',
translate(_('Access:')), [('', ' ')] + list(JOB_ACCESSES.items()),
clear=True)
html += form.grid_text(
'priority', translate(_('Priority:')), maxlength=4, clear=True)
html += Builder().h3(translate(_('Application areas')))
html += Builder().div(
form.custom_checkbox('area:all') + Literal(
' <label for="areaall">{0}</label>'.format(
translate(_('Available everywhere')))))
html += Literal('<br/>')
html += Builder().div(
form.custom_checkbox('area:cioexecute') + Literal(
' <label for="areacioexecute">{0}</label>'.format(
translate(_('Scheduled jobs')))))
for module in request.registry.get('modules', {}).values():
for area_id in module.areas:
html += Builder().div(
form.custom_checkbox('area:{0}'.format(area_id)) + Literal(
' <label for="area{0}">{1}</label>'.format(
area_id, translate(module.areas[area_id]))))
return html
# -------------------------------------------------------------------------
@classmethod
def _tab4edit_execution(cls, request, form):
"""Generate the execution tab for edition.
:type request: pyramid.request.Request
:param request:
Current request.
:type form: .lib.form.Form
:param form:
Current form object.
:rtype: chrysalio.helpers.literal.Literal
"""
translate = request.localizer.translate
html = form.grid_select(
'every_unit',
translate(_('Execution time unit:')),
[('', ' ')] + list(JOB_EVERY_UNIT.items()),
required=True,
clear=True)
html += form.grid_text(
'every',
translate(_('Execution period:')),
required=True,
maxlength=3,
hint=translate(_('0 = manual launch')),
clear=True)
html += form.grid_text(
'every_at',
translate(_('At:')),
maxlength=5,
hint=translate(
_('MM or HH:MM, ex.: 25 or 10:45 (not used if unit is minute')
),
clear=True)
html += form.grid_custom_checkbox(
'stopped', translate(_('Stopped:')), clear=True)
return html
# -------------------------------------------------------------------------
@classmethod
def _tab4edit_values(cls, request, form, dbjob):
"""Generate the value tab for edition.
:type request: pyramid.request.Request
:param request:
Current request.
:type form: .lib.form.Form
:param form:
Current form object.
:type dbjob: DBJob
:param dbjob:
Current user job SqlAlchemy object.
:rtype: chrysalio.helpers.literal.Literal
"""
if dbjob is None:
return _('Create the job before setting variables.')
service = request.registry['services'].get(dbjob.service) \
if 'services' in request.registry else None
html = '' if service is None else service.values_tabedit(
request, dbjob.context, form)
return html
# -------------------------------------------------------------------------
@classmethod
def _tab4edit_users(cls, request, form, user_filter, user_paging, dbjob):
"""Generate the user tab for edition.
:type request: pyramid.request.Request
:param request:
Current request.
:type form: .lib.form.Form
:param form:
Current form object.
:type user_filter: .lib.filter.Filter
:param user_filter:
Filter for users.
:type user_paging: .lib.paging.Paging
:param user_paging:
Paging for all users.
:type dbjob: DBJob
:param dbjob:
Current job SqlAlchemy object.
:rtype: chrysalio.helpers.literal.Literal
"""
translate = request.localizer.translate
if dbjob and dbjob.access == 'free':
return translate(_('Access to this job is not restricted.'))
html = DBUser.paging_filter(request, form, user_filter, user_paging)
html += cls._user_thead(request, user_paging).replace(
'<tr>', '<tr><th class="cioCheckbox" id="check_all"></th>')
for dbuser in user_paging:
html += '<tr>'\
'<td class="cioCheckbox cioSelect">{check}{show}</td>'\
'<td>{login}</td>'\
'<td class="cioOptional">{fname}</td><td>{lname}</td>'\
'</tr>\n'.format(
check=form.custom_checkbox(
'usr:{0}'.format(dbuser.user_id)),
show=form.hidden('shw:{0}'.format(dbuser.user_id)),
login=dbuser.login,
fname=dbuser.first_name or '',
lname=dbuser.last_name)
html += '</tbody>\n</table>\n'
return Literal(html)
# -------------------------------------------------------------------------
@classmethod
def _tab4edit_groups(cls, request, form, groups, dbjob):
"""Generate the group tab for edition.
:type request: pyramid.request.Request
:param request:
Current request.
:type form: .lib.form.Form
:param form:
Current form object.
:param dict groups:
A dictionary such as ``{group_id: (label, description),...}``.
:type dbjob: DBJob
:param dbjob:
Current job SqlAlchemy object.
:rtype: chrysalio.helpers.literal.Literal
"""
translate = request.localizer.translate
if dbjob and dbjob.access == 'free':
return translate(_('Access to this job is not restricted.'))
html = '<table>\n<thead>\n'\
'<tr><th></th><th>{label}</th>'\
'<th>{description}</th></tr>\n</thead>\n<tbody>\n'.format(
label=translate(_('Label')),
description=translate(_('Description')))
for group_id in sorted(groups):
html += \
'<tr><td>{selected}</td>'\
'<th><label for="{id}">{label}</label></th>'\
'<td>{description}</td></tr>\n'.format(
selected=form.custom_checkbox('grp:{0}'.format(group_id)),
id='grp{0}'.format(group_id),
label=groups[group_id][0],
description=groups[group_id][1])
html += '</tbody>\n</table>\n'
return Literal(html)
# -------------------------------------------------------------------------
@classmethod
def _user_thead(cls, request, user_paging):
"""Table header for job users.
:type request: pyramid.request.Request
:param request:
Current request.
:type user_paging: .lib.paging.Paging
:param user_paging:
Paging for users.
:rtype: str
"""
translate = request.localizer.translate
return \
'<table class="cioPagingList">\n<thead><tr>'\
'<th>{login}</th>'\
'<th class="cioOptional">{fname}</th><th>{lname}</th>'\
'</tr></thead>\n<tbody>\n'.format(
login=user_paging.sortable_column(
translate(_('Login')), 'login'),
fname=user_paging.sortable_column(
translate(_('First name')), 'first_name'),
lname=user_paging.sortable_column(
translate(_('Last name')), 'last_name'))
# =============================================================================
[docs]
class DBJobArea(DBDeclarativeClass):
"""Class to link jobs with their areas (one-to-many)."""
# pylint: disable = too-few-public-methods
__tablename__ = 'srv_jobs_areas'
__table_args__ = {'mysql_engine': 'InnoDB'}
job_id = Column(
String(ID_LEN),
ForeignKey('srv_jobs.job_id', ondelete='CASCADE'),
primary_key=True)
area_id = Column(String(MODULE_LEN), primary_key=True)
# =============================================================================
[docs]
class DBJobValue(DBDeclarativeClass):
"""Class to link jobs with their values (one-to-many)."""
# pylint: disable = too-few-public-methods
__tablename__ = 'srv_jobs_values'
__table_args__ = {'mysql_engine': 'InnoDB'}
job_id = Column(
String(ID_LEN),
ForeignKey('srv_jobs.job_id', ondelete='CASCADE'),
primary_key=True)
variable = Column(String(ID_LEN), primary_key=True)
value = Column(String(VALUE_LEN))
# =============================================================================
[docs]
class DBJobUser(DBDeclarativeClass):
"""Class to link jobs with their authorized users (many-to-many)."""
# pylint: disable = too-few-public-methods
__tablename__ = 'srv_jobs_users'
__table_args__ = {'mysql_engine': 'InnoDB'}
__mapper_args__ = {'confirm_deleted_rows': False}
job_id = Column(
String(ID_LEN),
ForeignKey('srv_jobs.job_id', ondelete='CASCADE'),
primary_key=True)
user_id = Column(
Integer,
ForeignKey('users.user_id', ondelete='CASCADE'),
primary_key=True)
# =============================================================================
[docs]
class DBJobGroup(DBDeclarativeClass):
"""Class to link jobs with their authorized groups (many-to-many)."""
# pylint: disable = too-few-public-methods
__tablename__ = 'srv_jobs_groups'
__table_args__ = {'mysql_engine': 'InnoDB'}
job_id = Column(
String(ID_LEN),
ForeignKey('srv_jobs.job_id', ondelete='CASCADE'),
primary_key=True)
group_id = Column(
String(ID_LEN),
ForeignKey('groups.group_id', ondelete='CASCADE'),
primary_key=True)
# =============================================================================
[docs]
class DBJobLog(DBDeclarativeClass):
"""Class to link jobs with their log entries (one-to-many)."""
# pylint: disable = too-few-public-methods
__tablename__ = 'srv_jobs_logs'
__table_args__ = {'mysql_engine': 'InnoDB'}
job_id = Column(
String(ID_LEN),
ForeignKey('srv_jobs.job_id', ondelete='CASCADE'),
primary_key=True)
timestamp = Column(Float(precision=32), primary_key=True, default=time())
caller = Column(String(EMAIL_LEN), primary_key=True, default='')
build_id = Column(String(ID_LEN + 49))
status = Column(
Enum(*JOB_LOG_STATUS_LABELS.keys(), name='status_enum'),
nullable=False)
text = Column(Text, nullable=False)
domain = Column(String(ID_LEN))