refactor: provider base

secrets_env/collect.py

@@ -5,7 +5,7 @@
 
 if typing.TYPE_CHECKING:
     from secrets_env.config0.parser import Config
-    from secrets_env.provider import ProviderBase, RequestSpec
+    from secrets_env.provider import Provider, RequestSpec
 
 logger = logging.getLogger(__name__)
 
@@ -37,17 +37,17 @@ def read_values(config: Config) -> dict[str, str]:
     return output_values
 
 
-def read1(provider: ProviderBase, name: str, spec: RequestSpec) -> str | None:
+def read1(provider: Provider, name: str, spec: RequestSpec) -> str | None:
     """Read single value.
 
     This function wraps :py:meth:`secrets_env.provider.ProviderBase.get` and
     captures all exceptions.
     """
     import secrets_env.exceptions
-    from secrets_env.provider import ProviderBase
+    from secrets_env.provider import Provider
 
     # type checking
-    if not isinstance(provider, ProviderBase):
+    if not isinstance(provider, Provider):
         raise TypeError(
             f'Expected "provider" to be a credential provider class, '
             f"got {type(provider).__name__}"

secrets_env/config0/parser.py

@@ -1,16 +1,17 @@
+from __future__ import annotations
+
 import itertools
 import logging
 import re
 import typing
-from typing import Any, Dict, Iterator, List, Optional, Tuple, TypedDict
+from typing import Any, Iterator, TypedDict
 
 import secrets_env.exceptions
 import secrets_env.providers
-from secrets_env.provider import RequestSpec
 from secrets_env.utils import ensure_dict, ensure_str
 
 if typing.TYPE_CHECKING:
-    from secrets_env.provider import ProviderBase
+    from secrets_env.provider import Provider, RequestSpec
 
 DEFAULT_PROVIDER_NAME = "main"
 
@@ -29,11 +30,11 @@ class Request(TypedDict):
 class Config(TypedDict):
     """The parsed configurations."""
 
-    providers: Dict[str, "ProviderBase"]
-    requests: List[Request]
+    providers: dict[str, Provider]
+    requests: list[Request]
 
 
-def parse_config(data: dict) -> Optional[Config]:
+def parse_config(data: dict) -> Config | None:
     """Parse and validate configs, build it into structured object."""
     requests = get_requests(data)
     if not requests:
@@ -48,11 +49,11 @@ def parse_config(data: dict) -> Optional[Config]:
     return Config(providers=providers, requests=requests)
 
 
-def get_providers(data: dict) -> Dict[str, "ProviderBase"]:
+def get_providers(data: dict) -> dict[str, Provider]:
     sections = list(extract_sources(data))
     logger.debug("%d raw provider configs extracted", len(sections))
 
-    providers: Dict[str, "ProviderBase"] = {}
+    providers: dict[str, Provider] = {}
     for data in sections:
         result = parse_source_item(data)
         if not result:
@@ -75,7 +76,7 @@ def get_providers(data: dict) -> Dict[str, "ProviderBase"]:
     return providers
 
 
-def extract_sources(data: dict) -> Iterator[Dict[str, Any]]:
+def extract_sources(data: dict) -> Iterator[dict[str, Any]]:
     """Extracts both "source(s)" section and ensure the output is list of dict"""
     for item in itertools.chain(
         get_list(data, "source"),
@@ -97,7 +98,7 @@ def get_list(data: dict, key: str) -> Iterator[dict]:
         logger.warning("Found invalid value in field <mark>%s</mark>", key)
 
 
-def parse_source_item(config: dict) -> Optional[Tuple[str, "ProviderBase"]]:
+def parse_source_item(config: dict) -> tuple[str, Provider] | None:
     # check name
     name = config.get("name") or DEFAULT_PROVIDER_NAME
     name, ok = ensure_str("source.name", name)
@@ -117,7 +118,7 @@ def parse_source_item(config: dict) -> Optional[Tuple[str, "ProviderBase"]]:
     return name, provider
 
 
-def get_requests(data: dict) -> List[Request]:
+def get_requests(data: dict) -> list[Request]:
     # accept both keyword `secret(s)`
     raw = {}
 

secrets_env/provider.py

@@ -6,53 +6,46 @@
 from __future__ import annotations
 
 import abc
-import sys
-import typing
-from typing import Dict, Union
+from typing import ClassVar, Union
 
-if typing.TYPE_CHECKING and sys.version_info >= (3, 10):
-    from typing import TypeAlias
+from pydantic import BaseModel
 
+RequestSpec = Union[dict[str, str], str]
+""":py:class:`RequestSpec` represents a path spec to read the value.
 
-RequestSpec: TypeAlias = Union[Dict[str, str], str]
-""":py:class:`RequestSpec` represents a secret spec (name/path) to be loaded.
+It should be a :py:class:`dict` in most cases; or :py:class:`str` if this
+provider accepts shortcut.
 """
 
 
-class ProviderBase(abc.ABC):
-    """Abstract base class for secret provider. All secret provider must implement
+class Provider(BaseModel, abc.ABC):
+    """Abstract base class for secret provider. All provider must implement
     this interface.
     """
 
-    @property
-    @abc.abstractmethod
-    def type(self) -> str:
-        """Provider name."""
+    type: ClassVar[str]
 
     @abc.abstractmethod
     def get(self, spec: RequestSpec) -> str:
-        """Get secret value.
+        """Get secret.
 
         Parameters
         ----------
-        spec : dict | str
-            Raw input from config file.
-
-            It should be :py:class:`dict` in most cases; or :py:class:`str` if
-            this provider accepts shortcut.
+        path : dict | str
+            Raw input from config file for reading the secret value.
 
         Return
         ------
-        The secret value.
+        The value
 
         Raises
         ------
-        ConfigError
-            The path dict is malformed.
-        ValueNotFound
-            The path dict is correct but the secret not exists.
-
-        Note
-        ----
-        Key ``source`` is preserved in ``spec`` dictionary.
+        ValidationError
+            If the input format is invalid.
+        UnsupportedError
+            When this operation is not supported.
+        AuthenticationError
+            Failed during authentication.
+        LookupError
+            If the secret is not found.
         """

secrets_env/providers/__init__.py

@@ -1,33 +1,50 @@
-import typing
+from __future__ import annotations
 
-import secrets_env.exceptions
+import logging
+import typing
 
 if typing.TYPE_CHECKING:
-    from secrets_env.provider import ProviderBase
+    from secrets_env.provider import Provider
 
 DEFAULT_PROVIDER = "vault"
 
+logger = logging.getLogger(__name__)
+
+
+def get_provider(config: dict) -> Provider:
+    """
+    Returns a provider instance based on the configuration.
+
+    Raises
+    ------
+    ValueError
+        If the provider type is not recognized.
+    ValidationError
+        If the provider configuration is invalid.
+    """
+    type_ = config.get("type")
+    if not type_:
+        type_ = DEFAULT_PROVIDER
+        logger.warning("Provider type unspecified, using default: %s", type_)
 
-def get_provider(data: dict) -> "ProviderBase":
-    type_ = data.get("type", DEFAULT_PROVIDER)
-    type_lower = type_.lower()
+    itype = type_.lower()
 
     # fmt: off
-    if type_lower == "null":
-        from . import null
-        return null.get_provider(type_, data)
-    if type_lower == "plain":
-        from . import plain
-        return plain.get_provider(type_, data)
-    if type_lower == "teleport":
-        from . import teleport
-        return teleport.get_provider(type_, data)
-    if type_lower == "vault":
-        from . import vault
-        return vault.get_provider(type_, data)
-    if type_lower.startswith("teleport+"):
-        from . import teleport
-        return teleport.get_adapted_provider(type_, data)
+    if itype == "null":
+        from secrets_env.providers.null import NullProvider
+        return NullProvider.model_validate(config)
+    if itype == "plain":
+        from secrets_env.providers.plain import PlainTextProvider
+        return PlainTextProvider.model_validate(config)
+    if itype == "teleport":
+        from secrets_env.providers.teleport import TeleportProvider
+        return TeleportProvider.model_validate(config)
+    if itype == "teleport+vault":
+        logger.error('"teleport+vault provider is not yet implemented')
+        raise NotImplementedError
+    if itype == "vault":
+        from secrets_env.providers.vault import VaultKvProvider
+        return VaultKvProvider.model_validate(config)
     # fmt: on
 
-    raise secrets_env.exceptions.ConfigError("Unknown provider type {}", type_)
+    raise ValueError(f"Unknown provider type {type_}")

secrets_env/providers/null.py

@@ -1,22 +1,18 @@
+from __future__ import annotations
+
 import typing
 
-from secrets_env.provider import ProviderBase
+from secrets_env.provider import Provider
 
 if typing.TYPE_CHECKING:
     from secrets_env.provider import RequestSpec
 
 
-class NullProvider(ProviderBase):
+class NullProvider(Provider):
     """A provider that always returns empty string. This provider is preserved
     for debugging."""
 
-    @property
-    def type(self) -> str:
-        return "null"
+    type = "null"
 
-    def get(self, spec: "RequestSpec") -> str:
+    def get(self, spec: RequestSpec) -> str:
         return ""
-
-
-def get_provider(type_: str, data: dict) -> NullProvider:
-    return NullProvider()

secrets_env/providers/plain.py

@@ -1,30 +1,21 @@
+from __future__ import annotations
+
 import typing
 
-from secrets_env.provider import ProviderBase
+from secrets_env.provider import Provider
 
 if typing.TYPE_CHECKING:
     from secrets_env.provider import RequestSpec
 
 
-class PlainTextProvider(ProviderBase):
+class PlainTextProvider(Provider):
     """Plain text provider returns text that is copied directly from the
     configuration file."""
 
-    @property
-    def type(self) -> str:
-        return "plain"
+    type = "plain"
 
-    def get(self, spec: "RequestSpec") -> str:
+    def get(self, spec: RequestSpec) -> str:
         if isinstance(spec, str):
-            value = spec
+            return spec
         elif isinstance(spec, dict):
-            value = spec.get("value")
-        else:
-            raise TypeError(
-                f'Expected "spec" to be a string or dict, got {type(spec).__name__}'
-            )
-        return value or ""
-
-
-def get_provider(type_: str, data: dict) -> PlainTextProvider:
-    return PlainTextProvider()
+            return spec.get("value") or ""