ftm_lakehouse.lake

Public convenience functions for the lakehouse – repositories are the dataset handle.

from ftm_lakehouse import ensure_dataset, get_entities, get_archive, get_lakehouse

ensure_dataset("my_data", title="My Dataset", shards=8)

entities = get_entities("my_data")
archive = get_archive("my_data")

for name in get_lakehouse().list_datasets():
    ...

Get a lakehouse catalog.

Parameters:

Name	Type	Description	Default
`uri`	`Uri \| None`	Storage URI (default from LAKEHOUSE_URI setting)	`None`

Returns:

Type	Description
`Catalog`	Catalog instance

Source code in ftm_lakehouse/lake.py

@lru_cache(maxsize=LRU_MAX)
def get_lakehouse(uri: Uri | None = None) -> Catalog:
    """
    Get a lakehouse catalog.

    Args:
        uri: Storage URI (default from LAKEHOUSE_URI setting)

    Returns:
        Catalog instance
    """
    settings = Settings()
    storage_uri = ensure_uri(uri or settings.uri)
    log.info("Loading catalog", uri=mask_uri(storage_uri))
    return Catalog(uri=storage_uri)

Dataset config lifecycle

Get or create a dataset.

Creates config.yml if the dataset doesn't exist, recording data at creation (e.g. ensure_dataset("big_leak", shards=8)); data is ignored when the dataset already exists.

Parameters:

Name	Type	Description	Default
`name`	`str`	Dataset name.	required
`uri`	`Uri \| None`	Dataset storage root override.	`None`
`**data`	`Any`	Config data recorded at creation.	`{}`

Returns:

Type	Description
`DatasetModel`	The dataset's model.

Source code in ftm_lakehouse/catalog.py

def ensure_dataset(name: str, uri: Uri | None = None, **data: Any) -> DatasetModel:
    """Get or create a dataset.

    Creates ``config.yml`` if the dataset doesn't exist, recording ``data``
    at creation (e.g. ``ensure_dataset("big_leak", shards=8)``); ``data`` is
    ignored when the dataset already exists.

    Args:
        name: Dataset name.
        uri: Dataset storage root override.
        **data: Config data recorded at creation.

    Returns:
        The dataset's model.
    """
    if dataset_exists(name, uri):
        return get_dataset_model(name, uri)
    model = update_dataset(name, uri, **data)
    log.info("Created dataset", dataset=name)
    return model

Merge data into the dataset's config.yml (versioned snapshot).

Invalidates the repository factory caches afterwards so newly fetched repositories see the fresh config; instances held across the write keep their old snapshot (see the module docstring for the freshness contract).

Parameters:

Name	Type	Description	Default
`name`	`str`	Dataset name.	required
`uri`	`Uri \| None`	Dataset storage root override.	`None`
`**data`	`Any`	Fields to update in the model.	`{}`

Returns:

Type	Description
`DatasetModel`	The updated model.

Source code in ftm_lakehouse/catalog.py

def update_dataset(name: str, uri: Uri | None = None, **data: Any) -> DatasetModel:
    """Merge ``data`` into the dataset's ``config.yml`` (versioned snapshot).

    Invalidates the repository factory caches afterwards so newly fetched
    repositories see the fresh config; instances held across the write keep
    their old snapshot (see the module docstring for the freshness
    contract).

    Args:
        name: Dataset name.
        uri: Dataset storage root override.
        **data: Fields to update in the model.

    Returns:
        The updated model.
    """
    store = _dataset_store(name, uri)
    ensure_zfs(name, dataset_uri(name, uri))
    model = _load_model(store, name, **data)
    factories.get_versions(name, uri).make(path.CONFIG, model)
    factories.clear_caches()
    log.info("Updated dataset config", dataset=name, uri=mask_uri(store.uri))
    return model

The dataset's config, read fresh from config.yml on every call.

Parameters:

Name	Type	Description	Default
`name`	`str`	Dataset name.	required
`uri`	`Uri \| None`	Dataset storage root override (default: `{LAKEHOUSE_URI}/{name}`).	`None`

Source code in ftm_lakehouse/catalog.py

def get_dataset_model(name: str, uri: Uri | None = None) -> DatasetModel:
    """The dataset's config, read fresh from ``config.yml`` on every call.

    Args:
        name: Dataset name.
        uri: Dataset storage root override (default:
            ``{LAKEHOUSE_URI}/{name}``).
    """
    return _load_model(_dataset_store(name, uri), name)

The dataset's published index.json, falling back to the config.

The index is the config enriched with export resources and statistics, written by the index export operation.

Source code in ftm_lakehouse/catalog.py

def get_dataset_index(name: str, uri: Uri | None = None) -> DatasetModel:
    """The dataset's published ``index.json``, falling back to the config.

    The index is the config enriched with export resources and statistics,
    written by the ``index`` export operation.
    """
    versions = factories.get_versions(name, uri)
    index = versions.get(path.INDEX, model=get_model_class(), raise_on_nonexist=False)
    if index is not None:
        return index
    return get_dataset_model(name, uri)

Whether the dataset exists (has a config.yml).

Source code in ftm_lakehouse/catalog.py

def dataset_exists(name: str, uri: Uri | None = None) -> bool:
    """Whether the dataset exists (has a ``config.yml``)."""
    return _dataset_store(name, uri).exists(path.CONFIG)

Repository Shortcuts

Get the entity repository for a dataset (cached; the api-mode subclass for http uris).

Source code in ftm_lakehouse/repository/factories.py

def get_entities(dataset: str, uri: Uri | None = None) -> EntityRepository:
    """Get the entity repository for a dataset (cached; the api-mode
    subclass for http uris)."""
    return cast(
        EntityRepository, _resolve(_build_entities, dataset, dataset_uri(dataset, uri))
    )

Get the archive repository for a dataset (cached).

Source code in ftm_lakehouse/repository/factories.py

def get_archive(dataset: str, uri: Uri | None = None) -> ArchiveRepository:
    """Get the archive repository for a dataset (cached)."""
    return cast(
        ArchiveRepository,
        _resolve(ArchiveRepository, dataset, dataset_uri(dataset, uri)),
    )

Get the document repository for a dataset (cached).

Source code in ftm_lakehouse/repository/factories.py

def get_documents(dataset: str, uri: Uri | None = None) -> DocumentRepository:
    """Get the document repository for a dataset (cached)."""
    return cast(
        DocumentRepository,
        _resolve(DocumentRepository, dataset, dataset_uri(dataset, uri)),
    )

Custom dataset models

Register a custom DatasetModel subclass process-wide.

Every config read – repository construction, get_dataset_model, update_dataset, the index export – constructs models via get_model_class, so downstream applications extend the dataset config schema with one call at process start:

import ftm_lakehouse

class MyModel(ftm_lakehouse.DatasetModel):
    my_field: str | None = None

ftm_lakehouse.set_model_class(MyModel)

Call this before any repository or config access – repositories snapshot their model at construction and are LRU-cached, so a later switch requires repository.factories.clear_caches().

Parameters:

Name	Type	Description	Default
`model_class`	`type[DatasetModel]`	The `DatasetModel` subclass to use.	required

Source code in ftm_lakehouse/model/dataset.py

def set_model_class(model_class: type[DatasetModel]) -> None:
    """Register a custom [`DatasetModel`][DatasetModel] subclass process-wide.

    Every config read – repository construction, ``get_dataset_model``,
    ``update_dataset``, the index export – constructs models via
    `get_model_class`, so downstream applications extend the dataset
    config schema with one call at process start:

    ```python
    import ftm_lakehouse

    class MyModel(ftm_lakehouse.DatasetModel):
        my_field: str | None = None

    ftm_lakehouse.set_model_class(MyModel)
    ```

    Call this **before** any repository or config access – repositories
    snapshot their model at construction and are LRU-cached, so a later
    switch requires ``repository.factories.clear_caches()``.

    Args:
        model_class: The [`DatasetModel`][DatasetModel] subclass to use.
    """
    global _model_class
    _model_class = model_class

Classes

Multi-dataset lakehouse catalog – enumeration and dataset addressing.

Example

from ftm_lakehouse import get_entities, get_lakehouse

catalog = get_lakehouse()
for name in catalog.list_datasets():
    print(name, get_entities(name).stats())

Source code in ftm_lakehouse/catalog.py

class Catalog:
    """Multi-dataset lakehouse catalog – enumeration and dataset addressing.

    Example:
        ```python
        from ftm_lakehouse import get_entities, get_lakehouse

        catalog = get_lakehouse()
        for name in catalog.list_datasets():
            print(name, get_entities(name).stats())
        ```
    """

    def __init__(self, uri: Uri) -> None:
        self.uri = uri
        self._log = get_logger(__name__, catalog=mask_uri(uri))

    def __repr__(self) -> str:
        return f"Catalog({mask_uri(self.uri)!r})"

    @cached_property
    def _store(self) -> Store:
        """Raw storage access."""
        return get_store(uri=ensure_api_uri(self.uri), serialization_mode="raw")

    def dataset_uri(self, name: str) -> str:
        """Validated canonical uri for ``name`` under this catalog's root.

        Raises:
            ValueError: If ``name`` is not a valid dataset name.
        """
        return dataset_uri(name, join_uri(self.uri, name))

    def list_datasets(self) -> Generator[str, None, None]:
        """Yield the names of all datasets that have a ``config.yml``."""
        for child in self._store._fs.ls(self.uri):
            name = Path(child).name
            if self._store.exists(f"{name}/{path.CONFIG}"):
                yield name

    def ensure_dataset(self, name: str, **data: Any) -> DatasetModel:
        """Get or create a dataset under this catalog.

        See [`ensure_dataset`][ensure_dataset].
        """
        return ensure_dataset(name, self.dataset_uri(name), **data)

`dataset_uri(name)`

Validated canonical uri for name under this catalog's root.

Raises:

Type	Description
`ValueError`	If `name` is not a valid dataset name.

Source code in ftm_lakehouse/catalog.py

def dataset_uri(self, name: str) -> str:
    """Validated canonical uri for ``name`` under this catalog's root.

    Raises:
        ValueError: If ``name`` is not a valid dataset name.
    """
    return dataset_uri(name, join_uri(self.uri, name))

`list_datasets()`

Yield the names of all datasets that have a config.yml.

Source code in ftm_lakehouse/catalog.py

def list_datasets(self) -> Generator[str, None, None]:
    """Yield the names of all datasets that have a ``config.yml``."""
    for child in self._store._fs.ls(self.uri):
        name = Path(child).name
        if self._store.exists(f"{name}/{path.CONFIG}"):
            yield name

`ensure_dataset(name, **data)`

Get or create a dataset under this catalog.

See ensure_dataset.

Source code in ftm_lakehouse/catalog.py

def ensure_dataset(self, name: str, **data: Any) -> DatasetModel:
    """Get or create a dataset under this catalog.

    See [`ensure_dataset`][ensure_dataset].
    """
    return ensure_dataset(name, self.dataset_uri(name), **data)