Source code for cioservice.services.skeleton

# -*- coding: utf-8 -*-
"""An example of service."""

from os import listdir
from os.path import basename, exists
from time import sleep
from collections import OrderedDict

from ..lib.i18n import _, translate
from ..lib.utils import location_path2abs_path
from ..lib.service import LOCATION_REGEX, Service


# =============================================================================
[docs]def includeme(configurator): """Function to include `skeleton` service. :type configurator: pyramid.config.Configurator :param configurator: Object used to do configuration declaration within the application. """ Service.register(configurator, ServiceSkeleton)
# =============================================================================
[docs]class ServiceSkeleton(Service): """Class to manage `skeleton` service.""" label = _('Skeleton service') _variables_groups = { 'str': _('Strings'), 'bool': _('Boolean'), 'num': _('Numbers'), 'list': _('Lists'), 'regex': _('Regular expressions')} _variables = OrderedDict(( ('__template__', { 'type': 'regex', 'label': _('Template'), 'regex': LOCATION_REGEX}), ('published', { 'group': 'bool', 'type': 'boolean', 'label': _('Published (boolean)')}), ('message', { 'group': 'str', 'type': 'string', 'label': _('Message (string)'), 'required': True, 'visible': True}), ('mark', { 'group': 'num', 'type': 'integer', 'label': _('Mark (integer)')}), ('timer', { 'group': 'num', 'type': 'integer', 'label': _('Timer (integer)'), 'default': 0, 'visible': True}), ('cost', { 'group': 'num', 'type': 'decimal', 'label': _('Cost (decimal)')}), ('temperature', { 'group': 'regex', 'type': 'regex', 'label': _('Temperature (regex)'), 'default': '18 °C', 'regex': '(\\d+ °C|\\d+ °F)', 'hint': _('37 °C or 100 °F')}), ('color', { 'group': 'list', 'type': 'list', 'label': _('Color (closed list)'), 'default': 'blue', 'options': { 'blue': _('Blue'), 'white': _('White'), 'red': _('Red')}}))) _need_write_permission = True # ------------------------------------------------------------------------- def _run(self, build): """Execute the service on the build ``build``. See: :meth:`.lib.service.Service._run` """ timer = build.values.get('timer') or 0 build.progress_step(total=timer + 1) for input_file in build.files: if build.aborted(): break build.info('**▪ {0}**'.format(basename(input_file))) build.current['input_file'] = input_file build.progress_step(0) build.progress_file(increase=1) for tick in range(timer): if build.aborted(): break build.progress_step(increase=1) build.progress_save() build.info(_('*Tick ${t}*...', {'t': tick + 1})) sleep(1) build.progress_step(timer + 1) build.progress_save() build.info(_('**Job**: ${j}', {'j': build.job_id})) build.info(_('**Context**: ${c}', {'c': build.context or ''})) template = location_path2abs_path( build.locations, build.values.get('__template__')) if template is not None and exists(template): build.info(_('**Template**: ${t}', { 't': ', '.join(listdir(template))})) build.info(_('**Published**: ${p}', { 'p': translate(_('yes') if build.values.get( 'published') else _('no'), build.lang)})) build.info(_('**Message**: *${m}*', { 'm': build.values.get('message', 'Hello')})) build.info(_('**Mark**: ${m}', {'m': build.values.get('mark', '')})) build.info(_('**Favorite color**: ${c}', { 'c': translate(self._variables['color']['options'].get( build.values.get('color') or 'blue'), build.lang)})) build.output_info() build.aborted_message() self.write_traces(build, domain='main')