Add support for inherent signature. (#6)

Author: victorywys

tests/test_registry.py

@@ -3,6 +3,7 @@
 import pytest
 from utilsd.config import ClassConfig, Registry, RegistryConfig, SubclassConfig, configclass
 from utilsd.config.type_def import TypeDef
+from utilsd.config.exception import ValidationError
 from tests.assets.import_class import BaseBar
 
 
@@ -84,6 +85,11 @@ def test_registry_config():
     assert config.m.a == 1
     assert config.m.type() == Converter1
     assert config.m.build().a == 1
+
+    # test overwrite during build() call
+    assert config.m.build(a=2).a == 2
+    with pytest.raises(RuntimeError):
+        config.m.build(c=3)
     assert isinstance(config.m.build(), Converter1)
 
 
@@ -145,3 +151,97 @@ def test_registry():
 
     Converters.register_module(module=Converter2)
     assert len(Converters) == 2
+
+
+class TestInhReg(metaclass=Registry, name='TestInh'):
+    pass
+
+
+@TestInhReg.register_module()
+class InhBase():
+    def __init__(self, a: int, b: int, **kwargs):
+        self.a = a
+        self.b = b
+        self.uncollected = kwargs
+        
+
+@TestInhReg.register_module(inherit=True)
+class InhChild1(InhBase):
+    def __init__(self, c: int, d: int, **kwargs):
+        super().__init__(**kwargs)
+        self.c = c
+        self.d = d
+
+
+@TestInhReg.register_module()
+class InhChild2(InhBase):
+    def __init__(self, c: int, d: int, **kwargs):
+        super().__init__(**kwargs)
+        self.c = c
+        self.d = d
+
+
+@configclass
+class InhRegCfg:
+    m: RegistryConfig[TestInhReg]
+
+
+def test_superclass_registry():
+    assert len(TestInhReg) == 3
+    assert "InhChild1" in TestInhReg
+    assert "InhChildNotDefined" not in TestInhReg
+
+    # when inherit is set True, the module will look back to its super class for areas
+    config = TypeDef.load(
+        InhRegCfg, dict(m=dict(type="InhChild1", a=1, b=2, c=3, d=4))
+    )
+    child1 = config.m.build()
+    assert child1.a == 1
+    assert child1.c == 3
+    assert config.m.build(e=5).uncollected['e'] == 5
+
+    with pytest.raises(ValidationError):
+        # when inherit is set False (default), all the keys must be specifed in the param list of __init__
+        config = TypeDef.load(
+            InhRegCfg, dict(m=dict(type="InhChild2", a=1, b=2, c=3, d=4))
+        )
+
+    with pytest.raises(ValidationError):
+        # when inherit is set True, use of positional arguments is banned to remove possible confusions
+        @TestInhReg.register_module(inherit=True)
+        class InhChildwithPosArgs(InhBase):
+            def __init__(self, c: int, d: int, *args, **kwargs):
+                super().__init__(**kwargs)
+                self.c = c
+                self.d = d
+
+        config = TypeDef.load(
+            InhRegCfg, dict(m=dict(type="InhChildwithPosArgs", a=1, b=2, c=3, d=4))
+        )
+
+    TestInhReg.unregister_module("InhChildwithPosArgs")
+
+    # won't raise TypeError with *args when inherit is set False (default) TODO: should this use also be banned?
+    @TestInhReg.register_module()
+    class InhChildwithPosArgs(InhBase):
+        def __init__(self, a: int, b: int, c: int, d: int, *args, **kwargs):
+            super().__init__(a=a, b=b)
+            self.c = c
+            self.d = d
+            self.uncollected_args = args
+
+    config = TypeDef.load(
+        InhRegCfg, dict(m=dict(type="InhChildwithPosArgs", a=1, b=2, c=3, d=4))
+    )
+    childwithPosArgs = config.m.build()
+    assert childwithPosArgs.uncollected_args == ()
+
+    with pytest.raises(ValidationError):
+        @TestInhReg.register_module()
+        class InhChildwithOverwriteArgs(InhBase):
+            def __init__(self, a: int, b: float):
+                super().__init__(a=a, b=int(b))
+                
+        config = TypeDef.load(
+            InhRegCfg, dict(m=dict(type="InhChildwithOverwriteArgs", a=1, b=2.5))
+        )

utilsd/config/registry.py

@@ -33,6 +33,7 @@ def __new__(cls, clsname, bases, attrs, name=None):
         cls = super().__new__(cls, clsname, bases, attrs)
         cls._name = name
         cls._module_dict = {}
+        cls._inherit_dict = {} # track whether a module should expand its superclass init parameters when specified with **kwargs
         return cls
 
     @property
@@ -47,7 +48,7 @@ def __len__(cls):
         return len(cls._module_dict)
 
     def __contains__(cls, key):
-        return cls.get(key) is not None
+        return key in cls._module_dict
 
     def __repr__(cls):
         format_str = cls.__name__ + f'(name={cls._name}, items={cls._module_dict})'
@@ -57,14 +58,19 @@ def get(cls, key):
         if key in cls._module_dict:
             return cls._module_dict[key]
         raise KeyError(f'{key} not found in {cls}')
-
+    
+    def get_module_with_inherit(cls, key):
+        if key in cls._module_dict:
+            return cls._module_dict[key], cls._inherit_dict[key]
+        raise KeyError(f'{key} not found in {cls}')
+        
     def inverse_get(cls, value):
         keys = [k for k, v in cls._module_dict.items() if v == value]
         if len(keys) != 1:
             raise ValueError(f'{value} needs to appear exactly once in {cls}')
         return keys[0]
 
-    def _register_module(cls, module_class, module_name=None, force=False):
+    def _register_module(cls, module_class, module_name=None, force=False, *, inherit=False):
         if not inspect.isclass(module_class):
             raise TypeError(f'module must be a class, but got {type(module_class)}')
 
@@ -76,8 +82,9 @@ def _register_module(cls, module_class, module_name=None, force=False):
             if not force and name in cls._module_dict:
                 raise KeyError(f'{name} is already registered in {cls.name}')
             cls._module_dict[name] = module_class
+            cls._inherit_dict[name] = inherit
 
-    def register_module(cls, name: Optional[str] = None, force: bool = False, module: Type = None):
+    def register_module(cls, name: Optional[str] = None, force: bool = False, module: Type = None, *, inherit=False):
         if not isinstance(force, bool):
             raise TypeError(f'force must be a boolean, but got {type(force)}')
 
@@ -89,12 +96,12 @@ def register_module(cls, name: Optional[str] = None, force: bool = False, module
 
         # use it as a normal method: x.register_module(module=SomeClass)
         if module is not None:
-            cls._register_module(module_class=module, module_name=name, force=force)
+            cls._register_module(module_class=module, module_name=name, force=force, inherit=inherit)
             return module
 
         # use it as a decorator: @x.register_module()
         def _register(reg_cls):
-            cls._register_module(module_class=reg_cls, module_name=name, force=force)
+            cls._register_module(module_class=reg_cls, module_name=name, force=force, inherit=inherit)
             return reg_cls
 
         return _register
@@ -139,37 +146,83 @@ class SubclassConfig(Generic[T], metaclass=DataclassType):
     """
 
 
-def dataclass_from_class(cls):
+def dataclass_from_class(cls, *, inherit_signature=False):
     """Create a configurable dataclass for a class
     based on its ``__init__`` signature.
     """
     class_name = cls.__name__ + 'Config'
     fields = [
         ('_type', ClassVar[Type], cls),
     ]
+    non_default_fields = []
+    default_fields = []
     init_signature = inspect.signature(cls.__init__)
+    # Track presented param names. The same name may appear in different classes when **kwargs is passed.
+    existing_names = dict()
+    expand_super = False
     for idx, param in enumerate(init_signature.parameters.values()):
         if idx == 0:
             # skip self
             continue
-        if param.kind in (param.VAR_POSITIONAL, param.VAR_KEYWORD):
-            # FIXME add support for args and kwargs later
+        if param.kind == param.VAR_POSITIONAL:
+            if inherit_signature:
+                # Prohibit uncollected positional varibles
+                # TODO: should positional params be banned from all use cases? 
+                raise TypeError(f'Use of positional params `*arg` in "{cls}" is prehibitted. Try to use `**kwargs` instead to avoid possible confusion.')
+            continue
+        if param.kind == param.VAR_KEYWORD:
+            # Expand __init__ of the super classes for signitures
+            if inherit_signature:
+                expand_super = True
             continue
 
         # TODO: fix type annotation for dependency injection
         if param.annotation == param.empty:
             raise TypeError(f'Parameter of `__init__` "{param}" of "{cls}" must have annotation.')
+        existing_names[param.name] = (cls, param.annotation)
         if param.default != param.empty:
-            fields.append((param.name, param.annotation, param.default))
+            default_fields.append((param.name, param.annotation, param.default))
         else:
-            fields.append((param.name, param.annotation))
+            non_default_fields.append((param.name, param.annotation))
+    
+    # check the super classes of cls 
+    for scls in cls.mro()[1:]:
+        scls_signature = inspect.signature(scls.__init__)
+        for idx, param in enumerate(scls_signature.parameters.values()):
+            if idx == 0:
+                # skip self
+                continue
+            if param.kind in (param.VAR_POSITIONAL, param.VAR_KEYWORD):
+                # mro has already contained all the super classes so we don't need to do expansion again.
+                continue
+            
+            if param.annotation == param.empty:
+                raise TypeError(f'Parameter of `__init__` "{param}" of the superclass "{scls}" of "{cls}" must have annotation.')
+            
+            if param.name in existing_names:
+                if existing_names[param.name][1] != param.annotation:
+                    raise TypeError(
+                        f'Inconsist annotations found for the same param for inherited classes:\n'
+                        f'\tParam name: {param.name}\n'
+                        f'\tAnnotation in {existing_names[param.name][0]}: {existing_names[param.name][1]}\n'
+                        f'\tAnnotation in {scls}: {param.annotation}'
+                    )
+            else:
+                if expand_super:
+                    if param.default != param.empty:
+                        default_fields.append((param.name, param.annotation, param.default))
+                    else:
+                        non_default_fields.append((param.name, param.annotation))
+
+    fields = fields + non_default_fields + default_fields
 
     def type_fn(self): return self._type
 
     def build_fn(self, **kwargs):
         result = {f.name: getattr(self, f.name) for f in dataclasses.fields(self)}
         for k in kwargs:
             # silently overwrite the arguments with given ones.
+            # FIXME: add type check when building?
             result[k] = kwargs[k]
         try:
             return self._type(**result)

utilsd/config/type_def.py

@@ -584,9 +584,8 @@ def from_plain(self, plain, ctx):
             raise TypeError(f'Expect a dict with key "type", but found {type(plain)}: {plain}')
         # copy the raw object to prevent unexpected modification
         plain = copy.copy(plain)
-
-        type_ = self.registry.get(plain.pop('type'))
-        dataclass = dataclass_from_class(type_)
+        type_, inherit = self.registry.get_module_with_inherit(plain.pop('type'))
+        dataclass = dataclass_from_class(type_, inherit_signature=inherit)
         return super().from_plain(plain, ctx, type_=dataclass)
 
     def to_plain(self, obj, ctx):