Stores

ftmq extends the statement stores of nomenklatura with querying and aggregation.

Initialize a store

from ftmq.store import get_store

# an in-memory store:
get_store("memory://")

# a leveldb store:
get_store("leveldb:///var/lib/data")

# a sqlite store
get_store("sqlite:///data/followthemoney.db")

# a duckdb store
get_store("duckdb:///data/followthemoney.duckdb")

13
14
15
16
17
18
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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

@cache
def get_store(
    uri: Uri | None = settings.DB_URL,
    dataset: Dataset | str | None = None,
    linker: Linker | None = None,
    cast_types: bool = True,
) -> Store:
    """Get an initialized [Store][ftmq.store.base.Store], inferred by the uri scheme.

    Example:
        ```python
        from ftmq.store import get_store

        # an in-memory store:
        get_store("memory://")

        # a leveldb store:
        get_store("leveldb:///var/lib/data")

        # a sqlite store
        get_store("sqlite:///data/followthemoney.db")

        # a duckdb store
        get_store("duckdb:///data/followthemoney.duckdb")
        ```

    Args:
        uri: The store backend uri
        dataset: A `followthemoney.Dataset` instance to limit the scope to
        linker: A `nomenklatura.Linker` instance with linked / deduped data
        cast_types: Normalize statement values on write (see
            [`ftmq.statements`][ftmq.statements])

    Returns:
        The initialized store. This is a cached object.
    """
    uri = str(uri)
    parsed = urlparse(uri)
    if parsed.scheme == "memory":
        return MemoryStore(dataset, linker=linker, cast_types=cast_types)
    if parsed.scheme == "leveldb":
        path = uri.replace("leveldb://", "")
        path = Path(path).absolute()
        try:
            from ftmq.store.level import LevelDBStore

            return LevelDBStore(
                dataset, path=path, linker=linker, cast_types=cast_types
            )
        except ImportError:
            raise ImportError("Can not load LevelDBStore. Install `plyvel`")
    if parsed.scheme == "duckdb":
        try:
            from ftmq.store.duckdb import DuckDBStore

            return DuckDBStore(dataset, uri=uri, linker=linker, cast_types=cast_types)
        except ImportError:
            raise ImportError("Can not load DuckDBStore. Install `duckdb-engine`")
    if "sql" in parsed.scheme:
        try:
            from ftmq.store.sql import SQLStore

            return SQLStore(dataset, uri=uri, linker=linker, cast_types=cast_types)
        except ImportError:
            raise ImportError("Can not load SqlStore. Install sql dependencies.")
    if uri.startswith("lake+"):
        try:
            from ftmq.store.lake import LakeStore

            return LakeStore(
                uri=uri[5:], dataset=dataset, linker=linker, cast_types=cast_types
            )
        except ImportError:
            raise ImportError("Can not load LakeStore. Install `[lake]` dependencies")
    raise NotImplementedError(uri)

Supported backends

• in memory: get_store("memory://")
• LevelDB: get_store("leveldb://data") (needs the level extra)
• Sql:
  • sqlite: get_store("sqlite:///data.db")
  • postgresql: get_store("postgresql://user:password@host/db")
  • duckdb: get_store("duckdb://data.duckdb") (needs the duckdb extra)
  • ...any other supported by sqlalchemy
• Delta lake: get_store("lake+s3://bucket/path") (needs the lake extra)

duckdb paths follow the scheme directly: duckdb://relative.duckdb, duckdb:///absolute/path.duckdb; an empty path (or duckdb://:memory:) opens an in-memory database. The delta lake store (lake+...) is different: it queries parquet files through duckdb instead of owning a database file.

Merged entities (resolver / linker)

A store resolves entity ids through a nomenklatura Linker, the deduplication decisions that merge several source ids into one canonical entity. Without one every id is its own entity.

Both sources live in ftmq.store.base:

• get_resolver: the read/write Resolver over a resolver table in a sql database. This is the default; a store opened without a linker puts the table in its own database (a non-sql store gets an ephemeral in-memory one). Decisions are loaded into memory on construction and the object is cached, so to see another writer's decisions call load_into_memory().
• get_linker: the read-only Linker. Its uri is a sql database as above, or an edge dump from Resolver.dump() / nomenklatura dump-resolver (json lines) on any anystore location.

from ftmq.store import get_store
from ftmq.store.base import get_linker

# merge decisions from a json dump, entities from a sql store
store = get_store("sqlite:///followthemoney.store", linker=get_linker("s3://data/resolver.ijson"))

A linker resolves ids, not data. Stores read by the canonical_id column, so the merge has to be in the statements. The sql-family and lake writers stamp the canonical id on write; a store written before the decisions existed keeps the old ids, and filters, counts, aggregations and the search index still see the cluster members separately (filter:id=<canonical> matches nothing).

Resolve the data before serving it: dump the statements, apply the decisions with the nomenklatura cli, and load them back with ftmq statements read / write (see the cli docs):

nomenklatura dump-resolver resolver.ijson
ftmq statements read -i sqlite:///followthemoney.store -o statements.csv
nomenklatura apply-statements -i statements.csv -o resolved.csv
ftmq statements write -i resolved.csv -o sqlite:///followthemoney.store

The reload upserts in place (the statement id does not cover canonical_id), and write keeps the canonical id each statement carries.

New data needs none of this; the writer applies the linker it was given:

store = get_store("sqlite:///resolved.store", linker=get_linker("resolver.ijson"))
with store.writer() as bulk:
    for proxy in smart_read_proxies("entities.ftm.json"):
        bulk.add_entity(proxy)

A resolved store still needs the linker to look up a referent id: get_entity("left-1") finds nothing unless the id is mapped to its canonical first (as ftmq.api.store.get_entity does). Filters, counts and aggregations read the resolved ids from the data.

The in-memory store keeps the canonical id a statement already carries, so apply merges before writing to it. Dumping statements (ftmq statements read, Store.statements) needs a SQL-family backend.

Read and query entities

Iterate all entities via Store.iterate:

from ftmq.store import get_store

store = get_store("sqlite:///followthemoney.store")
proxies = store.iterate()

Filter with a Query on a store view:

from ftmq import Query, M

q = Query().where(M(dataset="my_dataset"), M(schema="Person"))
view = store.default_view()
proxies = view.query(q)

A view hides external statements (unaccepted enrichment candidates) in query(), count(), stats() and aggregations(), unless built with store.view(scope, external=True).

Command line

ftmq -i sqlite:///followthemoney.store -d my_dataset -q 'filter:schema=Person'

cli reference

Write entities to a store

Use the bulk writer:

proxies = [...]

with store.writer() as bulk:
    for proxy in proxies:
        bulk.add_entity(proxy)

Or smart_write_proxies, which uses the same writer:

from ftmq.io import smart_write_proxies

smart_write_proxies("sqlite:///followthemoney.store", proxies)

The writer normalizes number and date values ("324,687.00" is stored as "324687.00", the raw string as original_value), which the SQL backends rely on for numeric aggregation and sorting. get_store(..., cast_types=False) skips it; migrate existing data with ftmq statements cast-types.

Command line

cat entities.ftm.json | ftmq -o sqlite:///followthemoney.store

Entities without a dataset are stored in the default dataset. Stamp a named one on with ftmq apply-dataset (--replace-dataset puts them in that dataset alone; a statement carries exactly one):

ftmq apply-dataset -d my_dataset --replace-dataset -i s3://data/entities.ftm.json -o sqlite:///followthemoney.store

cli reference