Repository: Avaiga/taipy
File: taipy/gui/data/data_accessor.py (Python)

Code before:
        new_row: t.Optional[t.List[t.Any]] = None,
    ) -> t.Optional[t.Any]:
        pass

    @abstractmethod
    def to_csv(self, var_name: str, value: t.Any) -> t.Optional[str]:
        pass


class _InvalidDataAccessor(_DataAccessor):

    def __init__(self, gui: "Gui") -> None:
        super().__init__(gui)

    @staticmethod
    def get_supported_classes() -> t.List[t.Type]:
        return []

    def get_data(
        self,
        var_name: str,
        value: t.Any,
        payload: t.Dict[str, t.Any],
        data_format: _DataFormat,
    ) -> t.Dict[str, t.Any]:
        transformed_value = self._gui.on_invalid_data(value)
        if transformed_value is not None:
            return {var_name: transformed_value}
        return {}
    def get_col_types(self, var_name: str, value: t.Any) -> t.Dict[str, str]:
        transformed_value = self._gui.on_invalid_data(value)
        if transformed_value is not None:
            return {var_name: str(type(transformed_value))}
        return {}
    def to_pandas(self, value: t.Any) -> t.Union[t.List[t.Any], t.Any]:
        transformed_value = self._gui.on_invalid_data(value)
        if transformed_value is not None:
            try:
                import pandas as pd
                return pd.DataFrame([transformed_value])
            except ImportError:
                pass
        return None
    def on_edit(self, value: t.Any, payload: t.Dict[str, t.Any]):
        transformed_value = self._gui.on_invalid_data(value)
        return transformed_value if transformed_value is not None else None
    def on_delete(self, value: t.Any, payload: t.Dict[str, t.Any]):
        transformed_value = self._gui.on_invalid_data(value)
        return transformed_value if transformed_value is not None else None
    def on_add(
        self,

Review comment (suggestion):
I think the callback should not be invoked here.
It should be invoked when evaluating the right `DataAccessor` for the data type.
this way, we can redirect to an existing `DataAccessor` and keep the `InvalidDataAccessor` for invalid data types.

Code after:

class _InvalidDataAccessor(_DataAccessor):
    @staticmethod
    def get_supported_classes() -> t.List[t.Type]:
        return []

    def get_data(
        self, var_name: str, value: t.Any, payload: t.Dict[str, t.Any], data_format: _DataFormat
    ) -> t.Dict[str, t.Any]:
        return {}

    def get_col_types(self, var_name: str, value: t.Any) -> t.Dict[str, str]:
        return {}

    def to_pandas(self, value: t.Any) -> t.Union[t.List[t.Any], t.Any]:
        return None

    def on_edit(self, value: t.Any, payload: t.Dict[str, t.Any]):
        return None

    def on_delete(self, value: t.Any, payload: t.Dict[str, t.Any]):
        return None

    def on_add(self, value: t.Any, payload: t.Dict[str, t.Any], new_row: t.Optional[t.List[t.Any]] = None):
        return None

    def to_csv(self, var_name: str, value: t.Any):
        return None


class _DataAccessors(object):
    def __init__(self, gui: "Gui") -> None:
        self.__access_4_type: t.Dict[t.Type, _DataAccessor] = {}
        self.__invalid_data_accessor = _InvalidDataAccessor(gui)
        self.__data_format = _DataFormat.JSON
        self.__gui = gui

        from .array_dict_data_accessor import _ArrayDictDataAccessor
        from .numpy_data_accessor import _NumpyDataAccessor
        from .pandas_data_accessor import _PandasDataAccessor

        self._register(_PandasDataAccessor)
        self._register(_ArrayDictDataAccessor)
        self._register(_NumpyDataAccessor)

    def _register(self, cls: t.Type[_DataAccessor]) -> None:
        """Register a new DataAccessor type."""
        if not inspect.isclass(cls):
            raise AttributeError("The argument of 'DataAccessors.register()' should be a class")
        if not issubclass(cls, _DataAccessor):
            raise TypeError(f"Class {cls.__name__} is not a subclass of DataAccessor")