Inject

Author: kmagusiak

README.md

@@ -122,6 +122,29 @@ You can read values or passwords from files, by using the template
 or, more securely, read the file in the code
 `alphaconf.get('secret_file', Path).read_text().strip()`.
 
+### Inject parameters
+
+We can inject default values to functions from the configuration.
+Either one by one, where we can map a factory function or a configuration key.
+Or inject all automatically base on the parameter name.
+
+```python
+from alphaconf.inject import inject, inject_auto
+
+@inject('name', 'application.name')
+@inject_auto(ignore={'name'})
+def main(name: str, example=None):
+    pass
+
+# similar to
+def main(name: str=None, example=None):
+    if name is None:
+        name = alphaconf.get('application.name', str)
+    if example is None:
+        example = alphaconf.get('example', default=example)
+    ...
+```
+
 ### Invoke integration
 
 Just add the lines below to parameterize invoke.

alphaconf/inject.py

@@ -1,60 +1,92 @@
+import functools
 import inspect
-from dataclasses import dataclass
-from typing import Callable, Dict, TypeVar
+from typing import Any, Callable, Dict, Optional, Union
 
 import alphaconf
 
-R = TypeVar('R')
-
-
-@dataclass
-class InjectArgument:
-    name: str
-    verify: bool = False
-    # rtype: type = None  # TODO add type transformer
-
-    def get_value(self, type_spec, required):
-        get_args: dict = {'key': self.name}
-        if type_spec:
-            get_args['type'] = type_spec
-        if not required:
-            get_args['default'] = None
-        value = alphaconf.get(**get_args)
-        return value
-
-
-class Injector:
-    args: Dict[str, InjectArgument]
-    prefix: str
-
-    def __init__(self, prefix: str = ""):
-        if prefix and not prefix.endswith("."):
-            prefix += "."
-        self.prefix = prefix
-        self.args = {}
-
-    def inject(self, name: str, optional, type, resolver):
-        pass
-
-    def decorate(self, func: Callable[..., R]) -> Callable[[], R]:
-        signature = inspect.signature(func)
-
-        def call():
-            args = {}  # TODO {**self.values}
-            for name, iarg in self.args.items():
-                param = signature.parameters.get(name, None)
-                if not param:
-                    if iarg.verify:
-                        raise TypeError("Missing argument", name)
-                    continue
-                arg_type = None
-                if param.annotation is not param.empty and isinstance(param.annotation, type):
-                    arg_type = param.annotation
-                required = param.default is param.empty
-                value = iarg.get_value(arg_type, required)
-                if value is None and not required:
-                    continue
-                args[name] = value
-            return func(**args)
-
-        return call
+from .internal.type_resolvers import type_from_annotation
+
+__all__ = ["inject", "inject_auto"]
+
+
+class ParamDefaultsFunction:
+    """Function wrapper that injects default parameters"""
+
+    _arg_factory: Dict[str, Callable[[], Any]]
+
+    def __init__(self, func: Callable):
+        self.func = func
+        self.signature = inspect.signature(func)
+        self._arg_factory = {}
+
+    def bind(self, name: str, factory: Callable[[], Any]):
+        self._arg_factory[name] = factory
+
+    def __call__(self, *a, **kw):
+        args = self.signature.bind_partial(*a, **kw).arguments
+        kw.update(
+            {name: factory() for name, factory in self._arg_factory.items() if name not in args}
+        )
+        return self.func(*a, **kw)
+
+    @staticmethod
+    def wrap(func) -> "ParamDefaultsFunction":
+        if isinstance(func, ParamDefaultsFunction):
+            return func
+        return functools.wraps(func)(ParamDefaultsFunction(func))
+
+
+def getter(
+    key: str, ktype: Optional[type] = None, *, param: Optional[inspect.Parameter] = None
+) -> Callable[[], Any]:
+    """Factory function that calls alphaconf.get
+
+    The parameter from the signature can be given to extract the type to cast to
+    and whether the configuration value is optional.
+
+    :param key: The key using in alphaconf.get
+    :param ktype: Type to cast to
+    :param param: The parameter object from the signature
+    """
+    if ktype is None and param and (ptype := param.annotation) is not param.empty:
+        ktype = next(type_from_annotation(ptype), None)
+    if param is not None and param.default is not param.empty:
+        xparam = param
+        return (
+            lambda: xparam.default
+            if (value := alphaconf.get(key, ktype, default=None)) is None
+            and xparam.default is not xparam.empty
+            else value
+        )
+    return lambda: alphaconf.get(key, ktype)
+
+
+def inject(name: str, factory: Union[None, str, Callable[[], Any]]):
+    """Inject an argument to a function from a factory or alphaconf"""
+
+    def do_inject(func):
+        f = ParamDefaultsFunction.wrap(func)
+        if isinstance(factory, str) or factory is None:
+            b = getter(factory or name, param=f.signature.parameters[name])
+        else:
+            b = factory
+        f.bind(name, b)
+        return f
+
+    return do_inject
+
+
+def inject_auto(*, prefix: str = "", ignore: set = set()):
+    """Inject automatically all paramters"""
+    if prefix and not prefix.endswith("."):
+        prefix += "."
+
+    def do_inject(func):
+        f = ParamDefaultsFunction.wrap(func)
+        for name, param in f.signature.parameters.items():
+            if name in ignore:
+                continue
+            f.bind(name, getter(prefix + name, param=param))
+        return f
+
+    return do_inject

alphaconf/internal/configuration.py

@@ -1,6 +1,5 @@
 import copy
 import os
-import typing
 import warnings
 from enum import Enum
 from typing import (
@@ -19,7 +18,7 @@
 
 from omegaconf import Container, DictConfig, OmegaConf
 
-from .type_resolvers import convert_to_type, pydantic
+from .type_resolvers import convert_to_type, pydantic, type_from_annotation
 
 T = TypeVar('T')
 
@@ -92,14 +91,14 @@ def get(self, key: Union[str, Type], type=None, *, default=raise_on_missing):
         )
         if value is raise_on_missing:
             if default is raise_on_missing:
-                raise ValueError(f"No value for: {key}")
+                raise KeyError(f"No value for: {key}")
             return default
         # check the returned type and convert when necessary
         if type is not None and isinstance(value, type):
             return value
         if isinstance(value, Container):
             value = OmegaConf.to_object(value)
-        if type is not None and default is not None:
+        if type is not None and value is not default:
             value = convert_to_type(value, type)
         return value
 
@@ -110,12 +109,12 @@ def __get_type(self, key: Type, *, default=raise_on_missing):
         key_str = self.__type_path.get(key)
         if key_str is None:
             if default is raise_on_missing:
-                raise ValueError(f"Key not found for type {key}")
+                raise KeyError(f"Key not found for type {key}")
             return default
         try:
             value = self.get(key_str, key)
             self.__type_value = value
-        except ValueError:
+        except KeyError:
             if default is raise_on_missing:
                 raise
             value = default
@@ -161,7 +160,7 @@ def setup_configuration(
         else:
             created_config = self.__prepare_config(conf, path=prefix)
             if not isinstance(created_config, DictConfig):
-                raise ValueError("Failed to convert to a DictConfig")
+                raise TypeError("Failed to convert to a DictConfig")
             config = created_config
         # add prefix and merge
         if prefix:
@@ -221,7 +220,7 @@ def __prepare_dictconfig(
         sub_configs = []
         for k, v in obj.items_ex(resolve=False):
             if not isinstance(k, str):
-                raise ValueError("Expecting only str instances in dict")
+                raise TypeError("Expecting only str instances in dict")
             if recursive:
                 v = self.__prepare_config(v, path + k + ".")
             if '.' in k:
@@ -252,9 +251,6 @@ def __prepare_pydantic(self, obj, path):
             # pydantic instance, prepare helpers
             self.__prepare_pydantic(type(obj), path)
             return obj.model_dump(mode="json")
-        # parse typing recursively for documentation
-        for t in typing.get_args(obj):
-            self.__prepare_pydantic(t, path)
         # check if not a type
         if not isinstance(obj, type):
             return obj
@@ -279,8 +275,9 @@ def __prepare_pydantic(self, obj, path):
                     from alphaconf import SECRET_MASKS
 
                     SECRET_MASKS.append(lambda s: s == path)
-                elif check_type and field.annotation:
-                    self.__prepare_pydantic(field.annotation, path + k + ".")
+                elif check_type:
+                    for ftype in type_from_annotation(field.annotation):
+                        self.__prepare_pydantic(ftype, path + k + ".")
             return defaults
         return None
 

alphaconf/internal/type_resolvers.py

@@ -1,4 +1,5 @@
 import datetime
+import typing
 from pathlib import Path
 
 from omegaconf import OmegaConf
@@ -61,3 +62,12 @@ def convert_to_type(value, type):
     if pydantic:
         return pydantic.TypeAdapter(type).validate_python(value)
     return type(value)
+
+
+def type_from_annotation(annotation) -> typing.Generator[type, None, None]:
+    """Given an annotation (optional), figure out the types"""
+    if isinstance(annotation, type) and annotation is not type(None):
+        yield annotation
+    else:
+        for t in typing.get_args(annotation):
+            yield from type_from_annotation(t)

tests/test_alphaconf.py

@@ -90,7 +90,7 @@ def test_setup_configuration():
 
 
 def test_setup_configuration_invalid():
-    with pytest.raises(ValueError):
+    with pytest.raises(TypeError):
         # invalid configuration (must be non-empty)
         alphaconf.setup_configuration(None)
 
@@ -132,8 +132,8 @@ def test_app_environ(application):
     )
     application.setup_configuration(load_dotenv=False, env_prefixes=True)
     config = application.configuration
-    with pytest.raises(ValueError):
-        # XXX should not be loaded
+    with pytest.raises(KeyError):
+        # prefix with underscore only should be loaded
         config.get('xxx')
     assert config.get('testmyenv.x') == 'overwrite'
     assert config.get('testmyenv.y') == 'new'

tests/test_configuration.py

@@ -48,6 +48,7 @@ def test_cast(config):
     # cast bool into int
     assert config.get('b') is True
     assert config.get('b', int) == 1
+    assert config.get('b', int, default=None) == 1
     # cast Path
     assert isinstance(config.get('home', Path), Path)
 
@@ -71,7 +72,7 @@ def test_select_empty(config):
 
 def test_select_required(config):
     assert config.get('z', default=None) is None
-    with pytest.raises(ValueError):
+    with pytest.raises(KeyError):
         print(config.get('z'))
     assert config.get('z', default='a') == 'a'
 
@@ -80,7 +81,7 @@ def test_select_required_incomplete(config_req):
     # when we have a default, return it
     assert config_req.get('req', default='def') == 'def'
     # when required, raise missing
-    with pytest.raises(ValueError):
+    with pytest.raises(KeyError):
         print(config_req.get('req'))