Skip to content

Stores

A store reads and writes statement-based FollowTheMoney (FtM) entities, and assembles them into their merged, canonical form on retrieval.

Stores are where deduplication decisions meet the data. When an entity is read from a store, its statements are grouped by canonical ID — as defined by a Linker — so a cluster of merged source records comes back as one entity. The source statements themselves are never rewritten; changing a judgement in the resolver changes what the store returns, not what it contains.

Three classes cooperate:

  • Store — the storage backend, bound to a dataset scope and a linker.
  • Writer — bulk write operations: add entities or individual statements, remove them by entity ID.
  • View — read access over a dataset scope: get an entity by ID, iterate all entities, or traverse inverted relationships (which entities reference this one?).

Choosing a backend

Use MemoryStore for datasets that fit in memory — this is what the nk command line uses when it reads entities from a file, via load_entity_file_store. Use SQLStore to persist statements to SQLite or PostgreSQL. Two further backends, LevelStore (LevelDB) and RedisStore, live in nomenklatura.store.level and nomenklatura.store.redis_ and require the optional plyvel and redis dependencies.

from pathlib import Path
from nomenklatura import Resolver
from nomenklatura.db import make_session
from nomenklatura.store import load_entity_file_store

with make_session() as session:
    resolver = Resolver(session, create=True)
    store = load_entity_file_store(Path("entities.ftm.json"), resolver)
    view = store.default_view()
    for entity in view.entities():
        print(entity.caption)

Interface

nomenklatura.store.Store

Bases: Generic[DS, SE]

A data storage and retrieval mechanism for statement-based entity data. Essentially, this is a triple store which can be implemented using various backends.

Source code in nomenklatura/store/base.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class Store(Generic[DS, SE]):
    """A data storage and retrieval mechanism for statement-based entity data.
    Essentially, this is a triple store which can be implemented using various
    backends."""

    def __init__(self, dataset: DS, linker: Linker[SE]):
        self.dataset = dataset
        self.linker = linker
        self.entity_class = cast("type[SE]", StatementEntity)

    def writer(self) -> "Writer[DS, SE]":
        raise NotImplementedError

    def view(self, scope: DS, external: bool = False) -> "View[DS, SE]":
        raise NotImplementedError

    def default_view(self, external: bool = False) -> "View[DS, SE]":
        return self.view(self.dataset, external=external)

    def assemble(self, statements: list[Statement]) -> SE | None:
        if not len(statements):
            return None
        canonicals: list[Statement] = []
        for stmt in statements:
            if get_prop_type(stmt.schema, stmt.prop) == registry.entity.name:
                canonical = self.linker.get_canonical(stmt._value)
                if canonical != stmt._value:
                    stmt = stmt.clone(
                        value=canonical,
                        original_value=stmt.original_value or stmt._value,
                    )
            canonicals.append(stmt)
        entity = self.entity_class.from_statements(self.dataset, canonicals)
        if entity.id is not None:
            entity.extra_referents.update(self.linker.get_referents(entity.id))
        return entity

    def update(self, id: str) -> None:
        canonical_id = self.linker.get_canonical(id)
        with self.writer() as writer:
            for referent in self.linker.get_referents(canonical_id):
                for stmt in writer.pop(referent):
                    stmt.canonical_id = canonical_id
                    writer.add_statement(stmt)

    def close(self) -> None:
        pass

    def __repr__(self) -> str:
        return f"<{type(self).__name__}({self.dataset.name!r})>"

nomenklatura.store.Writer

Bases: Generic[DS, SE]

Bulk writing operations.

Source code in nomenklatura/store/base.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class Writer(Generic[DS, SE]):
    """Bulk writing operations."""

    def __init__(self, store: Store[DS, SE]):
        self.store = store

    def add_statement(self, stmt: Statement) -> None:
        raise NotImplementedError

    def add_entity(self, entity: SE) -> None:
        for stmt in entity.statements:
            self.add_statement(stmt)

    def pop(self, entity_id: str) -> list[Statement]:
        raise NotImplementedError

    def flush(self) -> None:
        pass

    def close(self) -> None:
        self.store.close()

    def __enter__(self) -> "Writer[DS, SE]":
        return self

    def __exit__(
        self,
        type: type[BaseException] | None,
        value: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        self.flush()

    def __repr__(self) -> str:
        return f"<{type(self).__name__}({self.store!r})>"

nomenklatura.store.View

Bases: Generic[DS, SE]

Read access to the entities in a store, scoped to a dataset.

Entities come back in their merged, canonical form. Use get_entity() for a lookup by ID, entities() to stream the whole scope, and get_adjacent() to traverse relationships in both directions.

Statements marked external (enrichment candidates not yet accepted into the dataset) are excluded from all reads unless the view is constructed with external=True; an entity backed only by external statements is absent from an external=False view.

Source code in nomenklatura/store/base.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
class View(Generic[DS, SE]):
    """Read access to the entities in a store, scoped to a dataset.

    Entities come back in their merged, canonical form. Use `get_entity()` for
    a lookup by ID, `entities()` to stream the whole scope, and `get_adjacent()`
    to traverse relationships in both directions.

    Statements marked `external` (enrichment candidates not yet accepted into
    the dataset) are excluded from all reads unless the view is constructed
    with `external=True`; an entity backed only by external statements is
    absent from an `external=False` view."""

    def __init__(self, store: Store[DS, SE], scope: DS, external: bool = False):
        self.store = store
        self.scope = scope
        self.dataset_names = scope.leaf_names
        self.external = external

    def has_entity(self, id: str) -> bool:
        raise NotImplementedError

    def get_entity(self, id: str) -> SE | None:
        raise NotImplementedError

    def get_entities(self, ids: Iterable[str]) -> Generator[SE, None, None]:
        """Fetch several entities in one go.

        Bulk readers (e.g. the xref scoring loop) should prefer this over
        repeated `get_entity()` calls so that stores backed by query engines
        can serve the batch from a single query."""
        for id in ids:
            entity = self.get_entity(id)
            if entity is not None:
                yield entity

    def get_inverted(self, id: str) -> Generator[tuple[Property, SE], None, None]:
        raise NotImplementedError

    def get_adjacent(
        self, entity: SE, inverted: bool = True
    ) -> Generator[tuple[Property, SE], None, None]:
        for prop, value in entity.itervalues():
            if prop.type == registry.entity:
                child = self.get_entity(value)
                if child is not None:
                    yield prop, child

        if inverted and entity.id is not None:
            for prop, adjacent in self.get_inverted(entity.id):
                yield prop, adjacent

    def entities(
        self,
        include_schemata: list[Schema] | None = None,
        prefetch_nested: bool = False,
    ) -> Generator[SE, None, None]:
        """Iterate over all entities in the view.

        If `include_schemata` is provided, only entities of the provided schemata will be returned.
        Note that `schemata` will not be expanded via "is_a" relationships.

        With `prefetch_nested`, implementations may bulk-load the adjacency of
        the scanned entities so that `get_entity`, `get_inverted` and
        `get_adjacent` calls made while iterating avoid per-call lookups.
        Point-read stores ignore the flag; results are identical either way."""

        raise NotImplementedError

    def __repr__(self) -> str:
        return f"<{type(self).__name__}({self.scope.name!r})>"

entities(include_schemata=None, prefetch_nested=False)

Iterate over all entities in the view.

If include_schemata is provided, only entities of the provided schemata will be returned. Note that schemata will not be expanded via "is_a" relationships.

With prefetch_nested, implementations may bulk-load the adjacency of the scanned entities so that get_entity, get_inverted and get_adjacent calls made while iterating avoid per-call lookups. Point-read stores ignore the flag; results are identical either way.

Source code in nomenklatura/store/base.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def entities(
    self,
    include_schemata: list[Schema] | None = None,
    prefetch_nested: bool = False,
) -> Generator[SE, None, None]:
    """Iterate over all entities in the view.

    If `include_schemata` is provided, only entities of the provided schemata will be returned.
    Note that `schemata` will not be expanded via "is_a" relationships.

    With `prefetch_nested`, implementations may bulk-load the adjacency of
    the scanned entities so that `get_entity`, `get_inverted` and
    `get_adjacent` calls made while iterating avoid per-call lookups.
    Point-read stores ignore the flag; results are identical either way."""

    raise NotImplementedError

get_entities(ids)

Fetch several entities in one go.

Bulk readers (e.g. the xref scoring loop) should prefer this over repeated get_entity() calls so that stores backed by query engines can serve the batch from a single query.

Source code in nomenklatura/store/base.py 
132
133
134
135
136
137
138
139
140
141
def get_entities(self, ids: Iterable[str]) -> Generator[SE, None, None]:
    """Fetch several entities in one go.

    Bulk readers (e.g. the xref scoring loop) should prefer this over
    repeated `get_entity()` calls so that stores backed by query engines
    can serve the batch from a single query."""
    for id in ids:
        entity = self.get_entity(id)
        if entity is not None:
            yield entity

Implementations

nomenklatura.store.MemoryStore

Bases: Store[DS, SE]

Hold statements in plain dictionaries, with no persistence.

The right choice for datasets that fit into memory, e.g. when processing an entity file on the command line.

Source code in nomenklatura/store/memory.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class MemoryStore(Store[DS, SE]):
    """Hold statements in plain dictionaries, with no persistence.

    The right choice for datasets that fit into memory, e.g. when processing
    an entity file on the command line."""

    def __init__(self, dataset: DS, linker: Linker[SE]):
        super().__init__(dataset, linker)
        self.stmts: dict[str, set[Statement]] = {}
        self.inverted: dict[str, set[str]] = {}
        self.entities: dict[str, set[str]] = {}

    def writer(self) -> Writer[DS, SE]:
        return MemoryWriter(self)

    def view(self, scope: DS, external: bool = False) -> View[DS, SE]:
        return MemoryView(self, scope, external=external)

nomenklatura.store.sql.SQLStore

Bases: Store[DS, SE]

Persist statements to a SQL database (SQLite or PostgreSQL).

Use this when a dataset is too large to hold in memory, or when several processes need to work with the same store.

Source code in nomenklatura/store/sql.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
class SQLStore(Store[DS, SE]):
    """Persist statements to a SQL database (SQLite or PostgreSQL).

    Use this when a dataset is too large to hold in memory, or when several
    processes need to work with the same store."""

    def __init__(
        self,
        dataset: DS,
        linker: Linker[SE],
        uri: str = settings.DB_URL,
    ):
        super().__init__(dataset, linker)
        self._uri = uri
        metadata = get_metadata()
        self.engine: Engine = get_engine(uri)
        self.table = make_statement_table(metadata)
        metadata.create_all(self.engine, tables=[self.table], checkfirst=True)

    def writer(self) -> Writer[DS, SE]:
        return SQLWriter(self)

    def close(self) -> None:
        close_db(self._uri)

    def view(self, scope: DS, external: bool = False) -> View[DS, SE]:
        return SQLView(self, scope, external=external)

    def _execute(
        self, q: Select[Any], stream: bool = True
    ) -> Generator[Any, None, None]:
        # execute any read query against sql backend
        with self.engine.connect() as conn:
            if stream:
                conn = conn.execution_options(stream_results=True)
            cursor = conn.execute(q)
            while rows := cursor.fetchmany(10_000):
                yield from rows

    def _iterate_stmts(
        self, q: Select[Any], stream: bool = True
    ) -> Generator[Statement, None, None]:
        for row in self._execute(q, stream=stream):
            yield Statement.from_db_row(row)

    def _iterate(
        self, q: Select[Any], stream: bool = True
    ) -> Generator[SE, None, None]:
        current_id = None
        current_stmts: list[Statement] = []
        for stmt in self._iterate_stmts(q, stream=stream):
            entity_id = stmt.entity_id
            if current_id is None:
                current_id = entity_id
            if current_id != entity_id:
                proxy = self.assemble(current_stmts)
                if proxy is not None:
                    yield proxy
                current_id = entity_id
                current_stmts = []
            current_stmts.append(stmt)
        if current_stmts:
            proxy = self.assemble(current_stmts)
            if proxy is not None:
                yield proxy

nomenklatura.store.load_entity_file_store(path, resolver, cleaned=True)

Create a simple in-memory store by reading FtM entities from a file path.

Source code in nomenklatura/store/__init__.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def load_entity_file_store(
    path: Path,
    resolver: Resolver[StatementEntity],
    cleaned: bool = True,
) -> SimpleMemoryStore:
    """Create a simple in-memory store by reading FtM entities from a file path."""
    name = slugify(path.stem, sep="_") or Dataset.UNDEFINED
    dataset = Dataset.make({"name": name, "title": path.name})
    store = MemoryStore(dataset, resolver)
    with store.writer() as writer, open(path, "rb") as fh:
        while line := fh.readline():
            data = orjson.loads(line)
            proxy = StatementEntity.from_data(dataset, data, cleaned=cleaned)
            for ds in proxy.datasets:
                if ds not in dataset.dataset_names:
                    discovered = Dataset.make({"name": ds})
                    dataset.children.add(discovered)
            writer.add_entity(proxy)
    return store