Integrated second reimplementation of parametrized models with SplitT…

…oItemsModelNew as example model

src/omnipy/__init__.py

@@ -11,6 +11,7 @@
 from omnipy.compute.task import Task, TaskTemplate
 from omnipy.data.dataset import Dataset, ListOfParamModelDataset, MultiModelDataset, ParamDataset
 from omnipy.data.model import ListOfParamModel, Model, ParamModel
+from omnipy.data.param import conf, ConfHolder, ConfT, ParamModelMixin
 from omnipy.hub.runtime import runtime
 from omnipy.modules.general.models import (Chain2,
                                            Chain3,
@@ -117,7 +118,6 @@
                                        JoinColumnsToLinesModel,
                                        JoinItemsModel,
                                        JoinLinesModel,
-                                       Params,
                                        SplitLinesToColumnsModel,
                                        SplitToItemsModel,
                                        SplitToItemsModelNew,

src/omnipy/data/helpers.py

@@ -11,14 +11,6 @@ def __init__(self, t: T) -> None:
         raise ValueError()
 
 
-class Params:
-    params: dict[str, object]
-
-    def __class_getitem__(cls, params: dict[str, object] = {}) -> 'Params':
-        cls.params = params
-        return cls
-
-
 class YesNoMaybe(IntEnum):
     NO = 0
     YES = 1

src/omnipy/data/param.py

@@ -0,0 +1,43 @@
+from dataclasses import dataclass
+from typing import Callable, Generic, get_args, ParamSpec, Protocol
+
+from typing_extensions import TypeVar
+
+ParamsT = TypeVar('ParamsT')
+ParamsP = ParamSpec('ParamsP')
+
+
+class Conf(Protocol[ParamsT]):
+    settings: ParamsT
+
+
+def conf(param_cls: Callable[ParamsP, ParamsT]) -> Callable[ParamsP, type[Conf[ParamsT]]]:
+    def _conf(*args: ParamsP.args, **kwargs: ParamsP.kwargs) -> type[Conf[ParamsT]]:
+        # Factory for new _Conf classes. Needed to allow the classes to have individual settings
+        class _Conf(Conf[ParamsT]):
+            settings = param_cls(**kwargs)
+
+        return _Conf
+
+    return _conf
+
+
+class ParamModelMixin(Generic[ParamsT]):
+    @dataclass
+    class Params:
+        ...
+
+    @classmethod
+    def _get_conf_settings(cls) -> ParamsT:
+        conf_t = get_args(get_args(cls.full_type())[-1])[0]
+        if isinstance(conf_t, TypeVar):
+            conf_t = conf(cls.Params)()
+        return conf_t.settings
+
+
+ConfT = TypeVar('ConfT', bound=type[Conf])
+
+
+class ConfHolder(Generic[ConfT]):
+    def __init__(self, conf: ConfT) -> None:
+        raise ValueError()

src/omnipy/modules/raw/models.py

@@ -1,12 +1,11 @@
+from dataclasses import dataclass
 import os
-from typing import Generic, get_args
+from typing import Generic
 
-from pydantic import BaseModel
 from pydantic.generics import _generic_types_cache
-from typing_extensions import TypeVar
 
-from omnipy.data.helpers import Params, TypeVarStore
 from omnipy.data.model import DataWithParams, ListOfParamModel, Model, ParamModel
+from omnipy.data.param import ConfHolder, ConfT, ParamModelMixin
 
 
 class BytesModel(ParamModel[str | bytes, str]):
@@ -57,29 +56,29 @@ def _parse_data(cls, data: list[str] | str) -> str:
         return os.linesep.join(data)
 
 
-ParamsT = TypeVar('ParamsT', default=Params)
-
-
-class SplitToItemsModelNew(Model[list[str] | str | TypeVarStore[ParamsT]], Generic[ParamsT]):
-    class Params(BaseModel):
+class SplitToItemsModelNew(
+        Model[list[str] | str | ConfHolder[ConfT]],
+        ParamModelMixin['SplitToItemsModelNew.Params'],
+        Generic[ConfT],
+):
+    @dataclass
+    class Params:
         strip: bool = True
         strip_chars: str | None = None
-        delimiter: str = ','
+        delimiter: str = '\t'
 
     @classmethod
-    def _parse_data(cls, data: list[str] | str | TypeVarStore[ParamsT]) -> list[str]:
+    def _parse_data(cls, data: list[str] | str) -> list[str]:
         if isinstance(data, list):
             return data
 
-        params_typevar = get_args(get_args(cls.outer_type(with_args=True))[-1])[0]
-        params_dict = Model[params_typevar]().contents.params
-        params = cls.Params(**params_dict)
+        settings = cls._get_conf_settings()
 
-        if params.strip:
-            data = data.strip(params.strip_chars)
+        if settings.strip:
+            data = data.strip(settings.strip_chars)
 
-        items = data.split(params.delimiter)
-        data = [item.strip(params.strip_chars) for item in items] if params.strip else items
+        items = data.split(settings.delimiter)
+        data = [item.strip(settings.strip_chars) for item in items] if settings.strip else items
         return data
 
 

tests/modules/raw/test_models.py

@@ -6,8 +6,8 @@
 import pytest
 
 from omnipy.api.protocols.public.hub import IsRuntime
-from omnipy.data.helpers import Params
 from omnipy.data.model import Model
+from omnipy.data.param import conf
 from omnipy.modules.raw.models import (BytesModel,
                                        JoinColumnsToLinesModel,
                                        JoinItemsModel,
@@ -18,7 +18,7 @@
                                        SplitToLinesModel,
                                        StrModel)
 
-from ...helpers.protocols import AssertModelOrValFunc  # type: ignore[misc]
+from ...helpers.protocols import AssertModelOrValFunc
 
 
 def test_bytes_model():
@@ -91,10 +91,9 @@ def test_split_to_and_join_lines_model(
         'abhorred in my imagination it is! my gorge rises at',
         'it. Here hung those lips that I have kissed I know'
     ]
-    assert_model_if_dyn_conv_else_val(
-        last_lines[-1],  # type: ignore[index]
-        str,
-        'now, to mock your own grinning? quite chap-fallen.')
+    assert_model_if_dyn_conv_else_val(last_lines[-1],
+                                      str,
+                                      'now, to mock your own grinning? quite chap-fallen.')
 
     joined_lines = JoinLinesModel(last_lines[0:2])
     assert joined_lines.contents == dedent("""\
@@ -160,32 +159,35 @@ def test_split_to_and_join_items_new_model(
 
     data_tab = Model[str](raw_data_tab) if use_str_model else raw_data_tab
 
-    items_stripped_tab = SplitToItemsModelNew(data_tab)
+    items_stripped_tab: SplitToItemsModelNew = SplitToItemsModelNew(data_tab)
     assert items_stripped_tab.contents == ['abc', 'def', 'ghi', 'jkl']
 
-    assert_model_or_val(dyn_convert, items_stripped_tab[1], str, 'def')  # type: ignore[index]
+    assert_model_if_dyn_conv_else_val(items_stripped_tab[1], str, 'def')  # type: ignore[index]
     assert items_stripped_tab[-2:].contents == ['ghi', 'jkl']  # type: ignore[index]
 
-    items_unstripped_tab = SplitToItemsModelNew[Params[{'strip': False}]](data_tab)
+    split_conf = conf(SplitToItemsModelNew.Params)
+    items_unstripped_tab = \
+        SplitToItemsModelNew[split_conf(strip=False)](data_tab)  # type: ignore[operator, misc]
     assert items_unstripped_tab.contents == ['abc', ' def ', 'ghi', 'jkl']
-    assert_model_or_val(dyn_convert, items_unstripped_tab[1], str, ' def ')  # type: ignore[index]
-    assert items_unstripped_tab[-2:].contents == ['ghi', 'jkl']  # type: ignore[index]
+    assert_model_if_dyn_conv_else_val(items_unstripped_tab[1], str, ' def ')
+    assert items_unstripped_tab[-2:].contents == ['ghi', 'jkl']
 
     raw_data_comma = 'abc, def, ghi, jkl'
 
     data_comma = Model[str](raw_data_comma) if use_str_model else raw_data_comma
 
-    items_stripped_comma = SplitToItemsModelNew[Params[{'delimiter': ','}]](data_comma)
+    items_stripped_comma = \
+        SplitToItemsModelNew[split_conf(delimiter=',')](data_comma)  # type: ignore[operator, misc]
+
     assert items_stripped_comma.contents == ['abc', 'def', 'ghi', 'jkl']
-    assert_model_or_val(dyn_convert, items_stripped_comma[1], str, 'def')  # type: ignore[index]
-    assert items_stripped_comma[-2:].contents == ['ghi', 'jkl']  # type: ignore[index]
+    assert_model_if_dyn_conv_else_val(items_stripped_comma[1], str, 'def')
+    assert items_stripped_comma[-2:].contents == ['ghi', 'jkl']
 
     tab_joined_items = JoinItemsModel(items_stripped_tab[1:3])  # type: ignore[index]
     assert tab_joined_items.contents == 'def\tghi'
     assert tab_joined_items[1:-1].contents == 'ef\tgh'  # type: ignore[index]
 
-    comma_space_joined_items = JoinItemsModel(
-        items_stripped_comma[1:3], delimiter=', ')  # type: ignore[index]
+    comma_space_joined_items = JoinItemsModel(items_stripped_comma[1:3], delimiter=', ')
     assert comma_space_joined_items.contents == 'def, ghi'
     assert comma_space_joined_items[1:-1].contents == 'ef, gh'  # type: ignore[index]
 
@@ -212,8 +214,10 @@ def test_split_lines_to_columns_and_join_columns_to_lines_model(
 
     cols_unstripped_tab = SplitLinesToColumnsModel(data_tab, strip=False)
     assert cols_unstripped_tab[0].contents == ['abc', ' def ', 'ghi', ' jkl']  # type: ignore[index]
-    assert_model_if_dyn_conv_else_val(cols_unstripped_tab[0][1], str,
-                                      ' def ')  # type: ignore[index]
+    assert_model_if_dyn_conv_else_val(
+        cols_unstripped_tab[0][1],  # type: ignore[index]
+        str,
+        ' def ')
     assert cols_unstripped_tab[1:2].contents == [  # type: ignore[index]
         SplitToItemsModel('mno\t pqr\tstu\t vwx', strip=False)
     ]