Tables API reference¶

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

    filtered_tables = []
    for table_name in tables:
        tb_handler = self._get_table_group_handler(table_name)
        table_type = _get_meta(tb_handler).type
        if table_type == filter_types:
            filtered_tables.append(table_name)
    return filtered_tables

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)

    meta = _get_meta(table_handler)
    return ImplementedTables().get_table(
        meta=meta,
        handler=table_handler,
        backend=backend,
        strict=strict,
    )

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._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()
    if name not in existing_tables:
        existing_tables.append(name)
        self._group_handler.write_attrs({"tables": existing_tables})

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

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

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

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

@classmethod
def from_handler(
    cls,
    handler: ZarrGroupHandler,
    backend: TableBackend | None = None,
) -> "GenericTable":
    return cls._from_handler(
        handler=handler,
        backend=backend,
        meta_model=BackendMeta,
    )

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
            )