Skip to content

Core

Cross-cutting concerns used by all layers.

Settings

Configuration from environment variables.

from ftm_lakehouse.core.settings import Settings

settings = Settings()
print(settings.uri)          # LAKEHOUSE_URI
print(settings.journal_uri)  # LAKEHOUSE_JOURNAL_URI

ftm_lakehouse.core.settings.Settings

Bases: BaseSettings

Source code in ftm_lakehouse/core/settings.py 
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
class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="lakehouse_",
        env_nested_delimiter="__",
        env_file=".env",
        secrets_dir=str(SECRETS_DIR) if SECRETS_DIR.is_dir() else None,
        nested_model_default_partial_update=True,
        extra="ignore",
    )

    uri: str = "data"
    journal_uri: str = "sqlite:///:memory:"
    api_key: str | None = None
    api_secret: str | None = None
    on_zfs: bool = False
    zfs_pool: str | None = None
    """ZFS dataset path the lakehouse's tuned datasets are created under.
    Transport / agent configuration (socket, owner, peer auth) lives in the
    external ``zfs-agent`` package's own ``ZFS_*`` environment."""

    grace_period_days: int = 30
    max_buffer_rows: int = 1_000_000

    journal_pool_size: int = 5
    """Postgres journal connections kept warm between writers
    (``LAKEHOUSE_JOURNAL_POOL_SIZE``). ``0`` pools nothing. It is per dataset: a
    worker writing many datasets holds up to this many idle connections for each
    of them, which is the figure to size against postgres
    ``max_connections``."""

    lock_max_retries: int = 22
    """Retry bound when acquiring the dataset write fence (``.LOCK``). Retry
    ``n`` sleeps ``n + rand(0, 1)`` seconds, so the total wait is roughly
    ``N²/2`` seconds – the default of 22 gives up after ~4.5 minutes; a lock
    left behind by a crashed writer must be released via ``ftm-lakehouse
    operations unlock``."""

    duckdb_memory_limit: str = "8GB"
    duckdb_temp_directory: str | None = None
    duckdb_extension_directory: str | None = None

    public_url_prefix: str | None = None

    @property
    def api_mode(self) -> bool:
        return self.uri.startswith("http")

    @property
    def resolved_journal_uri(self) -> str:
        if self.api_mode:
            # force journal uri to use api as well
            return self.uri
        return self.journal_uri

journal_pool_size = 5 class-attribute instance-attribute

Postgres journal connections kept warm between writers (LAKEHOUSE_JOURNAL_POOL_SIZE). 0 pools nothing. It is per dataset: a worker writing many datasets holds up to this many idle connections for each of them, which is the figure to size against postgres max_connections.

lock_max_retries = 22 class-attribute instance-attribute

Retry bound when acquiring the dataset write fence (.LOCK). Retry n sleeps n + rand(0, 1) seconds, so the total wait is roughly N²/2 seconds – the default of 22 gives up after ~4.5 minutes; a lock left behind by a crashed writer must be released via ftm-lakehouse operations unlock.

zfs_pool = None class-attribute instance-attribute

ZFS dataset path the lakehouse's tuned datasets are created under. Transport / agent configuration (socket, owner, peer auth) lives in the external zfs-agent package's own ZFS_* environment.

Path Conventions

Standard paths within the lakehouse.

Path conventions for the FollowTheMoney data lakehouse.

The fundamental idea is to have a convention-based file system layout with well-known paths for metadata and information interchange between processing stages.

All paths are dataset-relative unless otherwise noted.

Dataset Layout

::

lakehouse/
    index.json                          # catalog index
    config.yml                          # catalog configuration
    versions/                           # versioned snapshots
        YYYY/MM/YYYY-MM-DDTHH:MM:SS/
            index.json
            config.yml

    [dataset]/
        index.json                      # dataset index
        config.yml                      # dataset configuration

        versions/                       # versioned snapshots
            YYYY/MM/...

        .LOCK                           # dataset-wide maintenance lock
        .LOCK-APPENDS/                  # in-flight append markers
        .locks/{tenant}/                 # operation-specific locks
        tags/{tenant}/                  # workflow state / cache

        archive/                        # content-addressed file storage
            ab/cd/ef/{checksum}/        # SHA256 split into segments
                blob                    # file blob (stored once)
                {file_id}.json          # metadata (one per source path)
                {origin}.txt            # extracted text (one per engine)

        statements/                     # statement store (shard-partitioned)
            shard={shard}/
                bucket={bucket}/
                    origin={origin}/
                        *.parquet

        entities.ftm.json[.zst|gzip]    # aggregated entities export

        exports/
            statistics.json             # entity counts, facets
            statements.csv[.zst|gzip]   # sorted statements
            documents.csv[.zst|gzip]    # document metadata
            documents.{origin}.csv[...] # document metadata (origin-scoped)
            graph.cypher                # neo4j export (optional)

        diffs/                          # dirs are codec-free (they double
            entities.ftm.json/          #   as freshness tags); the files
                {ts}.delta.json[.zst|gzip]         # entities delta
            exports/
                documents.csv/
                    {ts}.diff.csv[.zst|gzip]       # documents delta
                documents.{origin}.csv/
                    {ts}.diff.csv[.zst|gzip]       # origin-scoped delta

        jobs/
            runs/
                {job_type}/
                    {timestamp}.json    # job run results

CONFIG = StoreKey('config.yml') module-attribute

user editable config filename

INDEX = StoreKey('index.json') module-attribute

generated index filename

ARCHIVE = 'archive' module-attribute

Base path for archive

STATEMENTS = 'statements' module-attribute

Base path for storing statement data (partitioned by shard, bucket, origin)

EXPORTS = StoreKey('exports') module-attribute

Base path for exports

Tag Conventions

Standard tags for freshness tracking.

Global tags used to identify actions. Used for cache keys of workflow runs etc.

Export operations don't have constants here – their freshness tag is the path.* export target itself (e.g. exports/statements.csv), touched by DatasetJobOperation._run_local after a successful run.

JOURNAL_UPDATED = 'journal/last_updated' module-attribute

Statement journal was updated

JOURNAL_FLUSHED = 'journal/last_flushed' module-attribute

Journal store last flushed into statement store

STATEMENTS_UPDATED = 'statements/last_updated' module-attribute

Statement store was updated

STATEMENTS_OPTIMIZED = 'statements/last_optimized' module-attribute

Statement store was optimized (merge + compact + vacuum)

ARCHIVE_UPDATED = 'archive/last_updated' module-attribute

Archive last updated (file added or removed)