refactor: merge dataclass and namedtuple as one uniform struct_type

Author: lemorage

python/cocoindex/convert.py

@@ -57,17 +57,16 @@ def make_engine_value_decoder(
             f"Type mismatch for `{''.join(field_path)}`: "
             f"passed in {src_type_kind}, declared {dst_annotation} ({dst_type_info.kind})")
 
-    if dst_type_info.dataclass_type is not None or dst_type_info.namedtuple_type is not None:
-        struct_type = dst_type_info.dataclass_type or dst_type_info.namedtuple_type
+    if dst_type_info.struct_type is not None:
         return _make_engine_struct_value_decoder(
-            field_path, src_type['fields'], struct_type)
+            field_path, src_type['fields'], dst_type_info.struct_type)
 
     if src_type_kind in TABLE_TYPES:
         field_path.append('[*]')
         elem_type_info = analyze_type_info(dst_type_info.elem_type)
-        if elem_type_info.dataclass_type is None and elem_type_info.namedtuple_type is None:
+        if elem_type_info.struct_type is None:
             raise ValueError(f"Type mismatch for `{''.join(field_path)}`: "
-                            f"declared `{dst_type_info.kind}`, a dataclass or namedtuple type expected")
+                            f"declared `{dst_type_info.kind}`, a dataclass or NamedTuple type expected")
         engine_fields_schema = src_type['row']['fields']
         if elem_type_info.key_type is not None:
             key_field_schema = engine_fields_schema[0]
@@ -76,16 +75,14 @@ def make_engine_value_decoder(
                 field_path, key_field_schema['type'], elem_type_info.key_type)
             field_path.pop()
             value_decoder = _make_engine_struct_value_decoder(
-                field_path, engine_fields_schema[1:], 
-                elem_type_info.dataclass_type or elem_type_info.namedtuple_type)
+                field_path, engine_fields_schema[1:], elem_type_info.struct_type)
             def decode(value):
                 if value is None:
                     return None
                 return {key_decoder(v[0]): value_decoder(v[1:]) for v in value}
         else:
             elem_decoder = _make_engine_struct_value_decoder(
-                field_path, engine_fields_schema, 
-                elem_type_info.dataclass_type or elem_type_info.namedtuple_type)
+                field_path, engine_fields_schema, elem_type_info.struct_type)
             def decode(value):
                 if value is None:
                     return None
@@ -144,12 +141,8 @@ def make_closure_for_value(name: str, param: inspect.Parameter) -> Callable[[lis
         make_closure_for_value(name, param)
         for (name, param) in parameters.items()]
 
-    if is_dataclass:
-        return lambda values: dst_struct_type(
-            *(decoder(values) for decoder in field_value_decoder))
-    else:  # namedtuple
-        return lambda values: dst_struct_type(
-            *(decoder(values) for decoder in field_value_decoder))
+    return lambda values: dst_struct_type(
+        *(decoder(values) for decoder in field_value_decoder))
 
 def dump_engine_object(v: Any) -> Any:
     """Recursively dump an object for engine. Engine side uses `Pythonized` to catch."""

python/cocoindex/typing.py

@@ -72,8 +72,7 @@ class AnalyzedTypeInfo:
     elem_type: ElementType | None   # For Vector and Table
 
     key_type: type | None           # For element of KTable
-    dataclass_type: type | None     # For Struct
-    namedtuple_type: type | None    # For Struct
+    struct_type: type | None        # For Struct, a dataclass or namedtuple
 
     attrs: dict[str, Any] | None
     nullable: bool = False
@@ -121,19 +120,16 @@ def analyze_type_info(t) -> AnalyzedTypeInfo:
         elif isinstance(attr, TypeKind):
             kind = attr.kind
 
-    dataclass_type = None
-    namedtuple_type = None
+    struct_type = None
     elem_type = None
     key_type = None
     if _is_struct_type(t):
+        struct_type = t
+
         if kind is None:
             kind = 'Struct'
         elif kind != 'Struct':
             raise ValueError(f"Unexpected type kind for struct: {kind}")
-        if dataclasses.is_dataclass(t):
-            dataclass_type = t
-        elif is_namedtuple_type(t):
-            namedtuple_type = t
     elif base_type is collections.abc.Sequence or base_type is list:
         args = typing.get_args(t)
         elem_type = args[0]
@@ -180,8 +176,7 @@ def analyze_type_info(t) -> AnalyzedTypeInfo:
         vector_info=vector_info,
         elem_type=elem_type,
         key_type=key_type,
-        dataclass_type=dataclass_type,
-        namedtuple_type=namedtuple_type,
+        struct_type=struct_type,
         attrs=attrs,
         nullable=nullable,
     )
@@ -216,11 +211,10 @@ def _encode_type(type_info: AnalyzedTypeInfo) -> dict[str, Any]:
     encoded_type: dict[str, Any] = { 'kind': type_info.kind }
 
     if type_info.kind == 'Struct':
-        struct_type = type_info.dataclass_type or type_info.namedtuple_type
-        if struct_type is None:
+        if type_info.struct_type is None:
             raise ValueError("Struct type must have a dataclass or namedtuple type")
-        encoded_type['fields'] = _encode_fields_schema(struct_type, type_info.key_type)
-        if doc := inspect.getdoc(struct_type):
+        encoded_type['fields'] = _encode_fields_schema(type_info.struct_type, type_info.key_type)
+        if doc := inspect.getdoc(type_info.struct_type):
             encoded_type['description'] = doc
 
     elif type_info.kind == 'Vector':