diff --git a/HISTORY.md b/HISTORY.md index 942616e3..44613ab3 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -13,6 +13,9 @@ Our backwards-compatibility policy can be found [here](https://github.com/python ## NEXT +- Fix tagged unions with `forbid_extra_keys=True` dropping tags that are also member fields, both with and without a default member. + ([#483](https://github.com/python-attrs/cattrs/issues/483)) + - Fix the msgspec converter passing typed deques directly to the encoder, which cannot serialize them. Deques are now unstructured into lists, including when nested in attrs classes or dataclasses. ([#818](https://github.com/python-attrs/cattrs/issues/818) [#819](https://github.com/python-attrs/cattrs/pull/819)) diff --git a/src/cattrs/strategies/_unions.py b/src/cattrs/strategies/_unions.py index 52ac6a95..1c3caa9d 100644 --- a/src/cattrs/strategies/_unions.py +++ b/src/cattrs/strategies/_unions.py @@ -2,9 +2,16 @@ from typing import Any, Callable, Union from attrs import NOTHING, NothingType +from attrs.exceptions import NotAnAttrsClassError from .. import BaseConverter -from .._compat import get_newtype_base, is_literal, is_subclass, is_union_type +from .._compat import ( + adapted_fields, + get_newtype_base, + is_literal, + is_subclass, + is_union_type, +) from ..sentinels import is_sentinel from ..typealiases import get_type_alias_base, is_type_alias @@ -20,6 +27,18 @@ def default_tag_generator(typ: type) -> str: return typ.__name__ +def _class_consumes_key(cl: type, key: str) -> bool: + """Whether an init field consumes the key by name or private-field alias.""" + try: + fields = adapted_fields(cl) + except (TypeError, NotAnAttrsClassError): + return False + return any( + field.init and (field.name == key or field.name.removeprefix("_") == key) + for field in fields + ) + + def configure_tagged_union( union: Any, converter: BaseConverter, @@ -53,6 +72,7 @@ def configure_tagged_union( if is_type_alias(union): union = union.__value__ args = union.__args__ + forbid_extra_keys = getattr(converter, "forbid_extra_keys", False) tag_to_hook = {} exact_cl_unstruct_hooks = {} @@ -67,7 +87,7 @@ def structure_default(val: dict, _cl=default, _h=default_handler): tag_to_hook = defaultdict(lambda: structure_default) cl_to_tag = defaultdict(lambda: default) - if getattr(converter, "forbid_extra_keys", False): + if forbid_extra_keys: def structure_tagged_union( val: dict, @@ -78,8 +98,11 @@ def structure_tagged_union( _default=default, ) -> union: if _tag_name in val: + tag = val[_tag_name] + if tag in _tag_to_hook: + return _tag_to_hook[tag](val) val = val.copy() - return _tag_to_hook[val.pop(_tag_name)](val) + val.pop(_tag_name) return _dh(val, _default) else: @@ -97,20 +120,11 @@ def structure_tagged_union( return _dh(val, _default) else: - if getattr(converter, "forbid_extra_keys", False): - - def structure_tagged_union( - val: dict, _, _tag_to_cl=tag_to_hook, _tag_name=tag_name - ) -> union: - val = val.copy() - return _tag_to_cl[val.pop(_tag_name)](val) - else: - - def structure_tagged_union( - val: dict, _, _tag_to_cl=tag_to_hook, _tag_name=tag_name - ) -> union: - return _tag_to_cl[val[_tag_name]](val) + def structure_tagged_union( + val: dict, _, _tag_to_cl=tag_to_hook, _tag_name=tag_name + ) -> union: + return _tag_to_cl[val[_tag_name]](val) def unstructure_tagged_union( val: union, @@ -129,8 +143,19 @@ def unstructure_tagged_union( tag = tag_generator(cl) struct_handler = converter.get_structure_hook(cl) unstruct_handler = converter.get_unstructure_hook(cl) - - def structure_union_member(val: dict, _cl=cl, _h=struct_handler) -> cl: + consumes_tag = _class_consumes_key(cl, tag_name) if forbid_extra_keys else False + + def structure_union_member( + val: dict, + _cl=cl, + _h=struct_handler, + _consumes_tag=consumes_tag, + _forbid_extra_keys=forbid_extra_keys, + _tag_name=tag_name, + ) -> cl: + if _forbid_extra_keys and not _consumes_tag and _tag_name in val: + val = val.copy() + val.pop(_tag_name) return _h(val, _cl) def unstructure_union_member(val: union, _h=unstruct_handler) -> dict: diff --git a/tests/strategies/test_tagged_unions.py b/tests/strategies/test_tagged_unions.py index ba5ae49b..ec41e763 100644 --- a/tests/strategies/test_tagged_unions.py +++ b/tests/strategies/test_tagged_unions.py @@ -1,6 +1,7 @@ from typing import Optional, Sequence, Union -from attrs import define +import pytest +from attrs import NOTHING, define, field from cattrs import BaseConverter, Converter from cattrs.strategies import configure_tagged_union @@ -142,8 +143,11 @@ class B: c = Converter(forbid_extra_keys=True) configure_tagged_union(Union[A, B], c) - data = c.unstructure(A(), Union[A, B]) - c.structure(data, Union[A, B]) + for instance in (A(), B()): + data = c.unstructure(instance, Union[A, B]) + original = data.copy() + assert c.structure(data, Union[A, B]) == instance + assert data == original def test_forbid_extra_keys_default(): @@ -160,13 +164,145 @@ class B: c = Converter(forbid_extra_keys=True) configure_tagged_union(Union[A, B], c, default=A) - data = c.unstructure(A(), Union[A, B]) - assert c.structure(data, Union[A, B]) == A() + for instance in (A(), B()): + data = c.unstructure(instance, Union[A, B]) + original = data.copy() + assert c.structure(data, Union[A, B]) == instance + assert data == original + + data = {"_type": "unknown"} + # Repeated unknown tags must keep using the same fallback behavior. + for _ in range(2): + assert c.structure(data, Union[A, B]) == A() + assert data == {"_type": "unknown"} data.pop("_type") assert c.structure(data, Union[A, B]) == A() +@pytest.mark.parametrize("use_default", [False, True]) +@pytest.mark.parametrize("forbid_extra_keys", [False, True]) +def test_tag_is_member_field(use_default, forbid_extra_keys): + """A tag that is also a member field survives structuring and round-trips.""" + + @define + class SpecialDevice: + hostname: str + port: int + type: str + + @define + class StandardDevice: + hostname: str + type: str + + union = SpecialDevice | StandardDevice + + @define + class Settings: + device: union + + c = Converter(forbid_extra_keys=forbid_extra_keys) + configure_tagged_union( + union, + c, + tag_name="type", + tag_generator={SpecialDevice: "special", StandardDevice: "standard"}.get, + default=StandardDevice if use_default else NOTHING, + ) + + for payload, expected in ( + ( + {"type": "special", "hostname": "hostname", "port": 1234}, + SpecialDevice("hostname", 1234, "special"), + ), + ( + {"type": "standard", "hostname": "hostname"}, + StandardDevice("hostname", "standard"), + ), + ): + data = {"device": payload.copy()} + result = c.structure(data, Settings) + assert result == Settings(expected) + assert data == {"device": payload} + assert c.unstructure(result) == data + + +@pytest.mark.parametrize("use_default", [False, True]) +@pytest.mark.parametrize("use_alias, tag_name", [(False, "_type"), (True, "type")]) +def test_forbid_extra_keys_mixed_members(use_default, use_alias, tag_name): + """Only members consuming the tag retain it, including private field aliases.""" + + @define + class C: + _type: str + + union = C | A + c = Converter(forbid_extra_keys=True, use_alias=use_alias) + configure_tagged_union( + union, c, tag_name=tag_name, default=C if use_default else NOTHING + ) + + for data, expected in (({tag_name: "C"}, C("C")), ({tag_name: "A", "a": 1}, A(1))): + original = data.copy() + result = c.structure(data, union) + assert result == expected + assert data == original + assert c.unstructure(result, union) == data + + +def test_forbid_extra_keys_init_false_tag(): + """Fields excluded from init do not consume the tag.""" + + @define + class C: + type: str = field(init=False, default="unused") + + union = C | A + c = Converter(forbid_extra_keys=True) + configure_tagged_union(union, c, tag_name="type") + + assert c.structure({"type": "C"}, union) == C() + + +def test_forbid_extra_keys_default_with_tag_field(): + """Unknown tags are still removed before invoking the default handler.""" + + @define + class C: + type: str = "fallback" + + union = C | A + c = Converter(forbid_extra_keys=True) + configure_tagged_union(union, c, tag_name="type", default=C) + + assert c.structure({"type": "C"}, union) == C("C") + assert c.structure({}, union) == C() + data = {"type": "unknown"} + for _ in range(2): + assert c.structure(data, union) == C() + assert data == {"type": "unknown"} + + +def test_forbid_extra_keys_custom_member(): + """Members with custom hooks need not be attrs classes or dataclasses.""" + + class C: + pass + + c = Converter(forbid_extra_keys=True) + + @c.register_structure_hook + def structure_c(val, _) -> C: + assert val == {} + return C() + + c.register_unstructure_hook(C, lambda _: {}) + configure_tagged_union(C | A, c) + + assert isinstance(c.structure({"_type": "C"}, C | A), C) + + def test_nested_sequence_union(): @define class Top: