# copyright 2016 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
# contact http://www.logilab.fr -- mailto:contact@logilab.fr
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU Lesser General Public License as published by the Free
# Software Foundation, either version 2.1 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""Mappers with "jsonschema.collection" regid."""
from six import text_type
from logilab.common.registry import yes
from cubicweb import (
neg_role,
Unauthorized,
)
from cubicweb.predicates import (
match_kwargs,
)
from cubicweb_jsonschema import (
CREATION_ROLE,
)
from .base import (
JSONSchemaMapper,
JSONSchemaDeserializer,
JSONSchemaSerializer,
add_descriptive_metadata,
add_links,
object_schema,
)
from .predicates import (
_for_workflowable_entity,
yams_component_target,
)
__all__ = [
'EntityCollectionMapper',
'RelatedCollectionMapper',
'CollectionItemMapper',
]
[docs]@JSONSchemaDeserializer.register
@JSONSchemaSerializer.register
class EntityCollectionMapper(JSONSchemaMapper):
"""Mapper for a collection of entities."""
__regid__ = 'jsonschema.collection'
__select__ = match_kwargs('etype')
def __repr__(self):
return '<{0.__class__.__name__} etype={0.etype}'.format(self)
@property
def etype(self):
return self.cw_extra_kwargs['etype']
@property
def title(self):
"""Title of the collection, plural form of entity type."""
return self._cw._('{}_plural').format(self.etype)
@add_links
def schema_and_definitions(self, schema_role=None):
if schema_role == CREATION_ROLE:
return self._submission_schema_and_definitions()
return self._array_schema(schema_role=schema_role)
def _submission_schema_and_definitions(self):
"""Delegate generation of schema and definitions to the "entity"
mapper corresponding to the entity type in this collection.
"""
mapper = self.select_mapper(
'jsonschema.entity', **self.cw_extra_kwargs)
return mapper.schema_and_definitions(schema_role=CREATION_ROLE)
def _array_schema(self, schema_role=None):
item_mapper = self.select_mapper(
'jsonschema.item', **self.cw_extra_kwargs)
items_schema, items_defs = item_mapper.schema_and_definitions(
schema_role)
schema = {
'type': 'array',
'items': items_schema,
}
return add_descriptive_metadata(schema, self), items_defs
[docs] def links(self, schema_role=None):
"""Yield Link appobjects matching regid and selection context of this
mapper if schema_role is None.
"""
if schema_role is not None:
return
for link in super(EntityCollectionMapper, self).links(
schema_role=schema_role):
yield link
def values(self, instance):
mapper = self.select_mapper(
'jsonschema.entity', **self.cw_extra_kwargs)
return mapper.values(instance)
[docs] def serialize(self, entities):
"""Return a list of collection item representing each entity in
`entities`.
"""
mapper = self.select_mapper('jsonschema.item', **self.cw_extra_kwargs)
return [mapper.serialize(entity) for entity in entities]
class NonInlinedRelatedCollectionMapper(RelatedCollectionMapper):
"""Mapper for a collection of entities through a non-*inlined* relation."""
__select__ = (
match_kwargs('entity', 'rtype', 'role')
& ~yams_component_target()
)
def _submission_schema_and_definitions(self):
"""Return schema and definitions accounting for constraints on
possible targets of `rtype`, `role` relation information for `entity`
bound to this mapper.
"""
entity = self.cw_extra_kwargs['entity']
rschema = self._cw.vreg.schema[self.rtype]
# XXX similar loop in ETypeRelationItemMapper.relation_targets().
ids = []
for target_type in rschema.targets(role=self.role):
try:
rset = entity.unrelated(
self.rtype, target_type, role=self.role)
except Unauthorized:
continue
for target in rset.entities():
ids.append({
'type': 'string',
'enum': [text_type(target.eid)],
'title': target.dc_title(),
})
if not ids:
return False, None
properties = {
'id': {
'oneOf': ids,
},
}
schema = object_schema(properties, required=['id'])
return add_descriptive_metadata(schema, self), None
[docs]@JSONSchemaSerializer.register
class CollectionItemMapper(JSONSchemaMapper):
"""Mapper for an item of a collection."""
__regid__ = 'jsonschema.item'
__select__ = yes()
[docs] @add_links
def schema_and_definitions(self, schema_role=None):
"""Return either a string schema or an object with "type", "id and
"title" properties.
"""
if schema_role == CREATION_ROLE:
raise ValueError('{} is not appropriate for submission role')
schema = {
'type': 'object',
'properties': {
'type': {
'type': 'string',
},
'id': {
'type': 'string',
},
'title': {
'type': 'string',
},
},
}
return add_descriptive_metadata(schema, self), {}
def links(self, schema_role=None, **kwargs):
kwargs['anchor'] = '#'
return super(CollectionItemMapper, self).links(
schema_role=schema_role, **kwargs)
[docs] @staticmethod
def serialize(entity):
"""Return a dictionary with entity represented as a collection item."""
return {
'type': entity.cw_etype.lower(),
'id': text_type(entity.eid),
'title': entity.dc_title(),
}
class TrInfoCollectionMapper(EntityCollectionMapper):
"""Mapper for a collection of TrInfo associated with an entity."""
__select__ = (
match_kwargs({'etype': 'TrInfo'})
& _for_workflowable_entity()
)
def values(self, *args):
# This is handled by the view.
raise NotImplementedError()
def serialize(self):
entity = self.cw_extra_kwargs['for_entity']
wfobj = entity.cw_adapt_to('IWorkflowable')
entities = wfobj.workflow_history
return super(TrInfoCollectionMapper, self).serialize(entities)