Fix NotImplementedError on heterogeneous union model fields

faust/models/typing.py

@@ -331,7 +331,19 @@ def _filter_NoneType(self, union_args: Tuple) -> Iterator:
         return (x for x in union_args if not _is_NoneType(x))
 
     def build(self, var: Variable, *args: Type) -> str:
-        raise NotImplementedError(f"Union of types {args!r} not supported")
+        # A union that reaches this point is heterogeneous: its members are
+        # neither all Model subclasses nor all built-in scalar/"literal" types
+        # (both of those are collapsed to a single node by
+        # `_maybe_unroll_union` above). There is no unambiguous way to coerce a
+        # value into such a union -- e.g. ``str | list | dict | None`` -- so we
+        # pass the already-deserialized value through unchanged. This mirrors
+        # the handling of an all-literal union such as ``Union[str, int,
+        # float]`` (which compiles to a bare pass-through) and restores the
+        # behavior that existed before mode-streaming started recognizing PEP
+        # 604 (``X | Y``) unions; that change made annotations like ``str |
+        # list | dict | None`` reach this node and raise NotImplementedError at
+        # class-definition time (faust-streaming/mode#80).
+        return f"{var}"
 
 
 class LiteralNode(Node):

tests/functional/models/test_typing.py

@@ -397,6 +397,26 @@ def Xi(i):
     expected_types={NodeType.MODEL: {ModelT}},
 )
 
+# Regression for faust-streaming/mode#80: a heterogeneous union whose members
+# are neither all Models nor all scalar "literal" types (here list/dict are
+# neither) must pass the value through unchanged instead of raising
+# NotImplementedError.
+CASE_UNION_STR_LIST_DICT = TypeExpressionTest(
+    type_expression=Union[str, list, dict],
+    serialized_data={"key": "value"},
+    expected_result={"key": "value"},
+    expected_comprehension="a",
+    expected_types={},
+)
+
+CASE_UNION_STR_LIST_DICT_NONE = TypeExpressionTest(
+    type_expression=Union[str, list, dict, None],
+    serialized_data=[1, 2, 3],
+    expected_result=[1, 2, 3],
+    expected_comprehension="(a if a is not None else None)",
+    expected_types={},
+)
+
 CASES = [
     CASE_TUPLE_LIST_SET_X,
     CASE_LIST_LIST_X,
@@ -423,6 +443,8 @@ def Xi(i):
     CASE_LIST_NO_ARGS,
     CASE_UNION_STR_INT_FLOAT,
     CASE_UNION_X_Y,
+    CASE_UNION_STR_LIST_DICT,
+    CASE_UNION_STR_LIST_DICT_NONE,
 ]
 
 
@@ -439,3 +461,29 @@ def test_compile(case):
     fun = expr.as_function(globals=globals())
     assert fun(case.serialized_data) == case.expected_result
     assert expr.found_types == case.expected_types
+
+
+def test_record_heterogeneous_union_field():
+    # Regression test for faust-streaming/mode#80: defining a Record with a
+    # heterogeneous union field must not raise NotImplementedError at
+    # class-definition time, and the value round-trips through unchanged
+    # (including None, via the optional pass-through).
+    class SensitiveInfo(Record, namespace="test.SensitiveInfo"):
+        data: Union[str, list, dict, None]
+        meta: dict
+
+    v = SensitiveInfo.from_data({"data": [1, 2, 3], "meta": {"k": "v"}})
+    assert v.data == [1, 2, 3]
+    assert v.meta == {"k": "v"}
+    assert SensitiveInfo.from_data({"data": None, "meta": {}}).data is None
+
+
+def test_record_heterogeneous_union_field_pep604():
+    # The exact PEP 604 (X | Y) spelling from mode#80 must behave identically
+    # to the typing.Union spelling above.
+    class SensitiveInfo(Record, namespace="test.SensitiveInfoPEP604"):
+        data: str | list | dict | None
+        meta: dict
+
+    v = SensitiveInfo.from_data({"data": {"k": "v"}, "meta": {}})
+    assert v.data == {"k": "v"}