"""A module for Chrysalio to transform or generate files."""
from __future__ import annotations
from sys import exit as sys_exit
from logging import getLogger
from os import scandir
from os.path import dirname, join, exists
from shutil import rmtree
from lxml import etree
from pyramid.config import Configurator
from pyramid.asset import abspath_from_asset_spec
from chrysalio.includes.modules.models import DBModule
from chrysalio.lib.config import settings_get_list, settings_get_directories
from chrysalio.lib.xml import load_xml
from chrysalio.modules import Module
from .lib.i18n import _, translate
from .relaxng import RELAXNG_CIOPROCESSOR
from .lib.pservice import PService
from .lib.pbuild import INSTALL_TTL, CLEAN_PERIOD, PROCESSOR_XML
from .processors.validate import ProcessorValidate
from .processors.transform import ProcessorTransform
LOG = getLogger(__name__)
# =============================================================================
[docs]def includeme(configurator):
"""Function to include module.
:type configurator: pyramid.config.Configurator
:param configurator:
Object used to do configuration declaration within the application.
"""
# Registration
Module.register(configurator, ModuleCioProcessor)
PService.register(configurator, PService)
if not isinstance(configurator, Configurator):
return
# Translation
configurator.add_translation_dirs(join(dirname(__file__), 'Locale'))
# =============================================================================
[docs]class ModuleCioProcessor(Module):
"""Class for CioProcessor, a module for periodically performing tasks.
:param str config_ini:
Absolute path to the configuration file (e.g. development.ini).
This module has the following attributes:
* ``locations``: a dictionary of root directories for service locations
"""
name = _('Processor')
implements = ('processor',)
dependencies = ('cioservice',)
relaxng: dict = RELAXNG_CIOPROCESSOR
areas: dict = {}
_DBModule = DBModule
# -------------------------------------------------------------------------
def __init__(self, config_ini):
"""Constructor method."""
super(ModuleCioProcessor, self).__init__(config_ini)
# Enabled configurations
settings = self._settings(config_ini)
self._enabled = settings_get_directories(
settings, 'processor', PROCESSOR_XML)
if not self._enabled:
sys_exit(translate(
_('*** ${n}: no processor enabled.', {'n': self.uid})))
# Environment
self._environment = {
'develop': settings.get('pbuild.develop') == 'true',
'build_root': settings.get('pbuild.root'),
'clean_period': int(
settings.get('pbuild.clean_period', CLEAN_PERIOD)),
'install_ttl': int(
settings.get('pbuild.install_ttl', INSTALL_TTL)),
'processors': {}
}
if not self._environment.get('build_root'):
sys_exit(translate(
_('*** ${n}: build zone is not defined.', {'n': self.uid})))
for root in settings_get_list(settings, 'processor.roots'):
root = abspath_from_asset_spec(root)
if exists(root):
for entry in scandir(root):
if entry.is_dir() and exists(
join(entry.path, PROCESSOR_XML)):
self._environment['processors'][
entry.name] = entry.path
# Locations
locations = settings_get_list(settings, 'locations')
settings = self._settings(config_ini, section='CioService')
locations = settings_get_list(settings, 'locations') + locations
self.locations = {}
for location in locations:
key, sep, path = location.partition(':')
if sep != ':':
sys_exit(translate(_(
'*** ${n}: "locations" must be a list of ID:PATH.',
{'n': self.uid})))
self.locations[key] = path
# Processors
self._processors = {}
# -------------------------------------------------------------------------
[docs] def populate(self, args, registry, dbsession):
"""Method called by populate script to complete the operation.
See: :meth:`chrysalio.modules.Module.populate`
"""
if '--remove-builds' in args.extra \
or (hasattr(args, 'remove_builds') and args.remove_builds):
if exists(self._environment['build_root']):
LOG.info(translate(
_('====== Removal of the builds directory')))
rmtree(self._environment['build_root'])
# -------------------------------------------------------------------------
[docs] def deactivate(self, registry, dbsession):
"""Method to deactivate the module.
See: :meth:`chrysalio.modules.Module.deactivate`
"""
# Processors
self._processors = {}
# -------------------------------------------------------------------------
[docs] def processor(self, processor_id):
"""Return the processor with ID ``processor_id`` or ``None``.
:param str processor_id:
ID of the processor to return.
:rtype: tuple
:return:
A tuple such as ``(processor, error)`` where ``processor`` is a
:class:`.processors.Processor` instanciated with processor
``processor_id`` environment and ``error`` an error message.
"""
if processor_id in self._processors:
return self._processors[processor_id], None
# Load XML configuration
if processor_id not in self._enabled:
return None, _('Unknown processor "${p}"', {'p': processor_id})
namespace = self.relaxng['namespace']
relaxngs = {
'{{{0}}}{1}'.format(
namespace, self.relaxng['root']): self.relaxng['file']}
tree = load_xml(
join(self._enabled[processor_id], PROCESSOR_XML),
relaxngs=relaxngs, xinclude=True)
# pylint: disable = protected-access
if not isinstance(tree, etree._ElementTree):
return None, tree
# pylint: enable = protected-access
# Instanciate processor
processor_elt = tree.getroot().find('*')
processor_name = processor_elt.tag[len(namespace) + 2:]
if processor_name == 'validate':
processor = ProcessorValidate(
self._environment, processor_id, processor_elt)
elif processor_name == 'transform':
processor = ProcessorTransform(
self._environment, processor_id, processor_elt)
else: # pragma: nocover
return None, _(
'Unknown processor "${n}" for "${i}"',
{'n': processor_name, 'i': processor_id})
if 'error' in processor.environment:
return None, processor.environment['error'] # pragma: nocover
self._processors[processor_id] = processor
return self._processors[processor_id], None