From 0dd912342746a7114906e9bc2d3515836b311f15 Mon Sep 17 00:00:00 2001 From: yashrajrr <173883375+yashrajrr@users.noreply.github.com> Date: Fri, 14 Aug 2026 22:55:30 +0530 Subject: [PATCH] fix: add typing.Annotated support to serialize_type/deserialize_type Fixes #12338 - Annotated types could not be round-tripped through serialize_type/deserialize_type. Metadata values were rendered without quotes causing deserialization failures or silent corruption. Changes: - Added _parse_annotated_args() for quote-aware metadata parsing - serialize_type now uses repr() on metadata (strings keep quotes) - deserialize_type uses ast.literal_eval for safe metadata parsing - Added comprehensive test coverage --- haystack/utils/type_serialization.py | 70 +++++++++++++++++++ .../fix-annotated-type-serialization.yaml | 10 +++ test/utils/test_type_serialization.py | 33 +++++++++ 3 files changed, 113 insertions(+) create mode 100644 releasenotes/notes/fix-annotated-type-serialization.yaml diff --git a/haystack/utils/type_serialization.py b/haystack/utils/type_serialization.py index 1205f66b215..ffd292b0b2e 100644 --- a/haystack/utils/type_serialization.py +++ b/haystack/utils/type_serialization.py @@ -30,6 +30,45 @@ _SPECIAL_LITERALS: dict[str, Any] = {"None": None, "NoneType": NoneType, "...": ..., "Ellipsis": ...} +def _parse_annotated_args(args_str: str) -> list[str]: + """ + Parse Annotated arguments string, handling nested brackets and quoted strings. + + Annotated[int, "doc", 42] -> ["int", '"doc"', "42"] + Annotated[int, "a, b", "c"] -> ["int", '"a, b"', '"c"'] + Annotated[list[int], "meta"] -> ["list[int]", '"meta"'] + """ + args = [] + bracket_count = 0 + current_arg = "" + in_quotes = False + quote_char = None + + for char in args_str: + if char in ('"', "'") and not in_quotes: + in_quotes = True + quote_char = char + elif char == quote_char and in_quotes: + in_quotes = False + quote_char = None + + if char == "[" and not in_quotes: + bracket_count += 1 + elif char == "]" and not in_quotes: + bracket_count -= 1 + + if char == "," and bracket_count == 0 and not in_quotes: + args.append(current_arg.strip()) + current_arg = "" + else: + current_arg += char + + if current_arg: + args.append(current_arg.strip()) + + return args + + def _is_union_type(target: Any) -> bool: """ Check if target is a Union type. @@ -93,6 +132,16 @@ def serialize_type(target: Any) -> str: if typing.get_origin(target) is typing.Literal: return f"typing.Literal[{', '.join(repr(a) for a in get_args(target))}]" + # Annotated wraps a type with metadata (e.g. Annotated[int, "doc", 42]). + # The first argument is the wrapped type, the rest are metadata values. + # Serialize metadata with repr() so strings keep their quotes and other literals + # are rendered faithfully. + if typing.get_origin(target) is typing.Annotated: + args = get_args(target) + wrapped_type = serialize_type(args[0]) + metadata = ", ".join(repr(a) for a in args[1:]) + return f"typing.Annotated[{wrapped_type}, {metadata}]" + args = get_args(target) if isinstance(target, UnionType): @@ -248,6 +297,27 @@ def deserialize_type(type_str: str) -> Any: if main_type is typing.Literal: return typing.Literal[ast.literal_eval(f"({generics_str},)")] + # Annotated: first arg is the wrapped type, rest are metadata values. + # Use the same safe parsing as Literal, but the first arg is a type (not a literal). + if main_type is typing.Annotated: + # Parse the full argument list with the quote-aware splitter + annotated_args = _parse_annotated_args(generics_str) + if not annotated_args: + raise DeserializationError("Annotated requires at least one argument (the wrapped type)") + + # First arg is the wrapped type - deserialize it as a type + wrapped_type = deserialize_type(annotated_args[0]) + + # Remaining args are metadata - parse them safely with ast.literal_eval + metadata = [] + for meta_str in annotated_args[1:]: + try: + metadata.append(ast.literal_eval(meta_str)) + except (ValueError, SyntaxError) as e: + raise DeserializationError(f"Could not parse Annotated metadata: {meta_str}") from e + + return typing.Annotated[wrapped_type, *metadata] + generic_args = [_deserialize_type_arg(arg) for arg in _parse_generic_args(generics_str)] # Reconstruct diff --git a/releasenotes/notes/fix-annotated-type-serialization.yaml b/releasenotes/notes/fix-annotated-type-serialization.yaml new file mode 100644 index 00000000000..6898b14d943 --- /dev/null +++ b/releasenotes/notes/fix-annotated-type-serialization.yaml @@ -0,0 +1,10 @@ +--- +fixes: + - | + Fixes typing.Annotated round-trip through serialize_type / deserialize_type. + Previously, Annotated types (e.g. Annotated[int, "doc"]) could not be serialized + and deserialized correctly - metadata values were rendered without quotes causing + deserialization failures or silent corruption (e.g. Annotated[str, "int"] becoming + Annotated[str, int]). Now Annotated types with literal metadata (str, int, bool, + None, bytes) round-trip correctly, preserving both the wrapped type and metadata values. + Mirrors the Literal fix from PR #12286. \ No newline at end of file diff --git a/test/utils/test_type_serialization.py b/test/utils/test_type_serialization.py index 10365b09988..8f6305966dd 100644 --- a/test/utils/test_type_serialization.py +++ b/test/utils/test_type_serialization.py @@ -508,3 +508,36 @@ def test_type_de_se_union_and_optional(): assert serialize_type(Optional[dict]) == "typing.Optional[dict]" assert serialize_type(Optional[float]) == "typing.Optional[float]" assert serialize_type(Optional[bool]) == "typing.Optional[bool]" + + +def test_output_type_serialization_annotated(): + from typing import Annotated + assert serialize_type(Annotated[int, "doc"]) == "typing.Annotated[int, 'doc']" + assert serialize_type(Annotated[str, "int"]) == "typing.Annotated[str, 'int']" + assert serialize_type(Annotated[int, "a, b"]) == "typing.Annotated[int, 'a, b']" + assert serialize_type(Annotated[int, "doc", 42, True, None, b"bytes"]) == "typing.Annotated[int, 'doc', 42, True, None, b'bytes']" + assert serialize_type(Annotated[list[int], "meta"]) == "typing.Annotated[list[int], 'meta']" + + +def test_output_type_deserialization_annotated(): + from typing import Annotated + assert deserialize_type("typing.Annotated[int, 'doc']") == Annotated[int, "doc"] + assert deserialize_type("typing.Annotated[str, 'int']") == Annotated[str, "int"] + assert deserialize_type("typing.Annotated[int, 'a, b']") == Annotated[int, "a, b"] + assert deserialize_type("typing.Annotated[int, 'doc', 42, True, None, b'bytes']") == Annotated[int, "doc", 42, True, None, b"bytes"] + assert deserialize_type("typing.Annotated[list[int], 'meta']") == Annotated[list[int], "meta"] + + +def test_output_type_round_trip_annotated(): + from typing import Annotated + for type_ in [ + Annotated[int, "doc"], + Annotated[str, "int"], + Annotated[int, "a, b"], + Annotated[int, "doc", 42, True, None, b"bytes"], + Annotated[list[int], "meta"], + Annotated[Dict[str, int], "metadata"], + Optional[Annotated[int, "doc"]], + Union[Annotated[str, "x"], int], + ]: + assert deserialize_type(serialize_type(type_)) == type_