fix: nested source init arguments (#13)

examples/extentions/import_local_files/custom_file.py

@@ -8,7 +8,7 @@ def __getitem__(self, key: str) -> str:
 
 
 class CustomDestination(secret_transfer_core.AbstractDestination):
-    def set(self, key: str, value: secret_transfer_types.BaseArgumentType):
+    def set(self, key: str, value: secret_transfer_types.Literal):
         print(f"CustomDestination.set called with key: {key} and value: {value}")
 
     def clean(self, key: str) -> None:

secret_transfer/core/__init__.py

@@ -30,12 +30,21 @@
     TransferRegistry,
     TransferSettings,
 )
+from .types import (
+    BaseInitArgumentType,
+    BaseRawArgumentType,
+    InitArgumentType,
+    Proxy,
+)
 
 __all__ = [
     "AbstractCollection",
     "AbstractDestination",
     "AbstractSource",
     "AbstractTransfer",
+    "BaseInitArgumentType",
+    "BaseInitArgumentType",
+    "BaseRawArgumentType",
     "CollectionClassProxy",
     "CollectionClassSettings",
     "CollectionProxy",
@@ -46,6 +55,8 @@
     "DestinationProxy",
     "DestinationRegistry",
     "DestinationSettings",
+    "InitArgumentType",
+    "Proxy",
     "SourceClassProxy",
     "SourceClassSettings",
     "SourceProxy",

secret_transfer/core/base/base.py

@@ -2,20 +2,16 @@
 
 import typing_extensions
 
-import secret_transfer.utils.pydantic as pydantic_utils
-import secret_transfer.utils.types as utils_types
+import secret_transfer.core.types as core_types
+import secret_transfer.protocol as protocol
 
 
 class BaseResource:
-    _arguments_model: typing.Optional[type[pydantic_utils.BaseModel]] = None
+    ValidationError = protocol.BaseResourceProtocol.ValidationError
 
     @classmethod
-    def parse_init_arguments(cls, **arguments: utils_types.BaseArgumentType) -> typing.Mapping[str, typing.Any]:
-        if cls._arguments_model is None:
-            return arguments
-
-        model = cls._arguments_model.model_validate(arguments)
-        return model.shallow_model_dump()
+    def parse_init_arguments(cls, **arguments: core_types.InitArgumentType) -> typing.Mapping[str, typing.Any]:
+        return arguments
 
     @classmethod
     def get_default_instances(cls) -> typing.Mapping[str, typing_extensions.Self]:

secret_transfer/core/base/proxy.py

@@ -2,8 +2,7 @@
 import importlib
 import typing
 
-import lazy_object_proxy
-
+import secret_transfer.core.types as core_types
 import secret_transfer.protocol as protocol
 import secret_transfer.utils.types as utils_types
 
@@ -43,7 +42,7 @@ def from_settings(cls, settings: ResourceClassSettings[protocol.ResourceType]) -
         def factory() -> type[protocol.ResourceType]:
             return cls._get_target(settings=settings)
 
-        return typing.cast(type[protocol.ResourceType], lazy_object_proxy.Proxy(factory))
+        return typing.cast(type[protocol.ResourceType], core_types.Proxy(factory))
 
     @classmethod
     def _get_target(cls, settings: ResourceClassSettings[protocol.ResourceType]) -> type[protocol.ResourceType]:
@@ -66,18 +65,10 @@ def _get_target(cls, settings: ResourceClassSettings[protocol.ResourceType]) ->
 
 @typing.runtime_checkable
 class GettableResourceProtocol(typing.Protocol):
-    def __getitem__(self, key: str) -> utils_types.LiteralArgumentType:
+    def __getitem__(self, key: str) -> utils_types.Literal:
         ...
 
 
-InitArgumentsType = typing.Union[
-    utils_types.LiteralArgumentType,
-    protocol.BaseResourceProtocol,
-    dict[str, "InitArgumentsType"],
-    list["InitArgumentsType"],
-]
-
-
 @dataclasses.dataclass
 class ResourceArgument:
     type: str
@@ -183,7 +174,7 @@ def from_settings(cls, settings: ResourceSettings[protocol.ResourceType]) -> pro
         def factory() -> protocol.ResourceType:
             return cls._get_target(settings=settings)
 
-        return typing.cast(protocol.ResourceType, lazy_object_proxy.Proxy(factory))
+        return typing.cast(protocol.ResourceType, core_types.Proxy(factory))
 
     @classmethod
     def _get_target(cls, settings: ResourceSettings[protocol.ResourceType]) -> protocol.ResourceType:
@@ -211,7 +202,7 @@ def _get_resource_argument_value(
         cls,
         settings: ResourceSettings[protocol.ResourceType],
         raw_argument: str,
-    ) -> typing.Union[protocol.BaseResourceProtocol, utils_types.LiteralArgumentType]:
+    ) -> core_types.BaseInitArgumentType:
         try:
             argument = _parse_resource_argument(argument=raw_argument)
         except ParseResourceArgumentError as exc:
@@ -248,20 +239,20 @@ def _get_resource_argument_value(
                 f"Resource({resource_name!r}) does not implement [__getitem__(str) -> str | int | float | bool] method"
             )
 
-        def factory() -> utils_types.LiteralArgumentType:
+        def factory() -> utils_types.Literal:
             try:
                 return resource[key]
             except Exception as exc:
                 raise cls.ResourceError(f"Key({key!r}) is not found in resource({resource_name!r})") from exc
 
-        return typing.cast(utils_types.LiteralArgumentType, lazy_object_proxy.Proxy(factory))
+        return core_types.Proxy(factory)
 
     @classmethod
     def _get_literal_argument_value(
         cls,
         settings: ResourceSettings[protocol.ResourceType],
-        raw_argument: utils_types.LiteralArgumentType,
-    ) -> InitArgumentsType:
+        raw_argument: core_types.BaseRawArgumentType,
+    ) -> core_types.BaseInitArgumentType:
         if isinstance(raw_argument, str) and raw_argument.startswith("$"):
             return cls._get_resource_argument_value(settings=settings, raw_argument=raw_argument)
 
@@ -271,23 +262,25 @@ def _get_literal_argument_value(
     def _get_argument_value(
         cls,
         settings: ResourceSettings[protocol.ResourceType],
-        raw_argument: utils_types.BaseArgumentType,
-    ) -> InitArgumentsType:
-        if isinstance(raw_argument, dict):
-            return {
-                key: cls._get_argument_value(settings=settings, raw_argument=value)
-                for key, value in raw_argument.items()
-            }
-
-        if isinstance(raw_argument, list):
+        raw_argument: core_types.RawArgumentType,
+    ) -> core_types.InitArgumentType:
+        if raw_argument is None:
+            return None
+
+        if isinstance(raw_argument, utils_types.Literals):
+            return cls._get_literal_argument_value(settings=settings, raw_argument=raw_argument)
+
+        if isinstance(raw_argument, typing.Sequence):
             return [cls._get_argument_value(settings=settings, raw_argument=value) for value in raw_argument]
 
-        return cls._get_literal_argument_value(settings=settings, raw_argument=raw_argument)
+        return {
+            key: cls._get_argument_value(settings=settings, raw_argument=value) for key, value in raw_argument.items()
+        }
 
     @classmethod
     def _get_arguments(
         cls, settings: ResourceSettings[protocol.ResourceType]
-    ) -> typing.Mapping[str, InitArgumentsType]:
+    ) -> typing.Mapping[str, core_types.InitArgumentType]:
         return {
             key: cls._get_argument_value(settings=settings, raw_argument=value)
             for key, value in settings.raw_arguments.items()

secret_transfer/core/collection/base.py

@@ -20,13 +20,13 @@ def register(self, name: str) -> None:
         CollectionRegistry.register_instance(name, self)
 
     @abc.abstractmethod
-    def __getitem__(self, key: str) -> utils_types.LiteralArgumentType:
+    def __getitem__(self, key: str) -> utils_types.Literal:
         ...
 
     @abc.abstractmethod
     def __iter__(self) -> typing.Iterator[str]:
         ...
 
     @abc.abstractmethod
-    def items(self) -> typing.Iterator[tuple[str, utils_types.LiteralArgumentType]]:
+    def items(self) -> typing.Iterator[tuple[str, utils_types.Literal]]:
         ...

secret_transfer/core/destination/base.py

@@ -16,7 +16,7 @@ def register(self, name: str) -> None:
         DestinationRegistry.register_instance(name, self)
 
     @abc.abstractmethod
-    def set(self, key: str, value: utils_types.BaseArgumentType) -> None:
+    def set(self, key: str, value: utils_types.Literal) -> None:
         ...
 
     @abc.abstractmethod

secret_transfer/core/source/base.py

@@ -19,5 +19,5 @@ def register(self, name: str) -> None:
         SourceRegistry.register_instance(name, self)
 
     @abc.abstractmethod
-    def __getitem__(self, key: str) -> utils_types.LiteralArgumentType:
+    def __getitem__(self, key: str) -> utils_types.Literal:
         ...

secret_transfer/core/types.py

@@ -0,0 +1,31 @@
+import typing
+
+import lazy_object_proxy  # pyright: ignore[reportMissingTypeStubs]
+
+import secret_transfer.protocol as protocol
+import secret_transfer.utils.types as utils_types
+
+BaseRawArgumentType = utils_types.Literal
+
+
+RawArgumentType = typing.Union[
+    utils_types.Literal,
+    typing.Mapping[str, "BaseRawArgumentType"],
+    typing.Sequence["BaseRawArgumentType"],
+]
+
+
+class Proxy(lazy_object_proxy.Proxy):  # pyright: ignore[reportUntypedBaseClass]
+    ...
+
+
+BaseInitArgumentType = typing.Union[
+    utils_types.Literal,
+    protocol.BaseResourceProtocol,
+]
+
+InitArgumentType = typing.Union[
+    BaseInitArgumentType,
+    typing.Mapping[str, "InitArgumentType"],
+    typing.Sequence["InitArgumentType"],
+]

secret_transfer/default/collection/default.py

@@ -1,11 +1,10 @@
 import dataclasses
 import typing
 
-import pydantic
+import typing_extensions
 
 import secret_transfer.core as core
 import secret_transfer.protocol as protocol
-import secret_transfer.utils.pydantic as pydantic_utils
 import secret_transfer.utils.types as utils_types
 
 
@@ -14,11 +13,22 @@ class DefaultCollectionItem:
     source: protocol.SourceProtocol
     key: typing.Optional[str] = None
 
+    @classmethod
+    def from_init_arguments(cls, data: core.InitArgumentType) -> typing_extensions.Self:
+        if not isinstance(data, typing.Mapping):
+            raise ValueError("Should be a mapping.")
+
+        source = data.get("source")
+        if source is None:
+            raise ValueError("Should have a source.")
+        if not isinstance(source, protocol.SourceProtocol):
+            raise ValueError("Should have a source of type SourceProtocol.")
 
-class Arguments(pydantic_utils.BaseModel):
-    items: typing.Mapping[str, DefaultCollectionItem]
+        key = data.get("key")
+        if key is not None and not isinstance(key, str):
+            raise ValueError("Should have a key of type str or None.")
 
-    model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
+        return cls(source=source, key=key)
 
 
 class DefaultCollection(core.AbstractCollection):
@@ -28,11 +38,18 @@ def __init__(self, **items: DefaultCollectionItem):
         self._items = items
 
     @classmethod
-    def parse_init_arguments(cls, **arguments: utils_types.BaseArgumentType) -> typing.Mapping[str, typing.Any]:
-        model = Arguments.model_validate({"items": arguments})
-        return model.items
+    def parse_init_arguments(cls, **items: core.InitArgumentType) -> typing.Mapping[str, DefaultCollectionItem]:
+        result: dict[str, DefaultCollectionItem] = {}
+
+        for key, value in items.items():
+            try:
+                result[key] = DefaultCollectionItem.from_init_arguments(value)
+            except ValueError as exc:
+                raise cls.ValidationError(f"Error parsing item {key}: {exc}") from exc
+
+        return result
 
-    def __getitem__(self, key: str) -> utils_types.LiteralArgumentType:
+    def __getitem__(self, key: str) -> utils_types.Literal:
         try:
             item = self._items[key]
         except KeyError as exc:
@@ -47,6 +64,6 @@ def __getitem__(self, key: str) -> utils_types.LiteralArgumentType:
     def __iter__(self) -> typing.Iterator[str]:
         return iter(self._items)
 
-    def items(self) -> typing.Iterator[tuple[str, utils_types.LiteralArgumentType]]:
+    def items(self) -> typing.Iterator[tuple[str, utils_types.Literal]]:
         for key in self:
             yield key, self[key]

secret_transfer/default/destination/bash_export.py

@@ -7,7 +7,7 @@
 
 
 class BashExportDestination(core.AbstractDestination):
-    def set(self, key: str, value: utils_types.BaseArgumentType) -> None:
+    def set(self, key: str, value: utils_types.Literal) -> None:
         print(f"export {key}={value}")
 
     def clean(self, key: str) -> None:

secret_transfer/default/destination/env.py

@@ -1,11 +1,13 @@
 import os
 
+import typing_extensions
+
 import secret_transfer.core as core
 import secret_transfer.utils.types as utils_types
 
 
 class EnvDestination(core.AbstractDestination):
-    def set(self, key: str, value: utils_types.BaseArgumentType) -> None:
+    def set(self, key: str, value: utils_types.Literal) -> None:
         os.environ[key] = str(value)
 
     def clean(self, key: str) -> None:
@@ -15,5 +17,5 @@ def clean(self, key: str) -> None:
             pass
 
     @classmethod
-    def get_default_instances(cls) -> dict[str, "EnvDestination"]:
+    def get_default_instances(cls) -> dict[str, typing_extensions.Self]:
         return {"env": cls()}

secret_transfer/default/destination/gh_cli_secrets.py

@@ -1,20 +1,11 @@
 import secret_transfer.core as core
 import secret_transfer.utils.cli as cli_utils
-import secret_transfer.utils.pydantic as pydantic_utils
 import secret_transfer.utils.types as utils_types
 
 DEFAULT_BASE_URL = "https://github.com"
 
 
-class Arguments(pydantic_utils.BaseModel):
-    repo_name: str
-    owner_name: str
-    base_url: str = DEFAULT_BASE_URL
-
-
 class GithubCliSecretsDestination(core.AbstractDestination):
-    _arguments_model = Arguments
-
     def __init__(self, *, repo_name: str, owner_name: str, base_url: str = DEFAULT_BASE_URL):
         self._repo_name = repo_name
         self._owner_name = owner_name
@@ -24,7 +15,7 @@ def __init__(self, *, repo_name: str, owner_name: str, base_url: str = DEFAULT_B
     def _repo_url(self) -> str:
         return f"{self._base_url}/{self._owner_name}/{self._repo_name}"
 
-    def set(self, key: str, value: utils_types.BaseArgumentType) -> None:
+    def set(self, key: str, value: utils_types.Literal) -> None:
         cli_utils.GH.secret.set(key=key, value=str(value), repo_url=self._repo_url)
 
     def clean(self, key: str) -> None: