Allowing dataclasses which require list protobuf deserialization

flytekit/core/type_engine.py

@@ -148,7 +148,7 @@
         return self._type_assertions_enabled
 
     def assert_type(self, t: Type[T], v: T):
-        if not hasattr(t, "__origin__") and not isinstance(v, t):
+        if not ((get_origin(t) is not None) or isinstance(v, t)):
             raise TypeTransformerFailedError(f"Expected value of type {t} but got '{v}' of type {type(v)}")
 
     @abstractmethod
@@ -706,10 +706,24 @@
     def get_literal_type(self, t: Type[T]) -> LiteralType:
         return LiteralType(simple=SimpleType.STRUCT, metadata={ProtobufTransformer.PB_FIELD_KEY: self.tag(t)})
 
+    def _handle_list_literal(self, ctx: FlyteContext, elems: list) -> Literal:
+        if len(elems) == 0:
+            return Literal(collection=LiteralCollection(literals=[]))
+        st = type(elems[0])
+        lt = TypeEngine.to_literal_type(st)
+        lits = [TypeEngine.to_literal(ctx, x, st, lt) for x in elems]
+        return Literal(collection=LiteralCollection(literals=lits))
+
     def to_literal(self, ctx: FlyteContext, python_val: T, python_type: Type[T], expected: LiteralType) -> Literal:
         struct = Struct()
         try:
-            struct.update(_MessageToDict(cast(Message, python_val)))
+            message_dict = _MessageToDict(cast(Message, python_val))
+            try:
+                struct.update(message_dict)
+            except Exception:
+                if isinstance(message_dict, list):
+                    return self._handle_list_literal(ctx, message_dict)
+                raise
         except Exception:
             raise TypeTransformerFailedError("Failed to convert to generic protobuf struct")
         return Literal(scalar=Scalar(generic=struct))

tests/flytekit/unit/core/test_type_engine.py

@@ -3010,6 +3010,35 @@ class DatumDataUnion:
     assert datum_mashumaro_orjson.z.isoformat() == pv.z
 
 
+def test_DataclassTransformer_for_list_attributes():
+    dataclass_bases = [DataClassJSONMixin, DataClassJsonMixin, object]
+    for dataclass_base in dataclass_bases:
+        @dataclass
+        class DataclassWithList(dataclass_base):
+            some_strings: list[str]
+
+        @task
+        def create_dataclass_with_list(
+            some_strings: list[str],
+        ) -> DataclassWithList:
+            return DataclassWithList(
+                some_strings=some_strings,
+            )
+
+        @workflow
+        def convert_list_workflow(
+            inputs: DataclassWithList,
+        ) -> DataclassWithList:
+            result_list = create_dataclass_with_list(some_strings=inputs.some_strings)
+            return result_list
+
+        workflow_input = DataclassWithList(
+            some_strings=["hello", "world"],
+        )
+        output = convert_list_workflow(workflow_input)
+        assert output.some_strings == ["hello", "world"]
+
+
 def test_dataclass_encoder_and_decoder_registry():
     iterations = 10