Merge pull request #377 from yukinarit/feature/fix-nested-generic-classes

Fix nested generic class deserialization

Author: yukinarit

serde/compat.py

@@ -812,6 +812,28 @@ def is_ellipsis(typ: Any) -> bool:
     return typ is Ellipsis
 
 
+def get_type_var_names(cls: Type[Any]) -> Optional[List[str]]:
+    """
+    Get type argument names of a generic class.
+
+    >>> T = typing.TypeVar('T')
+    >>> class GenericFoo(typing.Generic[T]):
+    ...     pass
+    >>> get_type_var_names(GenericFoo)
+    ['T']
+    >>> get_type_var_names(int)
+    """
+    bases = getattr(cls, "__orig_bases__", ())
+    if not bases:
+        return None
+
+    type_arg_names: List[str] = []
+    for base in bases:
+        type_arg_names.extend(arg.__name__ for arg in get_args(base))
+
+    return type_arg_names
+
+
 def find_generic_arg(cls: Type[Any], field: TypeVar) -> int:
     """
     Find a type in generic parameters.
@@ -843,26 +865,47 @@ def find_generic_arg(cls: Type[Any], field: TypeVar) -> int:
     return -1
 
 
-def get_generic_arg(typ: Any, index: int) -> Any:
+def get_generic_arg(
+    typ: Any,
+    maybe_generic_type_vars: Optional[List[str]],
+    variable_type_args: Optional[List[str]],
+    index: int,
+) -> Any:
     """
-    Get generic type argument by index.
+    Get generic type argument.
 
     >>> T = typing.TypeVar('T')
     >>> U = typing.TypeVar('U')
     >>> class GenericFoo(typing.Generic[T, U]):
     ...     pass
-    >>> get_generic_arg(GenericFoo[int, str], 0).__name__
+    >>> get_generic_arg(GenericFoo[int, str], ['T', 'U'], ['T', 'U'], 0).__name__
     'int'
-    >>> get_generic_arg(GenericFoo[int, str], 1).__name__
+    >>> get_generic_arg(GenericFoo[int, str], ['T', 'U'], ['T', 'U'], 1).__name__
+    'str'
+    >>> get_generic_arg(GenericFoo[int, str], ['T', 'U'], ['U'], 0).__name__
     'str'
     """
-    if not is_generic(typ):
+    if not is_generic(typ) or maybe_generic_type_vars is None or variable_type_args is None:
         return typing.Any
-    else:
-        args = get_args(typ)
-        if index + 1 > len(args):
-            return typing.Any
-        return args[index]
+
+    args = get_args(typ)
+
+    if len(args) != len(maybe_generic_type_vars):
+        raise SerdeError(
+            f"Number of type args for {typ} does not match number of generic type vars: "
+            f"\n  type args: {args}\n  type_vars: {maybe_generic_type_vars}"
+        )
+
+    # Get the name of the type var used for this field in the parent class definition
+    type_var_name = variable_type_args[index]
+
+    try:
+        # Find the slot of that type var in the original generic class definition
+        orig_index = maybe_generic_type_vars.index(type_var_name)
+    except ValueError:
+        return typing.Any
+
+    return args[orig_index]
 
 
 def has_default(field: dataclasses.Field) -> bool:

serde/core.py

@@ -398,6 +398,7 @@ class Field:
     deserializer: Optional[Func] = None  # Custom field deserializer.
     flatten: Optional[FlattenOpts] = None
     parent: Optional[Type[Any]] = None
+    type_args: Optional[List[str]] = None
 
     @classmethod
     def from_dataclass(cls, f: dataclasses.Field, parent: Optional[Type[Any]] = None) -> "Field":

serde/de.py

@@ -9,7 +9,7 @@
 import functools
 import typing
 from dataclasses import dataclass, is_dataclass
-from typing import Any, Callable, Dict, List, Optional, TypeVar, Generic, overload
+from typing import Any, Callable, Dict, Generic, List, Optional, TypeVar, overload
 
 import jinja2
 from typing_extensions import Type, dataclass_transform
@@ -24,6 +24,7 @@
     get_args,
     get_generic_arg,
     get_origin,
+    get_type_var_names,
     has_default,
     has_default_factory,
     is_any,
@@ -602,6 +603,10 @@ def render(self, arg: DeField) -> str:
         """
         if arg.deserializer and arg.deserializer.inner is not default_deserializer:
             res = self.custom_field_deserializer(arg)
+        elif is_generic(arg.type):
+            arg.type_args = get_args(arg.type)
+            arg.type = get_origin(arg.type)
+            res = self.render(arg)
         elif is_dataclass(arg.type):
             res = self.dataclass(arg)
         elif is_opt(arg.type):
@@ -639,12 +644,10 @@ def render(self, arg: DeField) -> str:
         elif isinstance(arg.type, TypeVar):
             index = find_generic_arg(self.cls, arg.type)
             res = (
-                f"from_obj(get_generic_arg(maybe_generic, {index}), "
-                f" {arg.data}, named={not arg.iterbased}, reuse_instances=reuse_instances)"
+                f"from_obj(get_generic_arg(maybe_generic, maybe_generic_type_vars, "
+                f"variable_type_args, {index}), {arg.data}, named={not arg.iterbased}, "
+                "reuse_instances=reuse_instances)"
             )
-        elif is_generic(arg.type):
-            arg.type = get_origin(arg.type)
-            res = self.render(arg)
         elif is_literal(arg.type):
             res = self.literal(arg)
         else:
@@ -690,7 +693,12 @@ def dataclass(self, arg: DeField) -> str:
             else:
                 var = arg.datavar
 
-        opts = "maybe_generic=maybe_generic, reuse_instances=reuse_instances"
+        type_args_str = [str(t).lstrip("~") for t in arg.type_args] if arg.type_args else None
+
+        opts = (
+            "maybe_generic=maybe_generic, maybe_generic_type_vars=maybe_generic_type_vars, "
+            f"variable_type_args={type_args_str}, reuse_instances=reuse_instances"
+        )
 
         if arg.is_self_referencing():
             class_name = "cls"
@@ -718,6 +726,7 @@ def opt(self, arg: DeField) -> str:
         ...     o: Optional[List[int]]
         >>> Renderer('foo').render(DeField(Optional[Foo], 'f', datavar='data'))
         '(Foo.__serde__.funcs[\\'foo\\'](data=data["f"], maybe_generic=maybe_generic, \
+maybe_generic_type_vars=maybe_generic_type_vars, variable_type_args=None, \
 reuse_instances=reuse_instances)) if data.get("f") is not None else None'
         """
         value = arg[0]
@@ -771,13 +780,16 @@ def tuple(self, arg: DeField) -> str:
         >>> Renderer('foo').render(DeField(Tuple[str, int, List[int], Foo], 'd', datavar='data'))
         '(coerce(str, data["d"][0]), coerce(int, data["d"][1]), \
 [coerce(int, v) for v in data["d"][2]], \
-Foo.__serde__.funcs[\\'foo\\'](data=data["d"][3], maybe_generic=maybe_generic, reuse_instances=reuse_instances),)'
+Foo.__serde__.funcs[\\'foo\\'](data=data["d"][3], maybe_generic=maybe_generic, \
+maybe_generic_type_vars=maybe_generic_type_vars, variable_type_args=None, \
+reuse_instances=reuse_instances),)'
 
         >>> field = DeField(Tuple[str, int, List[int], Foo], 'd', datavar='data', index=0, iterbased=True)
         >>> Renderer('foo').render(field)
         "(coerce(str, data[0][0]), coerce(int, data[0][1]), \
 [coerce(int, v) for v in data[0][2]], Foo.__serde__.funcs['foo'](data=data[0][3], \
-maybe_generic=maybe_generic, reuse_instances=reuse_instances),)"
+maybe_generic=maybe_generic, maybe_generic_type_vars=maybe_generic_type_vars, \
+variable_type_args=None, reuse_instances=reuse_instances),)"
         """
         if is_bare_tuple(arg.type) or is_variable_tuple(arg.type):
             return f"tuple({arg.data})"
@@ -799,9 +811,10 @@ def dict(self, arg: DeField) -> str:
         >>> @deserialize
         ... class Foo: pass
         >>> Renderer('foo').render(DeField(Dict[Foo, List[Foo]], 'f', datavar='data'))
-        '{Foo.__serde__.funcs[\\'foo\\'](data=k, maybe_generic=maybe_generic, reuse_instances=reuse_instances): \
-[Foo.__serde__.funcs[\\'foo\\'](data=v, maybe_generic=maybe_generic, reuse_instances=reuse_instances) for v in v] \
-for k, v in data["f"].items()}'
+        '{Foo.__serde__.funcs[\\'foo\\'](data=k, maybe_generic=maybe_generic, \
+maybe_generic_type_vars=maybe_generic_type_vars, variable_type_args=None, reuse_instances=reuse_instances): \
+[Foo.__serde__.funcs[\\'foo\\'](data=v, maybe_generic=maybe_generic, maybe_generic_type_vars=maybe_generic_type_vars, \
+variable_type_args=None, reuse_instances=reuse_instances) for v in v] for k, v in data["f"].items()}'
         """
         if is_bare_dict(arg.type):
             return arg.data
@@ -898,13 +911,16 @@ def renderable(f: DeField) -> bool:
 
 def render_from_iter(cls: Type[Any], custom: Optional[DeserializeFunc] = None, type_check: TypeCheck = NoCheck) -> str:
     template = """
-def {{func}}(cls=cls, maybe_generic=None, data=None, reuse_instances = {{serde_scope.reuse_instances_default}}):
+def {{func}}(cls=cls, maybe_generic=None, maybe_generic_type_vars=None, data=None,
+             variable_type_args=None, reuse_instances = {{serde_scope.reuse_instances_default}}):
   if reuse_instances is Ellipsis:
     reuse_instances = {{serde_scope.reuse_instances_default}}
 
   if data is None:
     return None
 
+  maybe_generic_type_vars = maybe_generic_type_vars or {{cls_type_vars}}
+
   {% for f in fields %}
   __{{f.name}} = {{f|arg(loop.index-1)|rvalue}}
   {% endfor %}
@@ -924,7 +940,12 @@ def {{func}}(cls=cls, maybe_generic=None, data=None, reuse_instances = {{serde_s
     env.filters.update({"rvalue": renderer.render})
     env.filters.update({"arg": to_iter_arg})
     fields = list(filter(renderable, defields(cls)))
-    res = env.get_template("iter").render(func=FROM_ITER, serde_scope=getattr(cls, SERDE_SCOPE), fields=fields)
+    res = env.get_template("iter").render(
+        func=FROM_ITER,
+        serde_scope=getattr(cls, SERDE_SCOPE),
+        fields=fields,
+        cls_type_vars=get_type_var_names(cls),
+    )
 
     if renderer.import_numpy:
         res = "import numpy\n" + res
@@ -939,14 +960,16 @@ def render_from_dict(
     type_check: TypeCheck = NoCheck,
 ) -> str:
     template = """
-def {{func}}(cls=cls, maybe_generic=None, data=None,
-             reuse_instances = {{serde_scope.reuse_instances_default}}):
+def {{func}}(cls=cls, maybe_generic=None, maybe_generic_type_vars=None, data=None,
+             variable_type_args=None, reuse_instances = {{serde_scope.reuse_instances_default}}):
   if reuse_instances is Ellipsis:
     reuse_instances = {{serde_scope.reuse_instances_default}}
 
   if data is None:
     return None
 
+  maybe_generic_type_vars = maybe_generic_type_vars or {{cls_type_vars}}
+
   {% for f in fields %}
   __{{f.name}} = {{f|arg(loop.index-1)|rvalue}}
   {% endfor %}
@@ -973,7 +996,11 @@ def {{func}}(cls=cls, maybe_generic=None, data=None,
     env.filters.update({"arg": functools.partial(to_arg, rename_all=rename_all)})
     fields = list(filter(renderable, defields(cls)))
     res = env.get_template("dict").render(
-        func=FROM_DICT, serde_scope=getattr(cls, SERDE_SCOPE), fields=fields, type_check=type_check
+        func=FROM_DICT,
+        serde_scope=getattr(cls, SERDE_SCOPE),
+        fields=fields,
+        type_check=type_check,
+        cls_type_vars=get_type_var_names(cls),
     )
 
     if renderer.import_numpy:
@@ -984,7 +1011,8 @@ def {{func}}(cls=cls, maybe_generic=None, data=None,
 
 def render_union_func(cls: Type[Any], union_args: List[Type[Any]], tagging: Tagging = DefaultTagging) -> str:
     template = """
-def {{func}}(cls=cls, maybe_generic=None, data=None, reuse_instances = {{serde_scope.reuse_instances_default}}):
+def {{func}}(cls=cls, maybe_generic=None, maybe_generic_type_vars=None, data=None,
+             variable_type_args=None, reuse_instances = {{serde_scope.reuse_instances_default}}):
   errors = []
   {% for t in union_args %}
   try:
@@ -1042,7 +1070,8 @@ def {{func}}(cls=cls, maybe_generic=None, data=None, reuse_instances = {{serde_s
 
 def render_literal_func(cls: Type[Any], literal_args: List[Any], tagging: Tagging = DefaultTagging) -> str:
     template = """
-def {{func}}(cls=cls, maybe_generic=None, data=None, reuse_instances = {{serde_scope.reuse_instances_default}}):
+def {{func}}(cls=cls, maybe_generic=None, maybe_generic_type_vars=None, data=None,
+             variable_type_args=None, reuse_instances = {{serde_scope.reuse_instances_default}}):
   if data in ({%- for v in literal_args -%}{{v|repr}},{%- endfor -%}):
     return data
   raise SerdeError("Can not deserialize " + repr(data) + " as {{literal_name}}.")

tests/common.py

@@ -9,6 +9,7 @@
 from typing import Any, Callable, DefaultDict, Dict, FrozenSet, Generic, List, NewType, Optional, Set, Tuple, TypeVar
 
 import more_itertools
+
 from serde import from_dict, from_tuple, serde, to_dict, to_tuple
 from serde.json import from_json, to_json
 from serde.msgpack import from_msgpack, to_msgpack
@@ -39,13 +40,26 @@
 
 U = TypeVar("U")
 
+V = TypeVar("V")
+
 
 @serde
 class GenericClass(Generic[T, U]):
     a: T
     b: U
 
 
+@serde
+class Inner(Generic[T]):
+    c: T
+
+
+@serde
+class NestedGenericClass(Generic[U, V]):
+    a: U
+    b: Inner[V]
+
+
 def param(val, typ, filter: Optional[Callable] = None):
     """
     Create a test parameter
@@ -100,6 +114,7 @@ def toml_not_supported(se, de, opt) -> bool:
     param(10, NewType("Int", int)),  # NewType
     param({"a": 1}, Any),  # Any
     param(GenericClass[str, int]("foo", 10), GenericClass[str, int]),  # Generic
+    param(NestedGenericClass[str, int]("foo", Inner[int](10)), NestedGenericClass[str, int]),
     param(pathlib.Path("/tmp/foo"), pathlib.Path),  # Extended types
     param(pathlib.Path("/tmp/foo"), Optional[pathlib.Path]),
     param(None, Optional[pathlib.Path], toml_not_supported),

tests/test_compat.py

@@ -1,11 +1,14 @@
 import sys
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Dict, Generic, List, NewType, Optional, Set, Tuple, TypeVar, Union
+from typing import Any, Dict, Generic, List, NewType, Optional, Set, Tuple, TypeVar, Union
+
+import pytest
 
 import serde
 from serde.compat import (
     Literal,
+    get_generic_arg,
     is_dict,
     is_generic,
     is_list,
@@ -25,6 +28,7 @@
 from .data import Bool, Float, Int, Pri, PriOpt, Str
 
 T = TypeVar("T")
+U = TypeVar("U")
 
 
 def test_types():
@@ -257,3 +261,19 @@ def test_is_generic():
     assert not serde.is_serializable(GenericFoo[List[int]])
     assert serde.is_deserializable(GenericFoo)
     assert not serde.is_deserializable(GenericFoo[List[int]])
+
+
+def test_get_generic_arg():
+    class GenericFoo(Generic[T, U]):
+        pass
+
+    assert get_generic_arg(GenericFoo[int, str], ["T", "U"], ["T", "U"], 0) == int
+    assert get_generic_arg(GenericFoo[int, str], ["T", "U"], ["T", "U"], 1) == str
+    assert get_generic_arg(GenericFoo[int, str], ["T", "U"], ["U"], 0) == str
+    assert get_generic_arg(GenericFoo[int, str], ["T", "U"], ["V"], 0) == Any
+
+    with pytest.raises(serde.SerdeError):
+        get_generic_arg(GenericFoo[int, str], ["T"], ["T"], 0)
+
+    with pytest.raises(serde.SerdeError):
+        get_generic_arg(GenericFoo[int, str], ["T"], ["U"], 0)