fix support deserialize non dataclass (#240)

src/ffmpeg/dag/serialize.py

@@ -4,16 +4,9 @@
 import json  # noqa
 from dataclasses import fields, is_dataclass
 from enum import Enum
-from typing import Any, ClassVar, Dict, Protocol, TypeVar
+from typing import Any, TypeVar, cast
 
-
-class IsDataclass(Protocol):
-    # as already noted in comments, checking for this attribute is currently
-    # the most reliable way to ascertain that something is a dataclass
-    __dataclass_fields__: ClassVar[Dict[str, Any]]
-
-
-T = TypeVar("T", bound=IsDataclass)
+T = TypeVar("T")
 
 
 def load_class(path: str) -> Any:
@@ -78,10 +71,9 @@ def object_hook(self, obj: Any) -> Any:  # pylint: disable=method-hidden
 
 def loads(cls: type[T], raw: str) -> T:
     data = json.loads(raw, cls=Decoder)
-    assert isinstance(data, cls)
-    return data
+    return cast(T, data)
 
 
 # Serialization
-def dumps(instance: IsDataclass) -> str:
+def dumps(instance: Any) -> str:
     return json.dumps(instance, cls=Encoder)

src/ffmpeg/dag/tests/__snapshots__/test_serialize.ambr

@@ -5,6 +5,9 @@
 # name: test_load_and_dump[serialized]
   '{"__class__": "ffmpeg.dag.tests.test_serialize.Person", "name": "John Doe", "age": 30, "address": {"__class__": "ffmpeg.dag.tests.test_serialize.Address", "street": "123 Main St", "city": "Anytown"}}'
 # ---
+# name: test_load_and_dump_mixed_type[serialized]
+  '[{"__class__": "ffmpeg.streams.av.AVStream", "node": {"__class__": "ffmpeg.dag.nodes.InputNode", "args": [], "kwargs": [], "filename": "input.mp4"}, "selector": null, "index": null}, true]'
+# ---
 # name: test_load_and_dump_on_complex_filter[serialized]
   '{"__class__": "ffmpeg.dag.nodes.OutputStream", "node": {"__class__": "ffmpeg.dag.nodes.OutputNode", "args": [], "kwargs": [], "filename": "out.mp4", "inputs": [{"__class__": "ffmpeg.streams.video.VideoStream", "node": {"__class__": "ffmpeg.dag.nodes.FilterNode", "args": [], "kwargs": [["x", "50"], ["y", "50"], ["width", "120"], ["height", "120"], ["color", "red"], ["thickness", "5"], ["replace", "False"], ["box_source", "None"], ["enable", "None"]], "name": "drawbox", "inputs": [{"__class__": "ffmpeg.streams.video.VideoStream", "node": {"__class__": "ffmpeg.dag.nodes.FilterNode", "args": [], "kwargs": [["x", "0"], ["y", "0"], ["eof_action", "repeat"], ["eval", "frame"], ["shortest", "False"], ["format", "yuv420"], ["repeatlast", "True"], ["alpha", "straight"], ["ts_sync_mode", "default"], ["enable", "None"]], "name": "overlay", "inputs": [{"__class__": "ffmpeg.streams.video.VideoStream", "node": {"__class__": "ffmpeg.dag.nodes.FilterNode", "args": [], "kwargs": [["n", "2"], ["v", "1"], ["a", "0"], ["unsafe", "False"]], "name": "concat", "inputs": [{"__class__": "ffmpeg.streams.video.VideoStream", "node": {"__class__": "ffmpeg.dag.nodes.FilterNode", "args": [], "kwargs": [["start", "INT64_MAX"], ["end", "INT64_MAX"], ["start_pts", "I64_MIN"], ["end_pts", "I64_MIN"], ["duration", "0.0"], ["start_frame", 10], ["end_frame", 20]], "name": "trim", "inputs": [{"__class__": "ffmpeg.streams.av.AVStream", "node": {"__class__": "ffmpeg.dag.nodes.InputNode", "args": [], "kwargs": [], "filename": "input.mp4"}, "selector": null, "index": null}], "input_typings": [{"__class__": "ffmpeg.schema.StreamType", "value": "video"}], "output_typings": [{"__class__": "ffmpeg.schema.StreamType", "value": "video"}]}, "selector": null, "index": 0}, {"__class__": "ffmpeg.streams.video.VideoStream", "node": {"__class__": "ffmpeg.dag.nodes.FilterNode", "args": [], "kwargs": [["start", "INT64_MAX"], ["end", "INT64_MAX"], ["start_pts", "I64_MIN"], ["end_pts", "I64_MIN"], ["duration", "0.0"], ["start_frame", 30], ["end_frame", 40]], "name": "trim", "inputs": [{"__class__": "ffmpeg.streams.av.AVStream", "node": {"__class__": "ffmpeg.dag.nodes.InputNode", "args": [], "kwargs": [], "filename": "input.mp4"}, "selector": null, "index": null}], "input_typings": [{"__class__": "ffmpeg.schema.StreamType", "value": "video"}], "output_typings": [{"__class__": "ffmpeg.schema.StreamType", "value": "video"}]}, "selector": null, "index": 0}], "input_typings": [{"__class__": "ffmpeg.schema.StreamType", "value": "video"}, {"__class__": "ffmpeg.schema.StreamType", "value": "video"}], "output_typings": [{"__class__": "ffmpeg.schema.StreamType", "value": "video"}]}, "selector": null, "index": 0}, {"__class__": "ffmpeg.streams.video.VideoStream", "node": {"__class__": "ffmpeg.dag.nodes.FilterNode", "args": [], "kwargs": [["enable", "None"]], "name": "hflip", "inputs": [{"__class__": "ffmpeg.streams.av.AVStream", "node": {"__class__": "ffmpeg.dag.nodes.InputNode", "args": [], "kwargs": [], "filename": "overlay.png"}, "selector": null, "index": null}], "input_typings": [{"__class__": "ffmpeg.schema.StreamType", "value": "video"}], "output_typings": [{"__class__": "ffmpeg.schema.StreamType", "value": "video"}]}, "selector": null, "index": 0}], "input_typings": [{"__class__": "ffmpeg.schema.StreamType", "value": "video"}, {"__class__": "ffmpeg.schema.StreamType", "value": "video"}], "output_typings": [{"__class__": "ffmpeg.schema.StreamType", "value": "video"}]}, "selector": null, "index": 0}], "input_typings": [{"__class__": "ffmpeg.schema.StreamType", "value": "video"}], "output_typings": [{"__class__": "ffmpeg.schema.StreamType", "value": "video"}]}, "selector": null, "index": 0}]}, "selector": null, "index": null}'
 # ---

src/ffmpeg/dag/tests/test_serialize.py

@@ -4,6 +4,7 @@
 
 from ...base import input
 from ...filters import concat
+from ...streams.av import AVStream
 from ..serialize import dumps, loads
 
 
@@ -54,3 +55,14 @@ def test_load_and_dump_on_complex_filter(snapshot: SnapshotAssertion) -> None:
 
     assert isinstance(deserialized, type(stream))
     assert stream == deserialized
+
+
+def test_load_and_dump_mixed_type(snapshot: SnapshotAssertion) -> None:
+    in_file = input("input.mp4")
+
+    serialized = dumps((in_file, True))
+    assert snapshot(name="serialized") == serialized
+
+    deserialized = loads(tuple[AVStream, bool], serialized)
+
+    assert [in_file, True] == deserialized