omnipy.components.tables.models

CLASS	DESCRIPTION
ColumnModel
ColumnWiseTableWithColNamesAndIndexModel
ColumnWiseTableWithColNamesModel
ColumnWiseTableWithColNamesNoConvertModel
ConcatByAddArrayAdapterModel
CsvTableModel
CsvTableOfPydanticRecordsModel
IterRow
IteratingPydanticRecordsModel
JsonMaxLevel1ColumnModel
JsonMaxLevel1ColumnWiseTableWithColNamesModel
JsonMaxLevel2ColumnModel
JsonMaxLevel2ColumnWiseTableWithColNamesModel
JsonScalarColumnModel
JsonScalarColumnWiseTableWithColNamesModel
PrintableTable
PydanticRecordModel
PydanticRecordModelBase
PydanticRecordModelMetaclass
RowWiseTableFirstRowAsColNamesModel
RowWiseTableModel
RowWiseTableWithColNamesModel
RowWiseTableWithColNamesNoConvertModel
TableOfPydanticRecordsModel
TsvTableModel

ATTRIBUTE	DESCRIPTION
ColWiseAddOtherType	TYPE: TypeAlias
JsonMaxLevel1Types	TYPE: TypeAlias
JsonMaxLevel2Types	TYPE: TypeAlias

ColWiseAddOtherType module-attribute

ColWiseAddOtherType: TypeAlias = (
    ColumnWiseTableWithColNamesModel
    | RowWiseTableWithColNamesModel
    | Mapping[str, ColumnModel]
    | list[Mapping[str, JsonScalar]]
)

JsonMaxLevel1Types module-attribute

JsonMaxLevel1Types: TypeAlias = JsonScalar | JsonDictOfScalars | JsonListOfScalars

JsonMaxLevel2Types module-attribute

JsonMaxLevel2Types: TypeAlias = (
    JsonScalar
    | list[JsonScalar | JsonDictOfScalars | JsonListOfScalars]
    | dict[str, JsonScalar | JsonDictOfScalars | JsonListOfScalars]
)

ColumnModel

Bases: Model[_ColumnT], Generic[_ColumnT, _ItemT]

Source code in src/omnipy/components/tables/models.py

class ColumnModel(Model[_ColumnT], Generic[_ColumnT, _ItemT]):
    ...

ColumnWiseTableWithColNamesAndIndexModel

Bases: Model[dict[str, Mapping[str, JsonScalar]]]

Source code in src/omnipy/components/tables/models.py

class ColumnWiseTableWithColNamesAndIndexModel(Model[dict[str, Mapping[str, JsonScalar]]]):
    ...

ColumnWiseTableWithColNamesModel

Bases: _ColumnWiseTableWithColNamesMixin[_ColumnModelT, _ColumnModelItemT], _ColumnWiseTableWithColNamesModel[_ColumnModelT], PrintableTable, Generic[_ColumnModelT, _ColumnModelItemT]

Source code in src/omnipy/components/tables/models.py

class ColumnWiseTableWithColNamesModel(
        _ColumnWiseTableWithColNamesMixin[_ColumnModelT, _ColumnModelItemT],
        _ColumnWiseTableWithColNamesModel[_ColumnModelT],
        PrintableTable,
        Generic[_ColumnModelT, _ColumnModelItemT],
):
    ...

ColumnWiseTableWithColNamesNoConvertModel

Bases: Model[dict[str, JsonScalarColumnModel]]

Source code in src/omnipy/components/tables/models.py

class ColumnWiseTableWithColNamesNoConvertModel(Model[dict[str, JsonScalarColumnModel]]):
    ...

ConcatByAddArrayAdapterModel

Bases: Model[_ColumnT], Generic[_ColumnT, _ItemT]

Source code in src/omnipy/components/tables/models.py

class ConcatByAddArrayAdapterModel(Model[_ColumnT], Generic[_ColumnT, _ItemT]):
    _allowed_special_methods = frozenset({
        '__len__',
        '__length_hint__',
        '__getitem__',
        '__setitem__',
        '__delitem__',
        '__iter__',
        '__reversed__',
        '__contains__',
        '__copy__',
        '__deepcopy__',
        '__str__',
        '__format__',
    })

    @classmethod
    def _get_special_methods_info_dict(cls) -> dict[str, MethodInfo]:
        special_methods = super()._get_special_methods_info_dict()
        return {
            method_name: special_methods[method_name]
            for method_name in cls._allowed_special_methods
            if method_name in special_methods
        }

    @staticmethod
    def _as_item_list(values: _ColumnT) -> list[_ItemT]:
        return list(values)

    @classmethod
    def _concat_column_values(cls, left: _ColumnT, right: _ColumnT) -> _ColumnT:
        print(f'{cls.__name__}: default concat fallback via list conversion')
        return cls(cls._as_item_list(left) + cls._as_item_list(right)).content

    def __add__(self, other: object):
        other_content = self.__class__(other).content
        return self.__class__(self._concat_column_values(self.content, other_content))

CsvTableModel

Bases: Chain3[SplitToLinesModel, SplitLinesToColumnsByCommaModel, RowWiseTableFirstRowAsColNamesModel], PrintableTable

Source code in src/omnipy/components/tables/models.py

class CsvTableModel(
        Chain3[
            SplitToLinesModel,
            SplitLinesToColumnsByCommaModel,
            RowWiseTableFirstRowAsColNamesModel,
        ],
        PrintableTable,
):
    ...

CsvTableOfPydanticRecordsModel

Bases: Chain3[SplitToLinesModel, SplitLinesToColumnsByCommaModel, Model[list[PydanticRecordModel[_PydRecordT]]]], PrintableTable, Generic[_PydRecordT]

Source code in src/omnipy/components/tables/models.py

class CsvTableOfPydanticRecordsModel(
        Chain3[SplitToLinesModel,
               SplitLinesToColumnsByCommaModel,
               Model[list[PydanticRecordModel[_PydRecordT]]]],
        PrintableTable,
        Generic[_PydRecordT],
):
    ...

IterRow

Bases: Mapping[str, _ColumnModelT], Generic[_ColumnModelT, _ColumnModelItemT]

METHOD	DESCRIPTION
__init__

ATTRIBUTE	DESCRIPTION
row_number	TYPE: int

Source code in src/omnipy/components/tables/models.py

class IterRow(Mapping[str, _ColumnModelT], Generic[_ColumnModelT, _ColumnModelItemT]):
    def __init__(self, content: Mapping[str, _ColumnModelT]) -> None:
        self._content = {
            key: content[key].content if is_model_instance(content[key]) else content[key]
            for key in content
        }
        self.row_number: int = -1

    @override
    def __getitem__(self, key) -> _ColumnModelItemT:  # type: ignore[override]
        if key not in self._content:
            raise KeyError(f'Key {key} is not a column name')
        if self.row_number == -1:
            raise KeyError('Row number not set')
        return self._content[key][self.row_number]

    @override
    def __iter__(self) -> Iterator[str]:
        return iter(self._content)

    def __len__(self) -> int:
        return len(self._content)

    def __eq__(self, other: object) -> bool:
        if isinstance(other, IterRow):
            return self._content == other._content and self.row_number == other.row_number
        return False

    def __str__(self) -> str:
        arg_str = ', '.join(f'{repr(key)}: {repr(self[key])}' for key in self._content.keys())
        return f'{self.__class__.__name__}({arg_str})'

row_number instance-attribute

row_number: int = -1

__init__

__init__(content: Mapping[str, _ColumnModelT]) -> None

Source code in src/omnipy/components/tables/models.py

def __init__(self, content: Mapping[str, _ColumnModelT]) -> None:
    self._content = {
        key: content[key].content if is_model_instance(content[key]) else content[key]
        for key in content
    }
    self.row_number: int = -1

IteratingPydanticRecordsModel

Bases: _ColumnWiseTableWithColNamesMixin, PydanticRecordModelBase[_PydBaseModelT, _ColumnWiseTableModelT, RowWiseTableModel], PrintableTable, Generic[_PydBaseModelT, _ColumnWiseTableModelT, _ColumnModelT, _ColumnModelItemT]

Source code in src/omnipy/components/tables/models.py

class IteratingPydanticRecordsModel(
        _ColumnWiseTableWithColNamesMixin,
        PydanticRecordModelBase[
            _PydBaseModelT,
            _ColumnWiseTableModelT,
            RowWiseTableModel,
        ],
        PrintableTable,
        Generic[_PydBaseModelT, _ColumnWiseTableModelT, _ColumnModelT, _ColumnModelItemT],
):
    @classmethod
    def _validate_over_all_rows(
        cls,
        input_model: ColumnWiseTableWithColNamesModel | RowWiseTableModel,
        output_model: ColumnWiseTableWithColNamesModel,
        pyd_model: type[pyd.BaseModel],
        to_row_dict_func: Callable[[_ColumnModelT], dict[str, JsonScalar]] | None = None,
    ):
        new_cols = set()

        def _init_col(
            content: dict[str, _ColumnModelT],
            pyd_model: type[pyd.BaseModel],
            key: str,
        ) -> None:
            if key in new_cols:
                return

            col_len: int = len(input_model)
            # if key not in content:
            content[key] = [pyd_model.__fields__[key].get_default()] * col_len
            # else:
            #     assert len(content[key]) == col_len, \
            #         (f'Incorrect number of rows in {key} column: '
            #          f'{len(content[key])} != {col_len}')
            #     content[key] = [_ for _ in content[key]]

            new_cols.add(key)

        content = output_model.content
        # content = output_model
        for key, field in pyd_model.__fields__.items():
            if field.required and key not in content:
                _init_col(content, pyd_model, key)

        for i, row in enumerate(input_model):
            row_dict = to_row_dict_func(row) if to_row_dict_func else row
            values, _fields_set, error = pyd.validate_model(pyd_model, row_dict)
            if error:
                raise error

            for key, validated_val in values.items():
                if key not in content:
                    _init_col(content, pyd_model, key)

                if key in new_cols:
                    is_changed_val = True
                else:
                    is_changed_val = row_dict[key] != validated_val
                    # else:
                    #     is_changed_val = True

                if is_changed_val:
                    try:
                        _, prepared_val = prepare_value_for_validation_if_dataset_or_model(
                            validated_val)
                        content[key][i] = prepared_val
                    except (TypeError, AssertionError, ValueError, ValidationError):
                        _init_col(content, pyd_model, key)
                        content[key][i] = prepared_val
        output_model.validate_content()

    @override
    @classmethod
    def _validate_record_model_with_col_names(
        cls,
        pyd_model: type[pyd.BaseModel],
        data: ColumnWiseTableWithColNamesModel,
        header_names: tuple[str, ...],
    ) -> pyd.BaseModel | ColumnWiseTableWithColNamesModel:
        cls._validate_over_all_rows(input_model=data, output_model=data, pyd_model=pyd_model)
        return data

    @override
    @classmethod
    def _validate_record_model_without_col_names(
        cls,
        pyd_model: type[pyd.BaseModel],
        data: RowWiseTableModel,
        data_type: type[_DataWithColNamesModelT],
        header_names: tuple[str, ...],
    ) -> pyd.BaseModel | ColumnWiseTableWithColNamesModel:
        output = data_type()
        cls._validate_over_all_rows(
            input_model=data,
            output_model=output,
            pyd_model=pyd_model,
            to_row_dict_func=lambda row: dict(zip(header_names, row)),
        )
        return output

JsonMaxLevel1ColumnModel

Bases: ColumnModel[list[JsonMaxLevel1Types], JsonMaxLevel1Types]

Source code in src/omnipy/components/tables/models.py

class JsonMaxLevel1ColumnModel(ColumnModel[list[JsonMaxLevel1Types], JsonMaxLevel1Types]):
    ...

JsonMaxLevel1ColumnWiseTableWithColNamesModel

Bases: ColumnWiseTableWithColNamesModel[JsonMaxLevel1ColumnModel, JsonMaxLevel1Types]

Source code in src/omnipy/components/tables/models.py

class JsonMaxLevel1ColumnWiseTableWithColNamesModel(ColumnWiseTableWithColNamesModel[
        JsonMaxLevel1ColumnModel,
        JsonMaxLevel1Types,
]):
    ...

JsonMaxLevel2ColumnModel

Bases: ColumnModel[list[JsonMaxLevel2Types], JsonMaxLevel2Types]

Source code in src/omnipy/components/tables/models.py

class JsonMaxLevel2ColumnModel(ColumnModel[list[JsonMaxLevel2Types], JsonMaxLevel2Types]):
    ...

JsonMaxLevel2ColumnWiseTableWithColNamesModel

Bases: ColumnWiseTableWithColNamesModel[JsonMaxLevel2ColumnModel, JsonMaxLevel2Types]

Source code in src/omnipy/components/tables/models.py

class JsonMaxLevel2ColumnWiseTableWithColNamesModel(ColumnWiseTableWithColNamesModel[
        JsonMaxLevel2ColumnModel,
        JsonMaxLevel2Types,
]):
    ...

JsonScalarColumnModel

Bases: ColumnModel[list[JsonScalar], JsonScalar]

Source code in src/omnipy/components/tables/models.py

class JsonScalarColumnModel(ColumnModel[list[JsonScalar], JsonScalar]):
    ...

JsonScalarColumnWiseTableWithColNamesModel

Bases: ColumnWiseTableWithColNamesModel[JsonScalarColumnModel, JsonScalar]

Source code in src/omnipy/components/tables/models.py

class JsonScalarColumnWiseTableWithColNamesModel(ColumnWiseTableWithColNamesModel[
        JsonScalarColumnModel,
        JsonScalar,
]):
    ...

PrintableTable

Source code in src/omnipy/components/tables/models.py

class PrintableTable:
    ...

PydanticRecordModel

Bases: PydanticRecordModelBase[_PydBaseModelT, dict[str, JsonScalar], list[JsonScalar]], Generic[_PydBaseModelT]

Source code in src/omnipy/components/tables/models.py

class PydanticRecordModel(
        PydanticRecordModelBase[
            _PydBaseModelT,
            dict[str, JsonScalar],
            list[JsonScalar],
        ],
        Generic[_PydBaseModelT],
):
    @override
    @classmethod
    def _validate_record_model_with_col_names(
        cls,
        pyd_model: type[pyd.BaseModel],
        data: dict[str, JsonScalar],
        header_names: tuple[str, ...],
    ) -> pyd.BaseModel | dict[str, JsonScalar]:
        return pyd_model(**data)

    @override
    @classmethod
    def _validate_record_model_without_col_names(
        cls,
        pyd_model: type[pyd.BaseModel],
        data: list[JsonScalar],
        data_type: type[_DataWithColNamesModelT],
        header_names: tuple[str, ...],
    ) -> pyd.BaseModel | dict[str, JsonScalar]:
        return pyd_model(**dict(zip(header_names, data)))

PydanticRecordModelBase

Bases: Model[TypeVarStore[_PydBaseModelT] | _DataWithColNamesModelT | _DataWithoutColNamesModelT], Generic[_PydBaseModelT, _DataWithColNamesModelT, _DataWithoutColNamesModelT]

Source code in src/omnipy/components/tables/models.py

class PydanticRecordModelBase(
        Model[TypeVarStore[_PydBaseModelT] | _DataWithColNamesModelT | _DataWithoutColNamesModelT],
        Generic[_PydBaseModelT, _DataWithColNamesModelT, _DataWithoutColNamesModelT],
        metaclass=PydanticRecordModelMetaclass,
):
    @classmethod
    @abstractmethod
    def _validate_record_model_with_col_names(
        cls,
        pyd_model: type[pyd.BaseModel],
        data: _DataWithColNamesModelT,
        header_names: tuple[str, ...],
    ) -> pyd.BaseModel | _DataWithColNamesModelT:
        ...

    @classmethod
    @abstractmethod
    def _validate_record_model_without_col_names(
        cls,
        pyd_model: type[pyd.BaseModel],
        data: _DataWithoutColNamesModelT,
        output_type: type[_DataWithColNamesModelT],
        header_names: tuple[str, ...],
    ) -> pyd.BaseModel | _DataWithColNamesModelT:
        ...

    @override
    @classmethod
    def _parse_data(  # type: ignore[override]
        cls,
        data: _DataWithColNamesModelT | _DataWithoutColNamesModelT,
    ) -> pyd.BaseModel | _DataWithColNamesModelT:
        pyd_model, data_with_col_names_type, header_names, num_required_fields = cls.header_info
        content = data.content if is_model_instance(data) else data
        if isinstance(content, dict) or is_pure_pydantic_model(content):
            # cls._validate_and_set_value(data)
            data_with_col_names = cast(_DataWithColNamesModelT, data)
            return cls._validate_record_model_with_col_names(
                pyd_model,
                data_with_col_names,
                header_names,
            )
            return pyd_model(**data)

        if isinstance(content, list):
            if isinstance(data, RowWiseTableModel):
                num_data_cols = len(content[0]) if len(content) > 0 else 0
            else:
                num_data_cols = len(content)

            assert len(header_names) >= num_data_cols >= num_required_fields, \
                (f'Incorrect number of data elements: '
                 f'{len(header_names)} >= {num_data_cols} >= {num_required_fields}')

            data_without_col_names = cast(_DataWithoutColNamesModelT, data)
            return cls._validate_record_model_without_col_names(
                pyd_model,
                data_without_col_names,
                data_with_col_names_type,
                header_names,
            )
            return pyd_model(**dict(zip(header_names, data)))
        else:
            raise TypeError(f'Unsupported data type: {type(data)}')

PydanticRecordModelMetaclass

Bases: ModelMetaclass

METHOD	DESCRIPTION
__init__

ATTRIBUTE	DESCRIPTION
header_info	TYPE: _HeaderInfo

Source code in src/omnipy/components/tables/models.py

class PydanticRecordModelMetaclass(ModelMetaclass):
    def __init__(cls, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        cls._header_info: _HeaderInfo | None = None

    @property
    def header_info(cls) -> _HeaderInfo:
        if cls._header_info is None:
            type_args = get_args(cast(type[Model], cls).outer_type(with_args=True))
            type_var_store = type_args[0]
            pydantic_model = get_args(type_var_store)[0]
            if hasattr(pydantic_model, '__pydantic_model__'):
                pydantic_model = pydantic_model.__pydantic_model__

            data_with_col_names_type = type_args[1]
            headers = pydantic_model.__fields__

            num_required_fields = -1
            for i, header_field in enumerate(headers.values()):
                if not header_field.required:
                    if num_required_fields == -1:
                        num_required_fields = i
                    continue
                elif num_required_fields != -1 and i > num_required_fields:
                    raise ValueError('Required fields must not come after optional fields')

            if num_required_fields == -1:
                num_required_fields = len(headers)

            cls._header_info = _HeaderInfo(
                pydantic_model,
                data_with_col_names_type,
                tuple(headers.keys()),
                num_required_fields,
            )
        return cls._header_info

header_info property

header_info: _HeaderInfo

__init__

__init__(*args, **kwargs) -> None

Source code in src/omnipy/components/tables/models.py

def __init__(cls, *args, **kwargs) -> None:
    super().__init__(*args, **kwargs)
    cls._header_info: _HeaderInfo | None = None

RowWiseTableFirstRowAsColNamesModel

Bases: _RowWiseColNamesMixin, _RowWiseTableFirstRowAsColNamesModel, PrintableTable

Source code in src/omnipy/components/tables/models.py

class RowWiseTableFirstRowAsColNamesModel(
        _RowWiseColNamesMixin,
        _RowWiseTableFirstRowAsColNamesModel,
        PrintableTable,
):
    ...

RowWiseTableModel

Bases: Model[list[list[JsonScalar]]]

Source code in src/omnipy/components/tables/models.py

class RowWiseTableModel(Model[list[list[JsonScalar]]]):
    ...

RowWiseTableWithColNamesModel

Bases: _RowWiseColNamesMixin, _RowWiseTableWithColNamesModel, PrintableTable

Source code in src/omnipy/components/tables/models.py

class RowWiseTableWithColNamesModel(
        _RowWiseColNamesMixin,
        _RowWiseTableWithColNamesModel,
        PrintableTable,
):
    ...

RowWiseTableWithColNamesNoConvertModel

Bases: Model[list[dict[str, JsonScalar]]]

Source code in src/omnipy/components/tables/models.py

class RowWiseTableWithColNamesNoConvertModel(Model[list[dict[str, JsonScalar]]]):
    ...

TableOfPydanticRecordsModel

Bases: Chain3[SplitToLinesModel, SplitLinesToColumnsModel, Model[list[PydanticRecordModel[_PydRecordT]]]], PrintableTable, Generic[_PydRecordT]

Source code in src/omnipy/components/tables/models.py

class TableOfPydanticRecordsModel(
        Chain3[SplitToLinesModel,
               SplitLinesToColumnsModel,
               Model[list[PydanticRecordModel[_PydRecordT]]]],
        PrintableTable,
        Generic[_PydRecordT],
):
    ...

TsvTableModel

Bases: Chain3[SplitToLinesModel, SplitLinesToColumnsModel, RowWiseTableFirstRowAsColNamesModel], PrintableTable

Source code in src/omnipy/components/tables/models.py

class TsvTableModel(
        Chain3[
            SplitToLinesModel,
            SplitLinesToColumnsModel,
            RowWiseTableFirstRowAsColNamesModel,
        ],
        PrintableTable,
):
    ...