Move from dataclass to pydantic models for config classes (tmp)

src/omnipy/api/protocols/private/compute/job_creator.py

@@ -1,13 +1,14 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Optional, Protocol
+from typing import Optional, Protocol, runtime_checkable
 
 from omnipy.api.protocols.private.compute.mixins import IsNestedContext
 from omnipy.api.protocols.private.engine import IsEngine
 from omnipy.api.protocols.public.config import IsJobConfig
 
 
+@runtime_checkable
 class IsJobConfigHolder(Protocol):
     """"""
     @property
@@ -25,6 +26,7 @@ def set_engine(self, engine: IsEngine) -> None:
         ...
 
 
+@runtime_checkable
 class IsJobCreator(IsNestedContext, IsJobConfigHolder, Protocol):
     """"""
     @property

src/omnipy/api/protocols/private/log.py

@@ -2,7 +2,7 @@
 
 from datetime import datetime
 from logging import INFO, Logger
-from typing import Optional, Protocol, Tuple
+from typing import Optional, Protocol, runtime_checkable, Tuple
 
 from omnipy.api.enums import RunState
 from omnipy.api.protocols.private.compute.mixins import IsUniquelyNamedJob
@@ -18,6 +18,7 @@ def log(self, log_msg: str, level: int = INFO, datetime_obj: Optional[datetime]
         ...
 
 
+@runtime_checkable
 class IsRunStateRegistry(Protocol):
     """"""
     def __init__(self) -> None:

src/omnipy/api/protocols/public/config.py

@@ -1,33 +1,38 @@
 from __future__ import annotations
 
-from typing import Protocol
+from typing import Protocol, runtime_checkable
 
 from omnipy.api.enums import (ConfigOutputStorageProtocolOptions,
                               ConfigPersistOutputsOptions,
                               ConfigRestoreOutputsOptions)
 from omnipy.api.types import LocaleType
 
 
+@runtime_checkable
 class IsEngineConfig(Protocol):
     """"""
     ...
 
 
+@runtime_checkable
 class IsLocalRunnerConfig(IsEngineConfig, Protocol):
     """"""
     ...
 
 
+@runtime_checkable
 class IsPrefectEngineConfig(IsEngineConfig, Protocol):
     """"""
     use_cached_results: int = False
 
 
+@runtime_checkable
 class IsJobConfig(Protocol):
     """"""
     output_storage: IsOutputStorage
 
 
+@runtime_checkable
 class IsRootLogConfig(Protocol):
     """"""
     log_format_str: str
@@ -41,6 +46,7 @@ class IsRootLogConfig(Protocol):
     file_log_dir_path: str
 
 
+@runtime_checkable
 class IsOutputStorageBase(Protocol):
     persist_data_dir_path: str
 

src/omnipy/api/protocols/public/hub.py

@@ -2,7 +2,7 @@
 
 import logging
 from logging.handlers import TimedRotatingFileHandler
-from typing import Optional, Protocol
+from typing import Optional, Protocol, runtime_checkable
 
 from omnipy.api.enums import EngineChoice
 from omnipy.api.protocols.private.compute.job_creator import IsJobConfigHolder
@@ -14,6 +14,7 @@
                                                 IsRootLogConfig)
 
 
+@runtime_checkable
 class IsRootLogObjects(Protocol):
     """"""
     formatter: Optional[logging.Formatter] = None
@@ -25,6 +26,7 @@ def set_config(self, config: IsRootLogConfig) -> None:
         ...
 
 
+@runtime_checkable
 class IsRuntimeConfig(Protocol):
     """"""
     job: IsJobConfig
@@ -45,6 +47,7 @@ def __init__(
         ...
 
 
+@runtime_checkable
 class IsRuntimeObjects(Protocol):
     """"""
 

src/omnipy/config/engine.py

@@ -1,16 +1,18 @@
-from dataclasses import dataclass
+# from dataclasses import dataclass
 
+from pydantic import BaseModel
 
-@dataclass
-class EngineConfig:
+
+# @dataclass
+class EngineConfig(BaseModel):
     ...
 
 
-@dataclass
+# @dataclass
 class LocalRunnerConfig(EngineConfig):
     ...
 
 
-@dataclass
+# @dataclass
 class PrefectEngineConfig(EngineConfig):
     use_cached_results: int = False

src/omnipy/config/job.py

@@ -1,8 +1,10 @@
-from dataclasses import dataclass, field
+# from dataclasses import field
 import os
 # from datetime import datetime
 from pathlib import Path
 
+from pydantic import BaseModel, Field
+
 from omnipy.api.enums import (ConfigOutputStorageProtocolOptions,
                               ConfigPersistOutputsOptions,
                               ConfigRestoreOutputsOptions)
@@ -17,12 +19,12 @@ def _get_persist_data_dir_path() -> str:
     return str(Path.cwd().joinpath(Path('outputs')))
 
 
-@dataclass
+# @dataclass
 class LocalOutputStorage:
-    persist_data_dir_path: str = field(default_factory=_get_persist_data_dir_path)
+    persist_data_dir_path: str = Field(default_factory=_get_persist_data_dir_path)
 
 
-@dataclass
+# @dataclass
 class S3OutputStorage:
     persist_data_dir_path: str = os.path.join('omnipy', 'outputs')
     endpoint_url: str = ""
@@ -31,17 +33,17 @@ class S3OutputStorage:
     secret_key: str = ""
 
 
-@dataclass
+# @dataclass
 class OutputStorage:
     persist_outputs: ConfigPersistOutputsOptions = \
         ConfigPersistOutputsOptions.ENABLE_FLOW_AND_TASK_OUTPUTS
     restore_outputs: ConfigRestoreOutputsOptions = \
         ConfigRestoreOutputsOptions.DISABLED
     protocol: ConfigOutputStorageProtocolOptions = ConfigOutputStorageProtocolOptions.LOCAL
-    local: IsLocalOutputStorage = field(default_factory=LocalOutputStorage)
-    s3: IsS3OutputStorage = field(default_factory=S3OutputStorage)
+    local: IsLocalOutputStorage = Field(default_factory=LocalOutputStorage)
+    s3: IsS3OutputStorage = Field(default_factory=S3OutputStorage)
 
 
-@dataclass
-class JobConfig:
-    output_storage: IsOutputStorage = field(default_factory=OutputStorage)
+# @dataclass
+class JobConfig(BaseModel):
+    output_storage: IsOutputStorage = Field(default_factory=OutputStorage)

src/omnipy/config/root_log.py

@@ -1,17 +1,19 @@
-from dataclasses import dataclass, field
+# from dataclasses import field
 import locale as pkg_locale
 import logging
 from pathlib import Path
 
+from pydantic import BaseModel, Field
+
 from omnipy.api.types import LocaleType
 
 
 def _get_log_dir_path() -> str:
     return str(Path.cwd().joinpath(Path('logs')))
 
 
-@dataclass
-class RootLogConfig:
+# @dataclass
+class RootLogConfig(BaseModel):
     log_format_str: str = '{engine} {asctime} - {levelname}: {message} [{name}]'
     locale: LocaleType = pkg_locale.getlocale()
     log_to_stdout: bool = True
@@ -20,4 +22,4 @@ class RootLogConfig:
     stdout_log_min_level: int = logging.INFO
     stderr_log_min_level: int = logging.ERROR
     file_log_min_level: int = logging.WARNING
-    file_log_dir_path: str = field(default_factory=_get_log_dir_path)
+    file_log_dir_path: str = Field(default_factory=_get_log_dir_path)

src/omnipy/hub/entry.py

@@ -1,13 +1,15 @@
-from dataclasses import dataclass, field
+# from dataclasses import dataclass, field
 from typing import Optional
 
+from pydantic import Field
+
 from omnipy.api.protocols.public.hub import IsRuntime
 from omnipy.util.publisher import DataPublisher
 
 
-@dataclass
+# @dataclass
 class RuntimeEntryPublisher(DataPublisher):
-    _back: Optional[IsRuntime] = field(default=None, init=False, repr=False)
+    _back: Optional[IsRuntime] = Field(default=None, init=False, repr=False)
 
     def __setattr__(self, key, value):
         super().__setattr__(key, value)

src/omnipy/hub/root_log.py

@@ -13,12 +13,12 @@
 from omnipy.util.helpers import get_datetime_format
 
 
-@dataclass
+# @dataclass
 class RootLogConfigEntryPublisher(RootLogConfig, RuntimeEntryPublisher):
     ...
 
 
-@dataclass
+# @dataclass
 class RootLogObjects:
     _config: IsRootLogConfig = field(
         init=False, repr=False, default_factory=RootLogConfigEntryPublisher)

src/omnipy/hub/runtime.py

@@ -1,6 +1,8 @@
-from dataclasses import dataclass, field
+# from dataclasses import dataclass, field
 from typing import Any
 
+from pydantic import Field
+
 from omnipy.api.enums import EngineChoice
 from omnipy.api.protocols.private.compute.job_creator import IsJobConfigHolder
 from omnipy.api.protocols.private.engine import IsEngine
@@ -29,28 +31,31 @@ def _job_creator_factory():
     return JobBase.job_creator
 
 
-@dataclass
+# @dataclass
 class RuntimeConfig(RuntimeEntryPublisher):
-    job: IsJobConfig = field(default_factory=JobConfig)
+    job: IsJobConfig = Field(default_factory=JobConfig)
     engine: EngineChoice = EngineChoice.LOCAL
-    local: IsLocalRunnerConfig = field(default_factory=LocalRunnerConfig)
-    prefect: IsPrefectEngineConfig = field(default_factory=PrefectEngineConfig)
-    root_log: IsRootLogConfig = field(default_factory=RootLogConfigEntryPublisher)
+    local: IsLocalRunnerConfig = Field(default_factory=LocalRunnerConfig)
+    prefect: IsPrefectEngineConfig = Field(default_factory=PrefectEngineConfig)
+    root_log: IsRootLogConfig = Field(default_factory=RootLogConfigEntryPublisher)
 
 
-@dataclass
+# @dataclass
 class RuntimeObjects(RuntimeEntryPublisher):
-    job_creator: IsJobConfigHolder = field(default_factory=_job_creator_factory)
-    local: IsEngine = field(default_factory=LocalRunner)
-    prefect: IsEngine = field(default_factory=PrefectEngine)
-    registry: IsRunStateRegistry = field(default_factory=RunStateRegistry)
-    root_log: IsRootLogObjects = field(default_factory=RootLogObjects)
+    job_creator: IsJobConfigHolder = Field(default_factory=_job_creator_factory)
+    local: IsEngine = Field(default_factory=LocalRunner)
+    prefect: IsEngine = Field(default_factory=PrefectEngine)
+    registry: IsRunStateRegistry = Field(default_factory=RunStateRegistry)
+    root_log: IsRootLogObjects = Field(default_factory=RootLogObjects)
+
+
+# TODO: Add automatic parsing of config values when setting to string values
 
 
-@dataclass
+# @dataclass
 class Runtime(DataPublisher):
-    config: IsRuntimeConfig = field(default_factory=RuntimeConfig)
-    objects: IsRuntimeObjects = field(default_factory=RuntimeObjects)
+    config: IsRuntimeConfig = Field(default_factory=RuntimeConfig)
+    objects: IsRuntimeObjects = Field(default_factory=RuntimeObjects)
 
     def __post_init__(self):
         super().__init__()

src/omnipy/util/publisher.py

@@ -1,16 +1,21 @@
 from collections import defaultdict
-from dataclasses import dataclass, field
+# from dataclasses import dataclass, field
 from typing import Any, Callable, DefaultDict, List
 
+from pydantic import BaseModel, Field
+
 
 def _subscribers_factory():
     return defaultdict(list)
 
 
-@dataclass
-class DataPublisher:
+# @dataclass
+class DataPublisher(BaseModel):
     _subscriptions: DefaultDict[str, List[Callable[[Any], None]]] = \
-        field(default_factory=_subscribers_factory, init=False, repr=False)
+        Field(default_factory=_subscribers_factory, init=False, repr=False)
+
+    class Config:
+        arbitrary_types_allowed = True
 
     def subscribe(self, config_item: str, callback_fun: Callable[[Any], None]):
         if not hasattr(self, config_item):