Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_make_id(self):
proxy = model.make_entity("Thing")
assert not proxy.make_id(None)
assert proxy.make_id("banana")
assert proxy.make_id("banana") == proxy.make_id("banana")
ff = proxy.make_id(44)
assert ff is not None
proxy = model.make_entity("Thing", key_prefix="foo")
assert proxy.make_id(44)
assert proxy.make_id(44) != ff, proxy.make_id(44)
proxy.add("name", None)
assert len(proxy.get("name")) == 2
proxy.add("name", "")
assert len(proxy.get("name")) == 2
proxy.add("name", [""])
assert len(proxy.get("name")) == 2
proxy.add("name", {"name": "banana"})
assert len(proxy.get("name")) == 3, proxy.get("name")
assert name in proxy.get("name")
assert name in proxy.names, proxy.names
with assert_raises(InvalidData):
proxy.add("banana", "yellow")
proxy.add("banana", "yellow", quiet=True)
mem = model.make_entity("Membership")
mem.id = "foo"
with assert_raises(InvalidData):
proxy.add("directorshipDirector", mem)
with assert_raises(InvalidData):
proxy.add("sameAs", proxy)
with assert_raises(InvalidData):
proxy.get("banana")
assert [] == proxy.get("banana", quiet=True)
with assert_raises(InvalidData):
proxy.first("banana")
assert proxy.first("banana", quiet=True) is None
assert len(proxy.get("nationality")) == 0
def make_entity(self, schema):
return model.make_entity(schema, key_prefix=self.enricher.key_prefix)
def convert_release(release):
for party in release.pop('parties', []):
yield convert_party(party)
buyer = release.pop('buyer', {})
authority = model.make_entity('LegalEntity')
authority.make_id(buyer.pop('id', None))
authority.add('name', buyer.pop('name', None))
yield authority
tender = release.pop('tender', {})
contract = model.make_entity('Contract')
contract.make_id(release.pop('id', None))
contract.add('authority', authority)
contract.add('name', tender.pop('title', None))
if not contract.has('name'):
contract.add('name', tender.get('id', None))
contract.add('description', tender.pop('description', None))
contract.add('procedureNumber', tender.pop('id', None))
contract.add('type', tender.pop('mainProcurementCategory', None))
value = tender.pop('value', {})
if not contract.has('name'):
contract.add('name', tender.get('id', None))
contract.add('description', tender.pop('description', None))
contract.add('procedureNumber', tender.pop('id', None))
contract.add('type', tender.pop('mainProcurementCategory', None))
value = tender.pop('value', {})
contract.add('amount', value.pop('amount', None))
contract.add('currency', value.pop('currency', None))
# pprint(tender)
yield contract
# contract.add('modifiedAt', published_date)
lots = tender.pop('lots', [])
for award in release.pop('awards', []):
ca = model.make_entity('ContractAward')
ca.make_id(contract.id, award.pop('id', None))
ca.add('contract', contract)
ca.add('date', clean_date(award.pop('date', None)))
value = award.pop('value', {})
ca.add('amount', value.pop('amount', None))
ca.add('currency', value.pop('currency', None))
reason = tender.get('procurementMethodDetails', None)
ca.add('decisionReason', reason)
for document in award.pop('documents', []):
ca.add('sourceUrl', document.get('url'))
for item in award.pop('items', []):
classification = item.pop('classification', {})
ca.add('cpvCode', classification.get('url'))
def company_entity(self, data, entity=None):
if "company" in data:
data = ensure_dict(data.get("company", data))
if entity is None:
entity = model.make_entity("Company")
entity.make_id(data.get("opencorporates_url"))
entity.add("name", data.get("name"))
address = ensure_dict(data.get("registered_address"))
entity.add("country", address.get("country"))
entity.add("jurisdiction", data.get("jurisdiction_code"))
entity.add("alias", data.get("alternative_names"))
entity.add("address", data.get("registered_address_in_full"))
entity.add("sourceUrl", data.get("registry_url"))
entity.add("legalForm", data.get("company_type"))
entity.add("incorporationDate", data.get("incorporation_date"))
entity.add("dissolutionDate", data.get("dissolution_date"))
entity.add("status", data.get("current_status"))
entity.add("registrationNumber", data.get("company_number"))
entity.add("opencorporatesUrl", data.get("opencorporates_url"))
source = data.get("source", {})
entity.add("publisher", source.get("publisher"))
for document in award.pop('documents', []):
ca.add('sourceUrl', document.get('url'))
for item in award.pop('items', []):
classification = item.pop('classification', {})
ca.add('cpvCode', classification.get('url'))
related_lots = award.pop('relatedLots', [])
for lot in lots:
if lot.get('id') in related_lots:
ca.add('role', lot.get('title'))
ca.add('summary', lot.get('description'))
for supplier in award.pop('suppliers', []):
entity = model.make_entity('LegalEntity')
entity.make_id(supplier.pop('id', None))
entity.add('name', supplier.pop('name', None))
ca.add('supplier', entity)
yield entity
# pprint(award)
yield ca