How to use the followthemoney.model.get_qname function in followthemoney

To help you get started, we’ve selected a few followthemoney examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github alephdata / followthemoney / tests / graph / test_statement.py View on Github external
def test_base(self):
        prop = model.get_qname('Thing:name')
        node = Node(registry.entity, 'banana')
        stmt = Statement(node, prop, "Theodore Böln")
        assert stmt.subject == node

        value = stmt.to_tuple()
        other = stmt.from_tuple(model, value)
        assert other == stmt, (stmt, other)
        assert hash(other) == hash(stmt)
        assert repr(other) == repr(stmt)
github alephdata / followthemoney / tests / graph / test_statement.py View on Github external
def test_graph(self):
        g = DiGraph()
        proxy = model.get_proxy(ENTITY)
        node = proxy.node
        self.assertEqual(str(node), node.id)
        for stmt in proxy.statements:
            stmt.to_digraph(g)
        self.assertEqual(g.number_of_edges(), 8)
        self.assertEqual(g.number_of_nodes(), 9)
        self.assertIn(node.id, g.nodes)

        prop = model.get_qname('Thing:name')
        stmt = Statement(Node(registry.name, 'Bob'), prop, proxy.id,
                         inverted=True)
        stmt.to_digraph(g)
        self.assertEqual(g.number_of_edges(), 9)

        stmt = Statement(node, prop, 'Blub', weight=0)
        stmt.to_digraph(g)
        self.assertEqual(g.number_of_edges(), 9)

        prop = model.get_qname('Thing:summary')
        stmt = Statement(node, prop, 'This is a text')
        stmt.to_digraph(g)
        self.assertEqual(g.number_of_edges(), 9)
github alephdata / followthemoney / tests / test_graph.py View on Github external
node = Node(registry.entity, "passpoat")
        prop = model.get_qname("Passport:holder")
        adj = list(graph.get_adjacent(node, prop))
        assert len(adj) == 1, adj
        assert adj[0].source_prop == prop, adj[0].source_prop
        assert adj[0].target_prop == prop.reverse, adj[0].target_prop

        node = Node(registry.entity, "jodie")
        prop = model.get_qname("Person:familyPerson")
        adj = list(graph.get_adjacent(node, prop))
        assert len(adj) == 1, adj
        assert adj[0].source_prop == prop, adj[0].source_prop

        node = Node(registry.entity, "ralph")
        prop = model.get_qname("Person:familyRelative")
        adj2 = list(graph.get_adjacent(node, prop))
        assert len(adj2) == 1, adj2
        assert adj2[0].target_prop == prop, adj2[0].target_prop

        assert adj[0] == adj2[0], (adj[0], adj2[0])
        assert adj[0].id in repr(adj[0]), repr(adj[0])
github alephdata / followthemoney / tests / test_graph.py View on Github external
graph.add(model.get_proxy(ENTITY, cleaned=False))
        graph.add(model.get_proxy(ENTITY2, cleaned=False))
        graph.add(model.get_proxy(REL, cleaned=False))
        graph.add(model.get_proxy(PASS, cleaned=False))
        node = Node(registry.entity, "jodie")
        adj = list(graph.get_adjacent(node))
        assert len(adj) == 3, adj
        node = Node(registry.entity, "ralph")
        adj = list(graph.get_adjacent(node))
        assert len(adj) == 7, adj
        node = Node(registry.entity, "passpoat")
        adj = list(graph.get_adjacent(node))
        assert len(adj) == 2, adj

        node = Node(registry.entity, "passpoat")
        prop = model.get_qname("Passport:holder")
        adj = list(graph.get_adjacent(node, prop))
        assert len(adj) == 1, adj
        assert adj[0].source_prop == prop, adj[0].source_prop
        assert adj[0].target_prop == prop.reverse, adj[0].target_prop

        node = Node(registry.entity, "jodie")
        prop = model.get_qname("Person:familyPerson")
        adj = list(graph.get_adjacent(node, prop))
        assert len(adj) == 1, adj
        assert adj[0].source_prop == prop, adj[0].source_prop

        node = Node(registry.entity, "ralph")
        prop = model.get_qname("Person:familyRelative")
        adj2 = list(graph.get_adjacent(node, prop))
        assert len(adj2) == 1, adj2
        assert adj2[0].target_prop == prop, adj2[0].target_prop
github alephdata / followthemoney / tests / graph / test_statement.py View on Github external
def test_invert(self):
        prop = model.get_qname('Thing:name')
        node = Node(registry.entity, 'banana')
        stmt = Statement(node, prop, "Theodore")
        assert not stmt.inverted
        inv = stmt.invert()
        assert inv.inverted
        assert inv.rdf() is None

        banana = Node(registry.entity, 'banana')
        peach = Node(registry.entity, 'peach')
        prop = model.get_qname('Thing:sameAs')
        stmt = Statement(banana, prop, peach.value)
        inv = stmt.invert()
        assert inv.subject == peach
        assert inv.value_node == banana
        assert inv.prop == stmt.prop
github alephdata / followthemoney / tests / test_proxy.py View on Github external
def test_base_functions(self):
        proxy = EntityProxy.from_dict(model, ENTITY)
        assert "test" in repr(proxy), repr(proxy)
        assert hash(proxy) == hash(proxy.id)
        assert proxy.get("name") == ["Ralph Tester"]
        assert proxy.first("name") == "Ralph Tester"
        prop = model.get_qname("LegalEntity:name")
        assert proxy.get(prop) == ["Ralph Tester"]
        assert proxy.caption == "Ralph Tester"
        assert str(proxy) == "Ralph Tester"

        name = "Ralph the Great"
        proxy.add("name", name)
        assert len(proxy.get("name")) == 2
        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")
github alephdata / followthemoney / experiment.py View on Github external
def unpack(cls, ref, packed):
        qname, qualifier, value = packed.split('>', 2)
        prop = model.get_qname(qname)
        # TODO: parse qualifier
        return cls(ref, prop, value)
github alephdata / aleph / aleph / logic / graph / expand.py View on Github external
def iter_value_entities(type_, value):
    value = stringify(value)
    if type_.group is None or value is None:
        return
    key = cache.object_key(type(type_), value)
    degree_key = cache.object_key(type(type_), value, 'deg1')
    degree = cache.get(degree_key)
    if degree is not None:
        for item in cache.kv.sscan_iter(key):
            qname, entity_id = item.decode('utf-8').split('@', 1)
            prop = model.get_qname(qname)
            yield entity_id, prop
    else:
        degree = 0
        pipe = cache.kv.pipeline()
        for entity_id, prop in _iter_value_entities(type_, value):
            yield entity_id, prop
            item = '@'.join((prop.qname, entity_id))
            pipe.sadd(key, item)
            degree += 1
        pipe.set(degree_key, degree, ex=cache.EXPIRE)
        pipe.execute()
github alephdata / aleph / aleph / logic / entities.py View on Github external
facets = []
    for prop in model.properties:
        if prop.type != registry.entity:
            continue
        if not schema.is_a(prop.range):
            continue

        index = entities_read_index(prop.schema)
        field = 'properties.%s' % prop.name
        value = entity.get('id')
        facets.append((index, prop.qname, group, field, value))

    res = _filters_faceted_query(authz, facets)
    for (qname, total) in res.items():
        if total > 0:
            yield (model.get_qname(qname), total)