"""View callables to manage builds."""
from os.path import basename, dirname, join, commonpath, relpath
from time import time
from datetime import timedelta
from math import atan
from colander import SchemaNode, Mapping
from pyramid.view import view_config
from pyramid.httpexceptions import HTTPNotFound
from chrysalio.lib.utils import rst2html, deltatime_label
from chrysalio.lib.form import get_action, Form
from chrysalio.lib.i18n import translate_field
from chrysalio.helpers.literal import Literal
from chrysalio.views import BaseView
from ..models.dbjob import DBJob
from ..lib.utils import flag_image
from ..lib.i18n import _
BUILD_REFRESH = 1
# =============================================================================
[docs]
class BuildView(BaseView):
"""Class to manage jobs of a warehouse.
:type request: pyramid.request.Request
:param request:
Current request.
"""
# -------------------------------------------------------------------------
[docs]
@view_config(route_name='build_view',
renderer='cioservice:Templates/build_view.pt')
@view_config(route_name='build_view', renderer='json', xhr=True)
def view(self):
"""Show build."""
# Build environment
build_manager = self._request.registry[
'modules']['cioservice'].build_manager
build_id = self._request.matchdict['build_id']
build_env = build_manager.build_env(build_id, True)
# XHR
if self._request.is_xhr:
if build_env is None:
return {
'status': 'error', 'refresh': BUILD_REFRESH,
'file_percent': 0, 'file_name': '', 'step_percent': 0,
'step_trace': self._request.localizer.translate(_(
'This build no longer exists.')),
'runtime': ''}
progress = build_env.get('progress')
return {
'status': build_env['status'],
'refresh': self._refresh(build_env),
'file_percent': progress[0], 'file_name': progress[1],
'step_percent': progress[2],
'step_trace': self._request.localizer.translate(_(
'Pending...')) if build_env[
'status'] == 'in_spool' else rst2html(progress[3]),
'runtime': self._runtime(build_env)}
# Check build existence
if build_env is None:
raise HTTPNotFound(comment=_('This build no longer exists.'))
service = self._request.registry['services'].get(
build_env['job']['service_id'])
if service is None:
raise HTTPNotFound(comment=_('This service does not exist.'))
# Form
schema = SchemaNode(Mapping())
defaults = {}
dbjob = self._request.dbsession.query(DBJob).filter_by(
job_id=build_env['job']['job_id']).first()
service.values_schema(
schema, defaults, dbjob, not build_env['params'].get('all_values'))
defaults.update({
'val:{0}'.format(k): build_env['params']['values'][k]
for k in build_env['params']['values']})
form = Form(self._request, schema=schema, defaults=defaults)
# Action
action = get_action(self._request)[0]
if action == 'all!':
build_env['params']['all_values'] = not build_env['params'].get(
'all_values', False)
build_manager.register(
build_id, build_env['job'], build_env['params'])
elif action == 'run!':
build_env = self._run(build_manager, build_id, build_env, form)
elif action == 'stp!':
build_env = build_manager.stop(build_id, build_env)
# Files
files = build_env['params'].get('files', '')
if not files:
files = []
elif len(files) == 1:
files = [join(basename(dirname(files[0])), basename(files[0]))]
else:
files = commonpath(files)
files = [relpath(k, files) for k in build_env['params']['files']]
# Result
result = {'infos': [_('No results available.')]}
if build_env and build_env['status'] == 'completed' \
and 'result' in build_env:
result = build_env['result']
# Breadcrumbs & documentation
self._request.breadcrumbs(
translate_field(self._request, build_env['job']['i18n_label']))
self._request.documentation = '/build'
return {
'form': form, 'action': action, 'status': build_env['status'],
'refresh': self._refresh(build_env),
'progress': build_env.get('progress'),
'runtime': self._runtime(build_env),
'service': service,
'context': build_env['job']['context'],
'files': files,
'result': result,
'duration': deltatime_label(
build_env['duration'],
lang=self._request.session.get('lang', 'en')) if (
'result' in build_env and build_env['duration']) else '',
'all_values': build_env['params'].get('all_values', False),
'literal': Literal, 'rst2html': rst2html, 'flag_image': flag_image}
# -------------------------------------------------------------------------
@classmethod
def _refresh(cls, build_env):
"""Compute the best delay to refresh a page.
:param dict build_env:
The environment of the current build.
:rtype: int
"""
return int(
BUILD_REFRESH + 6 * atan((time() - build_env['launch']) / 40.0)) \
if build_env['launch'] else BUILD_REFRESH
# -------------------------------------------------------------------------
@classmethod
def _runtime(cls, build_env):
"""Return the current running time of the processing.
:param dict build_env:
The environment of the current build.
:rtype: str
"""
if not build_env['launch']:
return ''
return str(timedelta(seconds=time() - build_env['launch'])).split(
'.', maxsplit=1)[0]
# -------------------------------------------------------------------------
def _run(self, build_manager, build_id, build_env, form):
"""Execute a job on a build and return the new build environment.
:type build_manager: .lib.build_manager.BuildManager
:param build_manager:
Build manager.
:param str build_id:
Build ID.
:param dict build_env:
The environment of the current build.
:type form: chrysalio.lib.form.Form
:param form:
The current form.
:rtype: dict
"""
if not form.validate():
self._request.session.flash(_('Correct errors.'), 'alert')
return build_env
# Get values
modified = False
build_values = build_env['params']['values']
for value in form.values:
if value[:4] == 'val:':
if form.values[value] != build_values.get(value[4:]):
build_values[value[4:]] = form.values[value]
modified = True
if build_values.get(value[4:]) == ' ':
build_values[value[4:]] = ''
# Save modifications
if modified:
build_manager.register(
build_id, build_env['job'], build_env['params'])
# Launch the processing
build_manager.clean_logs(self._request.dbsession)
error = build_manager.run(build_id, build_env)
if error is not None:
self._request.session.flash(error, 'alert')
else:
modified = True
return build_manager.build_env(build_id) if modified else build_env