Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def clean(self, text, **kwargs):
entity_id = get_entity_id(text)
if entity_id is None:
return
entity_id = str(entity_id)
if self.ID_RE.match(entity_id) is not None:
return entity_id
def save(cls, session, subject, candidate, score=None,
judgement=None, priority=None):
obj = cls.by_id(session, subject, candidate)
if obj is None:
obj = cls()
obj.id = cls.make_id(subject, candidate)
obj.subject, _ = Namespace.parse(get_entity_id(subject))
obj.candidate, _ = Namespace.parse(get_entity_id(candidate))
priority = priority or DEFAULT_PRIORITY
if score is not None:
obj.score = score
obj.priority = score * priority
if judgement is not None:
obj.judgement = judgement
obj.updated_at = now()
session.add(obj)
return obj
def channel_tag(obj, clazz=None):
clazz = clazz or type(obj)
if clazz == str:
return obj
obj = get_entity_id(obj)
if obj is not None:
return '%s:%s' % (clazz.__name__, obj)
def _normalize_data(data):
"""Turn entities in properties into entity ids"""
entities = data['layout']['entities']
for obj in entities:
schema = model.get(obj.get('schema'))
if schema is None:
raise InvalidData("Invalid schema %s" % obj.get('schema'))
properties = obj.get('properties', {})
for prop in schema.properties.values():
if prop.type == registry.entity:
values = ensure_list(properties.get(prop.name))
if values:
properties[prop.name] = []
for value in values:
entity_id = get_entity_id(value)
properties[prop.name].append(entity_id)
return data
def publish(event, actor_id=None, params=None, channels=None):
""" Publish a notification to the given channels, while storing
the parameters and initiating actor for the event. """
assert isinstance(event, Event), event
params = params or {}
outparams = {}
channels = [channel_tag(c) for c in ensure_list(channels)]
for name, clazz in event.params.items():
obj = params.get(name)
outparams[name] = get_entity_id(obj)
Notification.publish(event,
actor_id=actor_id,
params=outparams,
channels=channels)
db.session.flush()
def map(self, proxy, record, entities):
if self.entity is not None:
entity = entities.get(self.entity)
if entity is not None:
proxy.add(self.prop, get_entity_id(entity))
inline_names(proxy, entity)
# clean the values returned by the query, or by using literals, or
# formats.
values = []
for value in self.record_values(record):
value = self.type.clean(value, proxy=proxy, **self.data)
if value is not None:
values.append(value)
if self.join is not None:
values = [self.join.join(values)]
if self.split is not None:
splote = []
for value in values:
def make_id(cls, subject, candidate):
subject, _ = Namespace.parse(get_entity_id(subject))
candidate, _ = Namespace.parse(get_entity_id(candidate))
return '.'.join((subject, candidate))
def __init__(self, model, data):
self.model = model
self._data = data
# Support output from Aleph's linkage API (profile_id):
self.id = data.get("canonical_id", data.get("profile_id"))
self.id = self.id or get_entity_id(data.get("canonical"))
self._canonical = None
self.entity_id = data.get("entity_id")
self.entity_id = self.entity_id or get_entity_id(data.get("entity"))
self._entity = None
self.decision = data.get("decision")
self._score = data.get("score", None)
def apply(self, proxy):
"""Rewrite an entity proxy so all IDs mentioned are limited to
the namespace.
An exception is made for sameAs declarations."""
signed = proxy.clone()
signed.id = self.sign(proxy.id)
for prop in proxy.iterprops():
if prop.type != registry.entity:
continue
for value in signed.pop(prop):
value = get_entity_id(value)
signed.add(prop, self.sign(value))
# linked.add('sameAs', proxy.id, quiet=True)
signed.remove("sameAs", signed.id)
return signed