Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def description(self):
return gettext(self._description)
def validate(self, data):
"""Validate a dataset against the given schema.
This will also drop keys which are not present as properties.
"""
errors = {}
properties = ensure_dict(data.get("properties"))
for name, prop in self.properties.items():
values = ensure_list(properties.get(name))
error = prop.validate(values)
if error is None and not len(values):
if prop.name in self.required:
error = gettext("Required")
if error is not None:
errors[name] = error
if len(errors):
msg = gettext("Entity validation failed")
raise InvalidData(msg, errors={"properties": errors})
def label(self):
return gettext(self._label)
def __init__(self, model, data, key_prefix=None, cleaned=True):
data = dict(data)
properties = data.pop("properties", {})
if not cleaned:
properties = ensure_dict(properties)
self.schema = model.get(data.pop("schema", None))
if self.schema is None:
raise InvalidData(gettext("No schema for entity."))
self.key_prefix = key_prefix
self.id = data.pop("id", None)
if not cleaned:
self.id = sanitize_text(self.id)
self.context = data
self._properties = {}
self._size = 0
for key, value in properties.items():
if key not in self.schema.properties:
continue
if not cleaned:
self.add(key, value, cleaned=cleaned, quiet=True)
else:
self._properties[key] = set(value)
def to_dict(self):
data = {"label": gettext(self.label), "plural": gettext(self.plural)}
if self.group:
data["group"] = self.group
if self.matchable:
data["matchable"] = True
if self.pivot:
data["pivot"] = True
return data
def _locale_names(self, locale):
# extra territories that OCCRP is interested in.
names = {
"zz": gettext("Global"),
"eu": gettext("European Union"),
# Overwrite "Czechia" label:
"cz": gettext("Czech Republic"),
"xk": gettext("Kosovo"),
"yucs": gettext("Yugoslavia"),
"csxx": gettext("Serbia and Montenegro"),
"suhh": gettext("Soviet Union"),
"ge-ab": gettext("Abkhazia"),
"x-so": gettext("South Ossetia"),
"so-som": gettext("Somaliland"),
"gb-wls": gettext("Wales"),
"gb-sct": gettext("Scotland"),
"gb-nir": gettext("Northern Ireland"),
"md-pmr": gettext("Transnistria"),
}
for code, label in locale.territories.items():
code = code.lower()
if code in names:
continue
try:
int(code)
except ValueError:
names[code] = label
return names
def _locale_names(self, locale):
# extra territories that OCCRP is interested in.
names = {
"zz": gettext("Global"),
"eu": gettext("European Union"),
# Overwrite "Czechia" label:
"cz": gettext("Czech Republic"),
"xk": gettext("Kosovo"),
"yucs": gettext("Yugoslavia"),
"csxx": gettext("Serbia and Montenegro"),
"suhh": gettext("Soviet Union"),
"ge-ab": gettext("Abkhazia"),
"x-so": gettext("South Ossetia"),
"so-som": gettext("Somaliland"),
"gb-wls": gettext("Wales"),
"gb-sct": gettext("Scotland"),
"gb-nir": gettext("Northern Ireland"),
"md-pmr": gettext("Transnistria"),
}
for code, label in locale.territories.items():
code = code.lower()
if code in names:
continue
try:
int(code)
except ValueError:
names[code] = label
return names
prop = self.schema.properties[prop_name]
# Don't allow setting the reverse properties:
if prop.stub:
if quiet:
return
msg = gettext("Stub property (%s): %s")
raise InvalidData(msg % (self.schema, prop))
for value in value_list(values):
if not cleaned:
value = prop.type.clean(value, proxy=self, fuzzy=fuzzy)
if value is None:
continue
if prop.type == registry.entity and value == self.id:
msg = gettext("Self-relationship (%s): %s")
raise InvalidData(msg % (self.schema, prop))
# Somewhat hacky: limit the maximum size of any particular
# field to avoid overloading upstream aleph/elasticsearch.
value_size = len(value)
if prop.type.max_size is not None:
if self._size + value_size > prop.type.max_size:
# msg = "[%s] too large. Rejecting additional values."
# log.warning(msg, prop.name)
continue
self._size += value_size
if prop_name not in self._properties:
self._properties[prop_name] = set()
self._properties[prop_name].add(value)
def label(self):
return gettext(self._label)
def add(self, prop, values, cleaned=False, quiet=False, fuzzy=False):
"""Add the given value(s) to the property if they are not empty."""
prop_name = self._prop_name(prop, quiet=quiet)
if prop_name is None:
return
prop = self.schema.properties[prop_name]
# Don't allow setting the reverse properties:
if prop.stub:
if quiet:
return
msg = gettext("Stub property (%s): %s")
raise InvalidData(msg % (self.schema, prop))
for value in value_list(values):
if not cleaned:
value = prop.type.clean(value, proxy=self, fuzzy=fuzzy)
if value is None:
continue
if prop.type == registry.entity and value == self.id:
msg = gettext("Self-relationship (%s): %s")
raise InvalidData(msg % (self.schema, prop))
# Somewhat hacky: limit the maximum size of any particular
# field to avoid overloading upstream aleph/elasticsearch.
value_size = len(value)
if prop.type.max_size is not None:
if self._size + value_size > prop.type.max_size: