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 
11
12
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
class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="lakehouse_",
        env_nested_delimiter="__",
        env_file=".env",
        secrets_dir="/run/secrets",
        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_socket: str | None = None
    zfs_owner: str | None = None
    zfs_allowed_uid: int | None = None

    grace_period_days: int = 30
    max_buffer_rows: int = 1_000_000

    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, just
    inside a 300s reverse-proxy read timeout (each API client waiting on the
    fence pins a worker thread, so a higher bound multiplies across retrying
    clients). Long-running local bulk jobs can raise it via
    ``LAKEHOUSE_LOCK_MAX_RETRIES``. On exhaustion the writer raises
    ``RuntimeError`` instead of waiting forever; 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

    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

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, just inside a 300s reverse-proxy read timeout (each API client waiting on the fence pins a worker thread, so a higher bound multiplies across retrying clients). Long-running local bulk jobs can raise it via LAKEHOUSE_LOCK_MAX_RETRIES. On exhaustion the writer raises RuntimeError instead of waiting forever; a lock left behind by a crashed writer must be released via ftm-lakehouse operations unlock.

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 lock
        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)

        mappings/
            {content_hash}/
                mapping.yml             # current CSV mapping configuration
                versions/               # versioned snapshots
                    YYYY/MM/...

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

        entities.ftm.json               # aggregated entities export

        exports/
            statistics.json             # entity counts, facets
            statements.csv              # sorted statements
            documents.csv               # document metadata
            graph.cypher                # neo4j export (optional)

        diffs/
            entities.ftm.json/
                20240116T103000000000Z.delta.json  # entities delta
            exports/
                documents.csv/
                    20240116T103000000000Z.diff.csv  # documents delta

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

CONFIG = 'config.yml' module-attribute

user editable config filename

INDEX = 'index.json' module-attribute

generated index filename

ARCHIVE = 'archive' module-attribute

Base path for archive

STATEMENTS = f'{ENTITIES}/statements' module-attribute

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

MAPPINGS = 'mappings' module-attribute

Base path for storing mappings

EXPORTS = '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 :meth:DatasetJobOperation._run_local after a successful run.

JOURNAL_UPDATED = 'journal/last_updated' module-attribute

Statement journal was updated

STATEMENTS_UPDATED = 'statements/last_updated' module-attribute

Statement store was updated

ARCHIVE_UPDATED = 'archive/last_updated' module-attribute

Archive last updated (file added or removed)