feat(backend): Enable json parsing with typing & conversion

autogpt_platform/backend/backend/data/execution.py

@@ -97,7 +97,7 @@ def from_graph(graph: AgentGraphExecution):
     def from_db(execution: AgentNodeExecution):
         if execution.executionData:
             # Execution that has been queued for execution will persist its data.
-            input_data = json.loads(execution.executionData)
+            input_data = json.loads(execution.executionData, target_type=dict[str, Any])
         else:
             # For incomplete execution, executionData will not be yet available.
             input_data: BlockInput = defaultdict()

autogpt_platform/backend/backend/data/graph.py

@@ -56,8 +56,8 @@ def from_db(node: AgentNode):
         obj = Node(
             id=node.id,
             block_id=node.AgentBlock.id,
-            input_default=json.loads(node.constantInput),
-            metadata=json.loads(node.metadata),
+            input_default=json.loads(node.constantInput, target_type=dict[str, Any]),
+            metadata=json.loads(node.metadata, target_type=dict[str, Any]),
         )
         obj.input_links = [Link.from_db(link) for link in node.Input or []]
         obj.output_links = [Link.from_db(link) for link in node.Output or []]
@@ -80,10 +80,9 @@ def from_db(execution: AgentGraphExecution):
         duration = (end_time - start_time).total_seconds()
         total_run_time = duration
 
-        if execution.stats:
-            stats = json.loads(execution.stats)
-            duration = stats.get("walltime", duration)
-            total_run_time = stats.get("nodes_walltime", total_run_time)
+        stats = json.loads(execution.stats or "{}", target_type=dict[str, Any])
+        duration = stats.get("walltime", duration)
+        total_run_time = stats.get("nodes_walltime", total_run_time)
 
         return GraphExecution(
             id=execution.id,
@@ -311,7 +310,9 @@ def from_db(graph: AgentGraph, hide_credentials: bool = False):
     def _process_node(node: AgentNode, hide_credentials: bool) -> Node:
         node_dict = node.model_dump()
         if hide_credentials and "constantInput" in node_dict:
-            constant_input = json.loads(node_dict["constantInput"])
+            constant_input = json.loads(
+                node_dict["constantInput"], target_type=dict[str, Any]
+            )
             constant_input = Graph._hide_credentials_in_input(constant_input)
             node_dict["constantInput"] = json.dumps(constant_input)
         return Node.from_db(AgentNode(**node_dict))

autogpt_platform/backend/backend/util/json.py

@@ -1,7 +1,10 @@
 import json
+from typing import Any, Type, TypeVar, overload
 
 from fastapi.encoders import jsonable_encoder
 
+from .type import convert
+
 
 def to_dict(data) -> dict:
     return jsonable_encoder(data)
@@ -11,4 +14,17 @@ def dumps(data) -> str:
     return json.dumps(jsonable_encoder(data))
 
 
-loads = json.loads
+T = TypeVar("T")
+
+
+@overload
+def loads(data: str, *args, target_type: Type[T], **kwargs) -> T: ...
+
+
+@overload
+def loads(data: str, *args, **kwargs) -> Any: ...
+
+
+def loads(data: str, *args, target_type: Type[T] | None = None, **kwargs) -> Any:
+    parsed = json.loads(data, *args, **kwargs)
+    return convert(parsed, target_type) if target_type else parsed

autogpt_platform/backend/backend/util/type.py

@@ -1,8 +1,8 @@
 import json
-from typing import Any, Type, TypeVar, get_args, get_origin
+from typing import Any, Type, TypeVar, cast, get_args, get_origin
 
 
-class ConversionError(Exception):
+class ConversionError(ValueError):
     pass
 
 
@@ -102,7 +102,7 @@ def __convert_bool(value: Any) -> bool:
         return bool(value)
 
 
-def convert(value: Any, target_type: Type):
+def _convert(value: Any, target_type: Type):
     origin = get_origin(target_type)
     args = get_args(target_type)
     if origin is None:
@@ -175,3 +175,13 @@ def convert(value: Any, target_type: Type):
             return __convert_bool(value)
         else:
             return value
+
+
+T = TypeVar("T")
+
+
+def convert(value: Any, target_type: Type[T]) -> T:
+    try:
+        return cast(T, _convert(value, target_type))
+    except Exception as e:
+        raise ConversionError(f"Failed to convert {value} to {target_type}") from e