Source code for ciowarehouse.models.dbindexfield

"""SQLAlchemy-powered model definitions for index fields."""

from json import dumps, loads

from sqlalchemy.orm import relationship
from sqlalchemy import Column, ForeignKey, String, Integer, Text, Enum, Boolean
from lxml import etree

from chrysalio.lib.i18n import translate_field, record_format_i18n
from chrysalio.lib.utils import make_id
from chrysalio.lib.xml import XML_NS, i18n_xml_text, db2xml_i18n_labels
from chrysalio.models.dbbase import DBBaseClass
from chrysalio.models import DBDeclarativeClass, ID_LEN, LABEL_LEN
from ..relaxng import RELAXNG_CIOWAREHOUSE
from ..lib.i18n import _


INDEXFIELD_TYPES = {
    'file_name': _('File name'),
    'file_type': _('File type'),
    'file_size': _('File size'),
    'file_date': _('File date'),
    'vcs_author': _('VCS File author'),
    'vcs_date': _('VCS Modification time'),
    'vcs_message': _('VCS Message'),
    'id': _('ID'),
    'string': _('String'),
    'stems': _('Word stems'),
    'keyword': _('Keyword'),
    'integer': _('Integer'),
    'decimal': _('Decimal'),
    'boolean': _('Boolean'),
    'time': _('UNIX epoch time'),
    'date': _('Date'),
    'list': _('Closed list'),
    'palette': _('Color palette')}
INDEXFIELD_BUILTIN = (
    'time', 'directory', 'file_name', 'file_id', 'file_type',
    'file_size', 'file_date', 'only_groups', 'shared')
INDEXFIELD_DEFAULTS = ('filename', 'file_size', 'file_date')


# =============================================================================
[docs]class DBIndexfield(DBDeclarativeClass, DBBaseClass): """SQLAlchemy-powered index field class.""" suffix = 'cioidx' __tablename__ = 'wrh_indexfields' __table_args__ = {'mysql_engine': 'InnoDB'} indexfield_id = Column(String(ID_LEN), primary_key=True) i18n_label = Column(Text(), nullable=False) field_type = Column( Enum(*INDEXFIELD_TYPES.keys(), name='indexfield_type'), nullable=False) in_filter = Column(Integer(), nullable=False) stored = Column(Boolean(name='stored'), default=False) in_list = Column(Integer()) in_cards = Column(Integer()) in_class = Column(String(64)) choices = relationship('DBIndexfieldChoice', cascade='all, delete') # -------------------------------------------------------------------------
[docs] @classmethod def xml2db( cls, dbsession, indexfield_elt, error_if_exists=True, kwargs=None): """Load a index field from a XML element. :type dbsession: sqlalchemy.orm.session.Session :param dbsession: SQLAlchemy session. :type indexfield_elt: lxml.etree.Element :param indexfield_elt: Index field XML element. :param bool error_if_exists: (default=True) It returns an error if index field 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 # Check if already exists indexfield_id = make_id(indexfield_elt.get('id'), 'token', ID_LEN) dbindexfield = dbsession.query(cls).filter_by( indexfield_id=indexfield_id).first() if dbindexfield is not None: if error_if_exists: return _( 'Index field "${i}" already exists.', {'i': indexfield_id}) return None # Create indexfield record = cls.record_from_xml(indexfield_id, indexfield_elt) error = cls.record_format(record) if error: return error dbindexfield = cls(**record) dbsession.add(dbindexfield) # Add choices dbsession.flush() namespace = RELAXNG_CIOWAREHOUSE['namespace'] for elt in indexfield_elt.xpath( 'ns0:item|ns0:color', namespaces={'ns0': namespace}): dbindexfield.choices.append(DBIndexfieldChoice( value=elt.get('value'), i18n_label=dumps(i18n_xml_text( elt, 'ns:label', {'ns': namespace})))) return None
# -------------------------------------------------------------------------
[docs] @classmethod def record_from_xml(cls, indexfield_id, indexfield_elt): """Convert a indexfield XML element into a dictionary. :param str indexfield_id: Indexfield ID. :type indexfield_elt: lxml.etree.Element :param indexfield_elt: Indexfield XML element. :rtype: dict """ namespace = RELAXNG_CIOWAREHOUSE['namespace'] in_filter = indexfield_elt.get('in-filter') if in_filter == 'unused': in_filter = 0 elif in_filter == 'static': in_filter = int( indexfield_elt.xpath('count(preceding-sibling::*)')) - 100 else: in_filter = int( indexfield_elt.xpath('count(preceding-sibling::*)')) + 1 return { 'indexfield_id': indexfield_id, 'i18n_label': dumps(i18n_xml_text( indexfield_elt, 'ns0:label', {'ns0': namespace})), 'field_type': indexfield_elt.get('type'), 'in_filter': in_filter, 'stored': indexfield_elt.get('stored') == 'true', 'in_list': int(indexfield_elt.get('in-list', '0')) or None, 'in_cards': int(indexfield_elt.get('in-cards', '0')) or None, 'in_class': indexfield_elt.get('class')}
# -------------------------------------------------------------------------
[docs] @classmethod def record_format(cls, record): """Check and possibly correct a record before inserting it in the database. :param dict record: Dictionary of values to check. :rtype: ``None`` or :class:`pyramid.i18n.TranslationString` :return: ``None`` or error message. """ for k in [i for i in record if record[i] is None]: del record[k] # Index field ID if not record.get('indexfield_id'): return _('Index field without ID.') record['indexfield_id'] = make_id( record['indexfield_id'], 'token', ID_LEN) # Labels if not record_format_i18n(record): return _('Index field without label.') # Field type if not record.get('field_type'): return _('Index field without type.') return None
# -------------------------------------------------------------------------
[docs] def db2xml(self, dbsession=None): """Serialize a index field to a XML representation. :type dbsession: sqlalchemy.orm.session.Session :param dbsession: (optional) SQLAlchemy session. :rtype: lxml.etree.Element """ # pylint: disable = unused-argument indexfield_elt = etree.Element('indexfield') indexfield_elt.set('id', self.indexfield_id) indexfield_elt.set('type', self.field_type) if not self.in_filter: indexfield_elt.set('in-filter', 'unused') elif self.in_filter < 0: indexfield_elt.set('in-filter', 'static') if self.stored: indexfield_elt.set('stored', 'true') if self.in_list: indexfield_elt.set('in-list', str(self.in_list)) if self.in_cards: indexfield_elt.set('in-cards', str(self.in_cards)) if self.in_class: indexfield_elt.set('class', self.in_class) db2xml_i18n_labels(self, indexfield_elt, 6) if self.field_type in ('list', 'palette'): for item in self.choices: choice_elt = etree.SubElement( indexfield_elt, 'item' if self.field_type == 'list' else 'color', value=item.value) i18n = loads(item.i18n_label) for lang in sorted(i18n): elt = etree.SubElement(choice_elt, 'label') elt.set('{0}lang'.format(XML_NS), lang) elt.text = i18n[lang] return indexfield_elt
# =============================================================================
[docs]class DBIndexfieldChoice(DBDeclarativeClass): """SQLAlchemy-powered index field choices class (one-to-many).""" __tablename__ = 'wrh_indexfields_choices' __table_args__ = {'mysql_engine': 'InnoDB'} indexfield_id = Column( String(ID_LEN), ForeignKey( 'wrh_indexfields.indexfield_id', ondelete='CASCADE'), primary_key=True) value = Column(String(LABEL_LEN), primary_key=True) i18n_label = Column(Text(), nullable=False) # -------------------------------------------------------------------------
[docs] def label(self, request): """Return a translated label. :type request: pyramid.request.Request :param request: Current request. :rtype: str """ return translate_field(request, loads(self.i18n_label))