cubicweb: comparison entity.py

equal deleted inserted replaced

-:717dea3362c0
+:716f0742ee7f
 else:
 raise ValueError(tag)
 except ImportError:
 AutomaticEntityForm = None
 def dispatch_rtags(*args):
 pass
 def _get_etype(bases, classdict):
 try:
 except KeyError:
 for base in bases:
 etype = getattr(base, 'id', None)
 if etype and etype != 'Any':
 return etype
 def _get_defs(attr, name, bases, classdict):
 try:
 yield name, classdict.pop(attr)
 except KeyError:
 for base in bases:
 value = getattr(base, attr)
 delattr(base, attr)
 yield base.__name__, value
 except AttributeError:
 continue
 class metaentity(type):
 """this metaclass sets the relation tags on the entity class
 and deals with the `widgets` attribute
 """
 def __new__(mcs, name, bases, classdict):
 class Entity(AppRsetObject, dict):
 """an entity instance has e_schema automagically set on
 the class and instances has access to their issuing cursor.
 A property is set for each attribute and relation on each entity's type
 class. Becare that among attributes, 'eid' is *NEITHER* stored in the
 dict containment (which acts as a cache for other attributes dynamically
 fetched)
 :type rest_var: str
 :cvar rest_var: indicates which attribute should be used to build REST urls
 If None is specified, the first non-meta attribute will
 be used
 :type skip_copy_for: list
 :cvar skip_copy_for: a list of relations that should be skipped when copying
 this kind of entity. Note that some relations such
 as composite relations or relations that have '?1' as object
 cardinality
 """
 __metaclass__ = metaentity
 __registry__ = 'etypes'
 __select__ = yes()
 # class attributes that must be set in class definition
 id = None
 rest_attr = None
 fetch_attrs = None
 skip_copy_for = ()
 # class attributes set automatically at registration time
 e_schema = None
 @classmethod
 def registered(cls, registry):
 """build class using descriptor at registration time"""
 assert cls.id is not None
 super(Entity, cls).registered(registry)
 if cls.id != 'Any':
 cls.__initialize__()
 return cls
 MODE_TAGS = set(('link', 'create'))
 CATEGORY_TAGS = set(('primary', 'secondary', 'generic', 'generated')) # , 'metadata'))
 @classmethod
 def __initialize__(cls):
 """initialize a specific entity class by adding descriptors to access
 attr = 'reverse_%s' % rschema.type
 setattr(cls, attr, ObjectRelation(rschema))
 if mixins:
 cls.__bases__ = tuple(mixins + [p for p in cls.__bases__ if not p is object])
 cls.debug('plugged %s mixins on %s', mixins, etype)
 @classmethod
 def fetch_rql(cls, user, restriction=None, fetchattrs=None, mainvar='X',
 settype=True, ordermethod='fetch_order'):
 """return a rql to fetch all entities of the class type"""
 restrictions = restriction or []
 rql = 'Any %s' % ','.join(selection)
 if orderby:
 rql +=  ' ORDERBY %s' % ','.join(orderby)
 rql += ' WHERE %s' % ', '.join(restrictions)
 return rql
 @classmethod
 def _fetch_restrictions(cls, mainvar, varmaker, fetchattrs,
 selection, orderby, restrictions, user,
 ordermethod='fetch_order', visited=None):
 eschema = cls.e_schema
 if rset is not None:
 self.eid = rset[row][col]
 else:
 self.eid = None
 self._is_saved = True
 def __repr__(self):
 return '<Entity %s %s %s at %s>' % (
 self.e_schema, self.eid, self.keys(), id(self))
 def __nonzero__(self):
 def pre_add_hook(self):
 """hook called by the repository before doing anything to add the entity
 (before_add entity hooks have not been called yet). This give the
 occasion to do weird stuff such as autocast (File -> Image for instance).
 This method must return the actual entity to be added.
 """
 return self
 def set_eid(self, eid):
 self.eid = self['eid'] = eid
 def has_eid(self):
 """return True if the entity has an attributed eid (False
 has an eid attributed though it's not saved (eg during before_add_entity
 hooks). You can use this method to ensure the entity has an eid *and* is
 saved in its source.
 """
 return self.has_eid() and self._is_saved
 @cached
 def metainformation(self):
 res = dict(zip(('type', 'source', 'extid'), self.req.describe(self.eid)))
 res['source'] = self.req.source_defs()[res['source']]
 return res
 def check_perm(self, action):
 self.e_schema.check_perm(self.req, action, self.eid)
 def has_perm(self, action):
 return self.e_schema.has_perm(self.req, action, self.eid)
 def view(self, vid, __registry='views', **kwargs):
 """shortcut to apply a view on this entity"""
 return self.vreg.render(__registry, vid, self.req, rset=self.rset,
 row=self.row, col=self.col, **kwargs)
 def mtc_transform(self, data, format, target_format, encoding,
 _engine=ENGINE):
 trdata = TransformData(data, format, encoding, appobject=self)
 data = _engine.convert(trdata, target_format).decode()
 if format == 'text/html':
 data = soup2xhtml(data, self.req.encoding)
 return data
 # entity cloning ##########################################################
 def copy_relations(self, ceid):
 """copy relations of the object with the given eid on this object
 def as_rset(self):
 """returns a resultset containing `self` information"""
 rset = ResultSet([(self.eid,)], 'Any X WHERE X eid %(x)s',
 {'x': self.eid}, [(self.id,)])
 return self.req.decorate_rset(rset)
 def to_complete_relations(self):
 """by default complete final relations to when calling .complete()"""
 for rschema in self.e_schema.subject_relations():
 if rschema.is_final():
 continue
 matching_groups = self.req.user.matching_groups
 if matching_groups(rschema.get_groups('read')) and \
 all(matching_groups(es.get_groups('read'))
 for es in rschema.objects(self.e_schema)):
 yield rschema, 'subject'
 def to_complete_attributes(self, skip_bytes=True):
 for rschema, attrschema in self.e_schema.attribute_definitions():
 # skip binary data by default
 if skip_bytes and attrschema.type == 'Bytes':
 continue
 if not self.req.user.matching_groups(rschema.get_groups('read')) \
 or attrschema.type == 'Password':
 self[attr] = None
 continue
 yield attr
 def complete(self, attributes=None, skip_bytes=True):
 """complete this entity by adding missing attributes (i.e. query the
 repository to fill the entity)
 :type skip_bytes: bool
 rrset = ResultSet([], rql, {'x': self.eid})
 self.req.decorate_rset(rrset)
 else:
 rrset = self.req.eid_rset(value)
 self.set_related_cache(rtype, x, rrset)
 def get_value(self, name):
 """get value for the attribute relation <name>, query the repository
 to get the value if necessary.
 :type name: str
 self[name] = value = None
 return value
 def related(self, rtype, role='subject', limit=None, entities=False):
 """returns a resultset of related entities
 :param role: is the role played by 'self' in the relation ('subject' or 'object')
 :param limit: resultset's maximum size
 :param entities: if True, the entites are returned; if False, a result set is returned
 """
 try:
 rql.split(' WHERE ', 1)[1])
 elif not ' ORDERBY ' in rql:
 args = tuple(rql.split(' WHERE ', 1))
 rql = '%s ORDERBY Z DESC WHERE X modification_date Z, %s' % args
 return rql
 # generic vocabulary methods ##############################################
 @obsolete('see new form api')
 def vocabulary(self, rtype, role='subject', limit=None):
 """vocabulary functions must return a list of couples
 (label, eid) that will typically be used to fill the
 edition view's combobox.
 If `eid` is None in one of these couples, it should be
 interpreted as a separator in case vocabulary results are grouped
 """
 from cubicweb.web.form import EntityFieldsForm
 from logilab.common.testlib import mock_object
 form = EntityFieldsForm(self.req, entity=self)
 field = mock_object(name=rtype, role=role)
 return form.form_field_vocabulary(field, limit)
 def unrelated_rql(self, rtype, targettype, role, ordermethod=None,
 vocabconstraints=True):
 """build a rql to fetch `targettype` entities unrelated to this entity
 using (rtype, role) relation
 """
 # ensure we have an order defined
 if not ' ORDERBY ' in rql:
 before, after = rql.split(' WHERE ', 1)
 rql = '%s ORDERBY %s WHERE %s' % (before, searchedvar, after)
 return rql
 def unrelated(self, rtype, targettype, role='subject', limit=None,
 ordermethod=None):
 """return a result set of target type objects that may be related
 by a given relation, with self as subject or object
 """
 before, after = rql.split(' WHERE ', 1)
 rql = '%s LIMIT %s WHERE %s' % (before, limit, after)
 if self.has_eid():
 return self.req.execute(rql, {'x': self.eid})
 return self.req.execute(rql)
 # relations cache handling ################################################
 def relation_cached(self, rtype, role):
 """return true if the given relation is already cached on the instance
 """
 return '%s_%s' % (rtype, role) in self._related_cache
 def related_cache(self, rtype, role, entities=True, limit=None):
 """return values for the given relation if it's cached on the instance,
 else raise `KeyError`
 """
 res = self._related_cache['%s_%s' % (rtype, role)][entities]
 if entities:
 res = res[:limit]
 else:
 res = res.limit(limit)
 return res
 def set_related_cache(self, rtype, role, rset, col=0):
 """set cached values for the given relation"""
 if rset:
 related = list(rset.entities(col))
 rschema = self.schema.rschema(rtype)
 for rentity in related:
 rentity._related_cache['%s_%s' % (rtype, target)] = (self.as_rset(), [self])
 else:
 related = []
 self._related_cache['%s_%s' % (rtype, role)] = (rset, related)
 def clear_related_cache(self, rtype=None, role=None):
 """clear cached values for the given relation or the entire cache if
 no relation is given
 """
 if rtype is None:
 self._related_cache = {}
 else:
 assert role
 self._related_cache.pop('%s_%s' % (rtype, role), None)
 # raw edition utilities ###################################################
 def set_attributes(self, **kwargs):
 assert kwargs
 relations = []
 for key in kwargs:
 relations.append('X %s %%(%s)s' % (key, key))
 self.update(kwargs)
 # and now update the database
 kwargs['x'] = self.eid
 self.req.execute('SET %s WHERE X eid %%(x)s' % ','.join(relations),
 kwargs, 'x')
 def delete(self):
 assert self.has_eid(), self.eid
 self.req.execute('DELETE %s X WHERE X eid %%(x)s' % self.e_schema,
 {'x': self.eid})
 # server side utilities ###################################################
 def set_defaults(self):
 """set default values according to the schema"""
 self._default_set = set()
 for attr, value in self.e_schema.defaults():
 if not self.has_key(attr):
 yielded = True
 if not yielded:
 yield self
 else:
 yield self
 def get_words(self):
 """used by the full text indexer to get words to index
 this method should only be used on the repository side since it depends
 on the indexer package
 :rtype: list
 :return: the list of indexable word of this entity
 """
 from indexer.query_objects import tokenize
 words = []
 self.exception("can't add value of %s to text index for entity %s",
 rschema, self.eid)
 continue
 if value:
 words += tokenize(value)
 for rschema, role in self.e_schema.fulltext_relations():
 if role == 'subject':
 for entity in getattr(self, rschema.type):
 words += entity.get_words()
 else: # if role == 'object':
 def __get__(self, eobj, eclass):
 if eobj is None:
 raise AttributeError('%s cannot be only be accessed from instances'
 % self._rtype)
 return eobj.related(self._rtype, self._role, entities=True)
 def __set__(self, eobj, value):
 raise NotImplementedError
 class SubjectRelation(Relation):
 """descriptor that controls schema relation access"""
 _role = 'subject'
 class ObjectRelation(Relation):
 """descriptor that controls schema relation access"""
 _role = 'object'
 from logging import getLogger

branch	tls-sprint
changeset 1474	716f0742ee7f
parent 1471	c889c3bcf5ec
child 1498	2c6eec0b46b9