ngio.tables API reference¶

@staticmethod
@abstractmethod
def table_type() -> str:
    """Return the type of the table."""
    ...

@staticmethod
@abstractmethod
def version() -> str:
    """The generic table does not have a version.

    Since does not follow a specific schema.
    """
    ...

def load_as_anndata(self) -> AnnData:
    """Load the table as an AnnData object."""
    if self._table_backend is None:
        raise NgioValueError("No backend set for the table.")
    return self._table_backend.load_as_anndata()

def load_as_pandas_df(self) -> pd.DataFrame:
    """Load the table as a pandas DataFrame."""
    if self._table_backend is None:
        raise NgioValueError("No backend set for the table.")
    return self._table_backend.load_as_pandas_df()

def load_as_polars_lf(self) -> pl.LazyFrame:
    """Load the table as a polars LazyFrame."""
    if self._table_backend is None:
        raise NgioValueError("No backend set for the table.")
    return self._table_backend.load_as_polars_lf()

def set_table_data(
    self,
    table_data: TabularData | None = None,
    refresh: bool = False,
) -> None:
    """Set the table.

    If an object is passed, it will be used as the table.
    If None is passed, the table will be loaded from the backend.

    If refresh is True, the table will be reloaded from the backend.
        If table is not None, this will be ignored.
    """
    if table_data is not None:
        if not isinstance(table_data, TabularData):
            raise NgioValueError(
                "The table must be a pandas DataFrame, polars LazyFrame, "
                " or AnnData object."
            )

        self._table_data = normalize_table(
            table_data,
            index_key=self.index_key,
            index_type=self.index_type,
        )
        return None

    if self._table_data is not None and not refresh:
        return None

    if self._table_backend is None:
        raise NgioValueError(
            "The table does not have a DataFrame in memory nor a backend."
        )
    self._table_data = self._table_backend.load()

def set_backend(
    self,
    handler: ZarrGroupHandler | None = None,
    backend: TableBackend | None = None,
) -> None:
    """Set the backend of the table.

    If `backend` is `None`, the backend stored in the table metadata
    is used.

    If no handler is provided and the table is not yet attached to a
    Zarr group, a string `backend` is only recorded as the table's
    preferred backend; the backend is instantiated when the table is
    written (e.g. by `add_table`).
    """
    if handler is None:
        if self._table_backend is None:
            if backend is None:
                return None
            if isinstance(backend, str):
                backends = ImplementedTableBackends()
                self._meta.backend = backends.normalize_backend_name(backend)
                return None
            raise NgioValueError(
                "A ZarrGroupHandler must be provided to attach a "
                "backend instance to the table."
            )
        handler = self._table_backend.group_handler

    meta = self._meta
    _backend = self._load_backend(
        meta=meta,
        handler=handler,
        backend=backend,
    )
    self._table_backend = _backend
    self._meta.backend = _backend.backend_name()

@classmethod
@abstractmethod
def from_handler(
    cls,
    handler: ZarrGroupHandler,
    backend: TableBackend | None = None,
    attrs: dict | None = None,
) -> Self:
    """Create a new ROI table from a Zarr group handler."""
    # Not `pass`: `abstractmethod` only guards instantiation, so a call on
    # a subclass that forgot to override this would silently return `None`
    # — `get_as(name, cls)` handed back nothing, with no error anywhere.
    raise NotImplementedError(f"{cls.__name__} does not implement `from_handler`.")

@classmethod
def from_table_data(cls, table_data: TabularData, meta: BackendMeta) -> Self:
    """Create a new ROI table from a Zarr group handler."""
    return cls(
        table_data=table_data,
        meta=meta,
    )

def consolidate(self) -> None:
    """Write the current state of the table to the Zarr file."""
    if self._table_backend is None:
        raise NgioValueError(
            "No backend set for the table. "
            "Please add the table to a OME-Zarr Image before calling consolidate."
        )

    self._table_backend.write(
        self.table_data,
        metadata=self._meta.model_dump(exclude_none=True),
    )

def available_implementations(self) -> list[str]:
    """Get the available table handler versions."""
    return list(self._implemented_tables.keys())

def get_table(
    self,
    meta: TableMeta,
    handler: ZarrGroupHandler,
    backend: TableBackend | None = None,
    strict: bool = True,
    attrs: dict | None = None,
) -> Table:
    """Try to get a handler for the given store based on the metadata version.

    `attrs` are the group attributes `meta` was decoded from, forwarded so
    the concrete table does not read the same document again.
    """
    if strict:
        default = None
    else:
        default = GenericTable

    table_cls = self._implemented_tables.get(meta.unique_name(), default)
    if table_cls is None:
        raise NgioValueError(
            f"Table handler for {meta.unique_name()} not implemented."
        )
    table = table_cls.from_handler(handler=handler, backend=backend, attrs=attrs)
    return table

def add_implementation(
    self,
    handler: type[Table],
    overwrite: bool = False,
    aliases: list[str] | None = None,
) -> None:
    """Register a new table handler."""
    meta = TableMeta(
        type=handler.table_type(),
        table_version=handler.version(),
    )

    self._add_implementation(handler, meta.unique_name(), overwrite)

    if aliases is not None:
        for alias in aliases:
            self._add_implementation(handler, alias, overwrite)

@staticmethod
def table_type() -> str:
    """Return the type of the table."""
    ...

@staticmethod
def version() -> str:
    """Return the version of the table."""
    ...

def set_table_data(
    self,
    table_data: TabularData | None = None,
    refresh: bool = False,
) -> None:
    """Make sure that the table data is set (exist in memory).

    If an object is passed, it will be used as the table.
    If None is passed, the table will be loaded from the backend.

    If refresh is True, the table will be reloaded from the backend.
        If table is not None, this will be ignored.
    """
    ...

def set_backend(
    self,
    handler: ZarrGroupHandler | None = None,
    backend: TableBackend | None = None,
) -> None:
    """Set the backend store and path for the table.

    If the handler is `None` it will be inferred from the current backend.
    If the backend is `None`, the table's own backend is preserved.
    """
    ...

@classmethod
def from_handler(
    cls,
    handler: ZarrGroupHandler,
    backend: TableBackend | None = None,
    attrs: dict | None = None,
) -> Self:
    """Create a new table from a Zarr group handler."""
    ...

@classmethod
def from_table_data(cls, table_data: TabularData, meta: BackendMeta) -> Self:
    """Create a new table from a DataFrame."""
    ...

def consolidate(self) -> None:
    """Consolidate the table on disk."""
    ...

def table_types(self, names: Sequence[str] | None = None) -> dict[str, str]:
    """Return the table type of each table, keyed by name.

    The type lives only in each table's own attributes — the `/tables`
    group records names — so learning it costs one group open per table.
    Under `cache=True` that result is memoised: a table's type does not
    change without going through `add`/`delete` here, and those drop the
    memo. Under `cache=False` every call re-reads each type, so a table
    overwritten with a different type by another handle is picked up.
    Names are re-read every call in both modes.

    Args:
        names: Restrict to these tables. Defaults to every table listed.
    """
    if names is None:
        names = self._get_tables_list()

    use_memo = self._group_handler.use_cache
    for name in names:
        if not use_memo or name not in self._types_memo:
            try:
                handler = self._get_table_group_handler(name)
            except NgioFileNotFoundError:
                # A stale name in the `tables` attribute with no group
                # behind it. Not memoised (and any earlier memo is
                # dropped): it stays invisible to typed listings but
                # keeps raising on a direct `get`.
                self._types_memo.pop(name, None)
                continue
            self._types_memo[name] = _get_meta(handler).type
    return {
        name: self._types_memo[name] for name in names if name in self._types_memo
    }

def list(self, filter_types: TypedTable | str | None = None) -> list[str]:
    """List all tables in the group.

    Args:
        filter_types: If provided, only return tables of this type.

    Returns:
        A list of table names.
    """
    tables = self._get_tables_list()
    if filter_types is None:
        return tables

    return [
        name
        for name, table_type in self.table_types(tables).items()
        if table_type == filter_types
    ]

def get(
    self,
    name: str,
    backend: TableBackend | None = None,
    strict: bool = True,
) -> Table:
    """Get a table from the group.

    Args:
        name: The name of the table.
        backend: The backend to use for reading the table.
        strict: If True, raise an error if the table type is not implemented.

    Returns:
        The table object.
    """
    if name not in self.list():
        raise NgioValueError(f"Table '{name}' not found in the group.")

    table_handler = self._get_table_group_handler(name)

    # Read once: `TableMeta` picks the class and the concrete model is
    # built by the class — both consume this same document.
    attrs = table_handler.load_attrs()
    meta = _get_meta(table_handler, attrs=attrs)
    return ImplementedTables().get_table(
        meta=meta,
        handler=table_handler,
        backend=backend,
        strict=strict,
        attrs=attrs,
    )

def get_as(
    self,
    name: str,
    table_cls: type[TableType],
    backend: TableBackend | None = None,
) -> TableType:
    """Get a table from the group as a specific type.

    Args:
        name: The name of the table.
        table_cls: The table class to use for loading the table.
        backend: The backend to use for reading the table.

    Returns:
        The table object of the specified type.
    """
    if name not in self.list():
        raise NgioValueError(f"Table '{name}' not found in the group.")

    table_handler = self._get_table_group_handler(name)
    return table_cls.from_handler(
        handler=table_handler,
        backend=backend,
    )

def delete(self, name: str, missing_ok: bool = False) -> None:
    """Delete a table from the group.

    Args:
        name (str): The name of the table to delete.
        missing_ok (bool): If True, do not raise an error if
            the table does not exist.
    """
    existing_tables = self._get_tables_list()
    if name not in existing_tables:
        if missing_ok:
            return
        raise NgioValueError(
            f"Table '{name}' not found in the Tables group. "
            f"Available tables: {existing_tables}"
        )

    self._group_handler.delete_group(name)
    existing_tables.remove(name)
    self._types_memo.pop(name, None)
    self._group_handler.write_attrs({"tables": existing_tables})

def add(
    self,
    name: str,
    table: Table,
    backend: TableBackend | None = None,
    overwrite: bool = False,
) -> None:
    """Add a table to the group.

    Args:
        name: The name of the table.
        table: The table object to add.
        backend: The backend to use for writing the table. If `None`
            (default), the table's own backend is preserved.
        overwrite: Whether to overwrite an existing table with the same name.
    """
    existing_tables = self._get_tables_list()
    if name in existing_tables and not overwrite:
        raise NgioValueError(
            f"Table '{name}' already exists in the group. "
            "Use overwrite=True to replace it."
        )

    table_handler = self._group_handler.get_handler(path=name, overwrite=overwrite)

    if backend is None:
        backend = table.backend_name

    table.set_table_data()
    table.set_backend(
        handler=table_handler,
        backend=backend,
    )
    table.consolidate()
    # The type is written by `consolidate`, and `overwrite=True` can change
    # it, so drop any memoised value rather than trusting it.
    self._types_memo.pop(name, None)
    if name not in existing_tables:
        existing_tables.append(name)
        self._group_handler.write_attrs({"tables": existing_tables})

def normalize_backend_name(self, backend_name: str) -> str:
    """Resolve a backend name or alias to its canonical name.

    Raises:
        NgioValueError: If the backend name is not implemented.
    """
    if backend_name not in self._implemented_backends:
        raise NgioValueError(f"Table backend {backend_name} not implemented.")
    return self._implemented_backends[backend_name].backend_name()

def get_backend(
    self,
    *,
    group_handler: ZarrGroupHandler,
    backend_name: str,
    index_key: str | None = None,
    index_type: Literal["int", "str"] | None = None,
) -> TableBackendProtocol:
    """Instantiate the named backend and attach it to `group_handler`.

    Raises:
        NgioValueError: If `backend_name` is not registered.
    """
    if backend_name not in self._implemented_backends:
        raise NgioValueError(f"Table backend {backend_name} not implemented.")
    backend = self._implemented_backends[backend_name]()
    backend.set_group_handler(
        group_handler=group_handler, index_key=index_key, index_type=index_type
    )
    return backend

def add_backend(
    self,
    table_backend: type[TableBackendProtocol],
    overwrite: bool = False,
    aliases: list[str] | None = None,
) -> None:
    """Register a new handler."""
    self._add_backend(
        table_backend=table_backend,
        name=table_backend.backend_name(),
        overwrite=overwrite,
    )
    if aliases is not None:
        for alias in aliases:
            self._add_backend(
                table_backend=table_backend, name=alias, overwrite=overwrite
            )

def set_group_handler(
    self,
    group_handler: ZarrGroupHandler,
    index_key: str | None = None,
    index_type: Literal["int", "str"] | None = None,
) -> None:
    """Attach a group handler to the backend.

    Index keys and index types are used to ensure that the
    serialization and deserialization of the table
    is consistent across different backends.

    Making sure that this is consistent is
    a duty of the backend implementations.
    """
    ...

@staticmethod
def backend_name() -> str:
    """Return the name of the backend.

    As a convention we set name as:
        {backend_name}_v{version}

    Where the version is a integer.
    """
    ...

@staticmethod
def implements_anndata() -> bool:
    """Check if the backend implements the anndata protocol.

    If this is True, the backend should implement the
    `write_from_anndata` method.

    AnnData objects are more complex than DataFrames,
    so if this is true the backend should implement the
    full serialization of the AnnData object.

    If this is False, these methods should raise a
    `NotImplementedError`.
    """
    ...

@staticmethod
def implements_pandas() -> bool:
    """Check if the backend implements the pandas protocol.

    If this is True, the backend should implement the
    `write_from_dataframe` methods.

    If this is False, these methods should raise a
    `NotImplementedError`.
    """
    ...

@staticmethod
def implements_polars() -> bool:
    """Check if the backend implements the polars protocol.

    If this is True, the backend should implement the
    `write_from_polars` methods.

    If this is False, these methods should raise a
    `NotImplementedError`.
    """
    ...

def load_as_anndata(self) -> AnnData:
    """Load the table as an AnnData object."""
    ...

def load_as_pandas_df(self) -> DataFrame:
    """Load the table as a pandas DataFrame."""
    ...

def load_as_polars_lf(self) -> LazyFrame:
    """Load the table as a polars LazyFrame."""
    ...

def load(self) -> TabularData:
    """The default load method.

    This method will be default way to load the table
    from the backend. This method should wrap one of the
    `load_as_anndata`, `load_as_dataframe` or `load_as_polars`
    methods depending on the backend implementation.
    """
    ...

def write_from_pandas(self, table: DataFrame) -> None:
    """Serialize the table from a pandas DataFrame."""
    ...

def write_from_anndata(self, table: AnnData) -> None:
    """Serialize the table from an AnnData object."""
    ...

def write_from_polars(self, table: LazyFrame | PolarsDataFrame) -> None:
    """Serialize the table from a polars DataFrame or LazyFrame."""
    ...

def write(
    self,
    table_data: DataFrame | AnnData | PolarsDataFrame | LazyFrame,
    metadata: dict[str, str] | None = None,
) -> None:
    """This is a generic write method.

    Will call the appropriate write method
    depending on the type of the table.

    Moreover it will also write the metadata
    if provided, and the backend methadata

    the backend should write in the zarr group attributes
        - backend: the backend name (self.backend_name())
        - index_key: the index key
        - index_type: the index type

    """