Stores

ftmq extends the statement based store implementation of nomenklatura with more granular querying and aggregation possibilities.

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
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102

@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]. The backend is inferred
    by the scheme of the store uri.

    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 "aleph" in parsed.scheme:
        try:
            from ftmq.store.aleph import AlephStore

            # no `cast_types`: the aleph writer posts entity proxies to the
            # remote api, no statements pass through it
            return AlephStore.from_uri(uri, dataset=dataset, linker=linker)
        except ImportError:
            raise ImportError("Can not load AlephStore. Install `alephclient`")
    if uri.startswith("lake+"):
        try:
            from ftmq.store.lake import LakeStore

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

Supported backends

• in memory: get_store("memory://")
• Redis (or kvrocks): get_store("redis://localhost")
• LevelDB: get_store("leveldb://data")
• 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
• Clickhouse via ftm-clickhouse: get_store("clickhouse://localhost")

The duckdb backend is the sql store against a duckdb database file. Its path is spelled directly after the scheme: duckdb://relative.duckdb, duckdb:///absolute/path.duckdb, and an empty path (or duckdb://:memory:) opens an in-memory database. Don't confuse it with the delta lake store (lake+...), which 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 make several source ids one canonical entity. Without one every id is its own entity, which is why a store opened without a linker still works.

Two sources are supported, both via ftmq.store.base:

• get_resolver returns the read/write Resolver backed by 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). The decisions are loaded into memory when the resolver is built - it is a cached object, so a process that has to see another writer's decisions calls load_into_memory() itself.
• get_linker returns the read-only Linker: the merges without the judgement history. Its uri is either such a sql database, or an edge dump as written by Resolver.dump() / nomenklatura dump-resolver (json lines), which needs no database at all and can live anywhere anystore reads from.

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. A statement store answers by the canonical_id column, so the merge has to be in the statements: the sql-family and lake writers stamp the canonical id onto everything they write, and a store written before the decisions existed keeps the old ids. Handing such a store a linker afterwards is not enough - a filter, a count, an aggregation and the search index all still see the cluster members as separate entities, and filter:id=<canonical> matches nothing.

So resolve the data before serving it. Dump the store's statements, stamp the decisions onto them with the nomenklatura cli, and load them back - ftmq statements read / write are the store-side halves of that round trip (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 updates the store in place: the statement id does not cover canonical_id, so a resolved statement upserts onto the row it came from. ftmq statements write keeps the canonical id the stream carries and never re-derives it, so the apply-statements pass is what decides the outcome.

Data that has not been written yet needs none of this - a 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)

An already resolved store still needs the linker on the read side, for one thing: a lookup by a referent id. The statements carry the canonical id, so get_entity("left-1") finds nothing - the reader maps the id through the linker first (the api does this in [ftmq.api.store.View.get_entity][]). Everything else - filters, counts, aggregations - reads the resolved ids straight out of the data.

The in-memory store is the exception to the write side: it keeps whatever canonical id a statement already carries, so merges have to be applied before writing to it. It cannot dump its statements either - ftmq statements read and Store.statements need a SQL-family backend.

Read and query entities

Iterate through all the entities via Store.iterate:

from ftmq.store import get_store

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

Filter entities with a Query object using a store view:

from ftmq import Query, M

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

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 the smart_write_proxies shorthand, which uses the same bulk writer under the hood:

from ftmq.io import smart_write_proxies

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

The writer normalizes number and date values on the way in ("324,687.00" is stored as "324687.00", the raw string as the statement's original_value); the SQL backends rely on this format when aggregating or sorting numerically. Pass cast_types=False to get_store to skip it, and migrate existing data with ftmq statements cast-types.

Command line

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

Input entities that don't carry a dataset property are stored in the default dataset. To put them into a named one, stamp it on with ftmq apply-dataset (with --replace-dataset, so the entities end up 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