Convert dataclass to dict

flowrep/workflow.py

@@ -6,6 +6,7 @@
 import textwrap
 from collections import deque
 from collections.abc import Callable, Iterable
+from dataclasses import asdict, is_dataclass
 from functools import cached_property, update_wrapper
 from typing import Any, Generic, TypeVar, cast
 
@@ -1119,7 +1120,10 @@ def get_hashed_node_dict(workflow_dict: dict[str, dict]) -> dict[str, Any]:
                 hash_dict_tmp["inputs"][inp_name] = data["hash"]
                 hash_dict_tmp["node"]["connected_inputs"].append(inp_name)
             elif "value" in data:
-                hash_dict_tmp["inputs"][inp_name] = data["value"]
+                if is_dataclass(data["value"]):
+                    hash_dict_tmp["inputs"][inp_name] = asdict(data["value"])
+                else:
+                    hash_dict_tmp["inputs"][inp_name] = data["value"]
             else:
                 break_flag = True
         if break_flag:

tests/unit/test_workflow.py

@@ -1,4 +1,5 @@
 import unittest
+from dataclasses import dataclass
 
 import networkx as nx
 
@@ -195,6 +196,16 @@ def workflow_with_if_else(a=10, b=20):
     return x
 
 
+@dataclass
+class TestClass:
+    a: int = 10
+    b: int = 20
+
+
+def some_function(test: TestClass):
+    return test
+
+
 class TestWorkflow(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
@@ -635,6 +646,19 @@ def workflow_with_data(a=10, b=20):
         for node in hashed_dict.values():
             self.assertNotIn("hash", node)
 
+        @fwf.workflow
+        def workflow_with_class(test: TestClass):
+            test = some_function(test)
+            return test
+
+        test_instance = TestClass()
+        workflow_dict = workflow_with_class.run(test=test_instance)
+        hashed_dict = fwf.get_hashed_node_dict(workflow_dict)
+        for node in hashed_dict.values():
+            self.assertIn("hash", node)
+            self.assertIsInstance(node["hash"], str)
+            self.assertEqual(len(node["hash"]), 64)
+
     def test_get_and_set_entry(self):
 
         def yet_another_workflow(a=10, b=20):