From ba9c322f8dcbc25cbc10ff7d31adf28441f71997 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 12 Oct 2026 16:25:46 -0400 Subject: [PATCH] 0.3.0: a field that admits several classes stores its value under a member tag A dataclass field whose declared type is a union of dataclasses, an abstract dataclass or a registered class is written as {tag: fields} and read back by the tag. The tag is the member's registry key when it is registered, otherwise its class name. Before, a union member was guessed from the fields present: two members with the same fields were ambiguous and the stored value did not say which class it was. An unknown tag is an error naming the known members. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 7 + pyproject.toml | 2 +- src/python_introspect/__init__.py | 4 +- src/python_introspect/dataclass_projection.py | 26 +- src/python_introspect/jsonable.py | 238 +++++++++++++++++- tests/test_dataclass_projection.py | 17 -- tests/test_jsonable.py | 200 ++++++++++++++- 7 files changed, 467 insertions(+), 27 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f5035a9..cb230b0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ 0.2.2 is the first 0.2.x release published to PyPI. The 0.2.0 and 0.2.1 tags exist in git, but their publish runs were cancelled, so upgrading from PyPI goes straight from 0.1.16 to 0.2.2. The notes below cover every change in that jump. +## 0.3.0 + +Changed (the stored form of some fields changes; there is no reader for the old form): +- A dataclass field whose declared type admits several classes is stored as `{tag: fields}`, and `dataclass_from_mapping` reads the class from the tag. Before, the member of a union was guessed from which fields were present: two members with the same fields could not be told apart and the stored value did not say which class it was. Three declarations form such a family (`MemberFamily.declared_by`): a union with two or more dataclass members (other arms such as `None` or `str` stay untagged), an abstract dataclass (its concrete subclasses are the members), and a class whose metaclass is `AutoRegisterMeta`. The tag is the member's registry key when it is registered, so it survives a module move or a rename; otherwise it is the class name. The rule holds inside `tuple`, `list`, `Sequence`, `Mapping`, optional and `Annotated` declarations. +- An unknown tag is a `ValueError` naming the known members. A value whose class is not a member of the declared family is a `TypeError` when encoding. Two members with the same tag are a `TypeError`. +- `to_jsonable` reads the declared field types of every dataclass it encodes, so those annotations must resolve at run time (a name imported only under `TYPE_CHECKING` no longer encodes). + ## 0.2.8 Added: diff --git a/pyproject.toml b/pyproject.toml index 1fa66e6..5e91acb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "python-introspect" -version = "0.2.8" +version = "0.3.0" description = "Pure Python introspection toolkit for function signatures, dataclasses, and type hints" readme = "README.md" requires-python = ">=3.10" diff --git a/src/python_introspect/__init__.py b/src/python_introspect/__init__.py index 8f09aa4..7eebd2b 100644 --- a/src/python_introspect/__init__.py +++ b/src/python_introspect/__init__.py @@ -9,7 +9,7 @@ type resolution for framework-specific types (lazy configs, proxies, etc.) """ -__version__ = "0.2.8" +__version__ = "0.3.0" from .signature_analyzer import ( SignatureAnalyzer, @@ -57,6 +57,7 @@ JsonScalar, JsonValue, RegisteredClassChoices, + MemberFamily, registered_class_from_key, to_jsonable, ) @@ -135,6 +136,7 @@ "JsonObject", "JsonScalar", "JsonValue", + "MemberFamily", "registered_class_from_key", "to_jsonable", # Module public surfaces diff --git a/src/python_introspect/dataclass_projection.py b/src/python_introspect/dataclass_projection.py index 2a8ec9d..9d82e15 100644 --- a/src/python_introspect/dataclass_projection.py +++ b/src/python_introspect/dataclass_projection.py @@ -18,7 +18,7 @@ from metaclass_registry import AutoRegisterMeta from .annotation_types import is_union_type, resolved_class_annotations -from .jsonable import registered_class_from_key +from .jsonable import MemberFamily, registered_class_from_key from .validation import validate_annotated_dataclass, validate_annotation_value DataclassT = TypeVar("DataclassT") @@ -157,10 +157,32 @@ def _mapping_value_for_annotation( if not isinstance(value, (list, tuple)): raise TypeError(f"{path} must be an array.") return tuple(value) + family = MemberFamily.declared_by(annotation) + if family is not None and isinstance(value, Mapping): + if len(value) != 1: + raise ValueError( + f"{path} must be one member of {family.name} stored as " + f"{{tag: fields}}; got the keys {sorted(map(str, value))}." + ) + ((tag, member_fields),) = value.items() + try: + member = family.member_tagged(tag) + except ValueError as error: + raise ValueError(f"{path}: {error}") from error + if not isinstance(member_fields, Mapping): + raise TypeError(f"{path}.{tag} must be an object.") + return dataclass_from_mapping(member, member_fields) if is_union_type(annotation): successes: list[object] = [] errors: list[Exception] = [] - for member_type in get_args(annotation): + untagged_arms = tuple( + arm + for arm in get_args(annotation) + if family is None or not (isinstance(arm, type) and is_dataclass(arm)) + ) + if family is not None and isinstance(value, family.members()): + return value + for member_type in untagged_arms: try: converted = _mapping_value_for_annotation( member_type, diff --git a/src/python_introspect/jsonable.py b/src/python_introspect/jsonable.py index ec5943f..4daee51 100644 --- a/src/python_introspect/jsonable.py +++ b/src/python_introspect/jsonable.py @@ -10,20 +10,29 @@ annotated ``type[Family]``. A family whose boundary spelling is not its registry (for example one resolved through an active subset) registers its own decoder on its metaclass. + +A dataclass field whose declared type admits several classes (``MemberFamily``: +a union of dataclasses, an abstract dataclass, or a registered class) holds its +value as ``{tag: fields}``. The tag is the member's registry key when its +metaclass registers it and its class name otherwise, so the stored form names +the class instead of leaving the decoder to guess it from the fields present. """ from __future__ import annotations import inspect -from collections.abc import Callable, Mapping +from abc import ABC, abstractmethod +from collections.abc import Callable, Mapping, Sequence, Set from dataclasses import dataclass, fields, is_dataclass from enum import Enum from functools import singledispatch from pathlib import Path -from typing import ForwardRef, TypeAlias +from typing import Annotated, ForwardRef, TypeAlias, get_args, get_origin +from weakref import WeakKeyDictionary from metaclass_registry import AutoRegisterMeta +from .annotation_types import is_union_type, resolved_class_annotations from .choices import AnnotationChoices from .signature_analyzer import signature_analysis_target @@ -40,15 +49,94 @@ JsonObject: TypeAlias = Mapping[str, JsonValue] +_FIELD_TYPES_BY_DATACLASS: WeakKeyDictionary[type, Mapping[str, object]] = WeakKeyDictionary() + + +def _declared_field_types(owner: type) -> Mapping[str, object]: + if owner not in _FIELD_TYPES_BY_DATACLASS: + try: + _FIELD_TYPES_BY_DATACLASS[owner] = resolved_class_annotations(owner) + except NameError as error: + raise TypeError( + f"{owner.__qualname__} cannot be projected: a field type does not resolve " + f"at run time ({error}). A field's declared type decides its stored form, " + "so import the name outside TYPE_CHECKING." + ) from error + return _FIELD_TYPES_BY_DATACLASS[owner] + + def _jsonable_dataclass(value: object) -> JsonValue: - """Project fields without ``asdict`` deep-copying immutable containers.""" + """Project each field by its declared type, without ``asdict`` deep copies.""" + declared_types = _declared_field_types(type(value)) return { - declared_field.name: to_jsonable(getattr(value, declared_field.name)) + declared_field.name: _jsonable_as_declared( + declared_types[declared_field.name], getattr(value, declared_field.name) + ) for declared_field in fields(value) } +def _projects_by_declared_fields(value: object) -> bool: + """Whether ``to_jsonable`` writes ``value`` as its dataclass fields. + + A type with its own registered projection owns its whole stored form, + including how that form names the class. + """ + + return ( + is_dataclass(value) + and not isinstance(value, type) + and to_jsonable.dispatch(type(value)) in (to_jsonable.registry[object], _jsonable_callable) + ) + + +def _jsonable_as_declared(annotation: object, value: object) -> JsonValue: + """Project ``value`` as a field declared ``annotation`` stores it. + + The declaration is followed through optional, annotated and container types + to the places where it admits several classes; a value there is wrapped in + its member tag. Everything else projects by its own type. + """ + + if value is None: + return None + origin = get_origin(annotation) + if origin is Annotated: + return _jsonable_as_declared(get_args(annotation)[0], value) + family = MemberFamily.declared_by(annotation) + if family is not None and _projects_by_declared_fields(value): + return {family.tag_of(type(value)): to_jsonable(value)} + if is_union_type(annotation): + arms = tuple( + arm + for arm in get_args(annotation) + if isinstance(runtime_class := get_origin(arm) or arm, type) + and runtime_class is not type(None) + and isinstance(value, runtime_class) + ) + return _jsonable_as_declared(arms[0], value) if len(arms) == 1 else to_jsonable(value) + item_types = get_args(annotation) + if ( + isinstance(origin, type) + and item_types + and isinstance(value, origin) + and not isinstance(value, (str, bytes, bytearray)) + ): + if issubclass(origin, Mapping) and len(item_types) == 2: + return { + str(to_jsonable(key)): _jsonable_as_declared(item_types[1], item) + for key, item in value.items() + } + if origin is tuple and Ellipsis not in item_types and len(item_types) == len(value): + return [ + _jsonable_as_declared(item_type, item) for item_type, item in zip(item_types, value) + ] + if issubclass(origin, (Sequence, Set)): + return [_jsonable_as_declared(item_types[0], item) for item in value] + return to_jsonable(value) + + @singledispatch def to_jsonable(value: object) -> JsonValue: """Project dataclasses and registered values into JSON-native data.""" @@ -100,7 +188,7 @@ def _jsonable_registered_type(value: AutoRegisterMeta) -> JsonValue: ) if declaring_owner is None: raise TypeError( - f"Registered type {value.__qualname__} has no declared " f"{key_attribute!r} key." + f"Registered type {value.__qualname__} has no declared {key_attribute!r} key." ) key = vars(declaring_owner)[key_attribute] if key is None: @@ -149,6 +237,146 @@ def _registered_class_from_registry(family: AutoRegisterMeta, key: JsonValue) -> return matches[0] +class MemberFamily(ABC): + """The classes a declared field type admits, each named by one tag. + + A value of such a field is stored as ``{tag: fields}``. Three declarations + form a family: a union with two or more dataclass members, an abstract + dataclass (its concrete subclasses are the members) and a class registered + by ``AutoRegisterMeta`` (the registry's classes are the members). + """ + + @staticmethod + def declared_by(annotation: object) -> MemberFamily | None: + """The family ``annotation`` declares, or None when it names one class.""" + + if is_union_type(annotation): + members = tuple( + member + for member in get_args(annotation) + if isinstance(member, type) and is_dataclass(member) + ) + return UnionMembers(members) if len(members) > 1 else None + if isinstance(annotation, AutoRegisterMeta): + return RegisteredMembers(annotation) + if ( + isinstance(annotation, type) + and is_dataclass(annotation) + and inspect.isabstract(annotation) + ): + return ConcreteSubclasses(annotation) + return None + + @property + @abstractmethod + def name(self) -> str: + """How error messages refer to the family.""" + + @abstractmethod + def members(self) -> tuple[type, ...]: + """The classes a value of the field may be.""" + + def member_tagged(self, tag: str) -> type: + """The member a stored tag names.""" + + members = self._members_by_tag() + if tag not in members: + raise ValueError(f"{tag!r} is not a member of {self.name}; members: {sorted(members)}.") + return members[tag] + + def tag_of(self, member: type) -> str: + """The tag that names ``member`` in stored data.""" + + tags = {known: tag for tag, known in self._members_by_tag().items()} + if member not in tags: + raise TypeError( + f"{member.__qualname__} is not a member of {self.name}; " + f"members: {sorted(tags.values())}." + ) + return tags[member] + + def _members_by_tag(self) -> dict[str, type]: + members: dict[str, type] = {} + for member in self.members(): + tag = class_tag(member) + if members.setdefault(tag, member) is not member: + raise TypeError( + f"{self.name} has two members tagged {tag!r}: " + f"{members[tag].__qualname__} and {member.__qualname__}." + ) + return members + + +def class_tag(member: type) -> str: + """The tag naming ``member``: its registry key if it is registered, else its name.""" + + tag = to_jsonable(member) if isinstance(member, AutoRegisterMeta) else member.__name__ + if not isinstance(tag, str): + raise TypeError( + f"{member.__qualname__} has the registry key {tag!r}; a member tag is a string." + ) + return tag + + +@dataclass(frozen=True) +class UnionMembers(MemberFamily): + """The dataclasses listed in a union; its other arms are stored untagged. + + An arm that is itself a family contributes that family's members. + """ + + listed: tuple[type, ...] + + @property + def name(self) -> str: + return " | ".join(member.__qualname__ for member in self.listed) + + def members(self) -> tuple[type, ...]: + members: list[type] = [] + for arm in self.listed: + family = MemberFamily.declared_by(arm) + members.extend((arm,) if family is None else family.members()) + return tuple(members) + + +@dataclass(frozen=True) +class ConcreteSubclasses(MemberFamily): + """The concrete subclasses of an abstract dataclass, as loaded when asked.""" + + base: type + + @property + def name(self) -> str: + return self.base.__qualname__ + + def members(self) -> tuple[type, ...]: + found: dict[type, None] = {} + pending = list(self.base.__subclasses__()) + while pending: + subclass = pending.pop(0) + pending.extend(subclass.__subclasses__()) + if not inspect.isabstract(subclass): + found[subclass] = None + return tuple(found) + + +@dataclass(frozen=True) +class RegisteredMembers(ConcreteSubclasses): + """The classes a registered family's registry holds, tagged by registry key. + + A stored tag is read by ``registered_class_from_key``, so a family with its + own spelling decodes tags the way it decodes ``type[Family]`` fields. + """ + + def members(self) -> tuple[type, ...]: + return tuple( + member for member in self.base.__registry__.values() if issubclass(member, self.base) + ) + + def member_tagged(self, tag: str) -> type: + return registered_class_from_key(self.base, tag) + + @dataclass(frozen=True) class RegisteredClassChoices(AnnotationChoices): """A field holding one class of a registered family. diff --git a/tests/test_dataclass_projection.py b/tests/test_dataclass_projection.py index 3817038..da5a087 100644 --- a/tests/test_dataclass_projection.py +++ b/tests/test_dataclass_projection.py @@ -244,23 +244,6 @@ class NumericValue: assert type(dataclass_from_mapping(NumericValue, {"value": 1}).value) is int -def test_dataclass_mapping_rejects_ambiguous_structured_unions() -> None: - @dataclass(frozen=True) - class Left: - value: int - - @dataclass(frozen=True) - class Right: - value: int - - @dataclass(frozen=True) - class Envelope: - payload: Left | Right - - with pytest.raises(TypeError, match="ambiguously matches multiple members"): - dataclass_from_mapping(Envelope, {"payload": {"value": 1}}) - - @dataclass(frozen=True) class AnnotatedSequenceEnvelope: nodes: Annotated[Sequence["RecursiveSequenceNode"], "display text"] diff --git a/tests/test_jsonable.py b/tests/test_jsonable.py index 80b4ecb..0852bba 100644 --- a/tests/test_jsonable.py +++ b/tests/test_jsonable.py @@ -2,7 +2,10 @@ from __future__ import annotations -from dataclasses import dataclass, field +import json +from abc import ABC, abstractmethod +from collections.abc import Mapping +from dataclasses import dataclass, field, replace from enum import Enum from pathlib import Path, PurePath from typing import Annotated @@ -229,3 +232,198 @@ def test_registered_class_choices_validate_keys_in_pydantic(): adapter.validate_python("triangle") with pytest.raises(pydantic.ValidationError): adapter.validate_python(Triangle) + + +# A field whose declared type admits several classes stores {tag: fields}. + + +@dataclass(frozen=True) +class Left: + value: int + + +@dataclass(frozen=True) +class Right: + value: int + + +@dataclass(frozen=True) +class Measurement(ABC): + @abstractmethod + def unit(self) -> str: ... + + +@dataclass(frozen=True) +class Length(Measurement): + amount: float + + def unit(self) -> str: + return "m" + + +@dataclass(frozen=True) +class Duration(Measurement): + amount: float + + def unit(self) -> str: + return "s" + + +class Tool(metaclass=AutoRegisterMeta): + __registry_key__ = "tool_key" + tool_key = None + + +@dataclass(frozen=True) +class Hammer(Tool): + tool_key = "hammer" + weight: int + + +@dataclass(frozen=True) +class Saw(Tool): + tool_key = "saw" + weight: int + + +@dataclass(frozen=True) +class Envelope: + side: Left | Right + measurement: Measurement + tool: Tool + sides: tuple[Left | Right, ...] = () + measurements: Mapping[str, Measurement] = field(default_factory=dict) + spare: Tool | None = None + note: Left | Right | str | None = None + + +FAMILY_MEMBERS = ( + ("side", Left(1), {"Left": {"value": 1}}), + ("side", Right(1), {"Right": {"value": 1}}), + ("measurement", Length(2.0), {"Length": {"amount": 2.0}}), + ("measurement", Duration(2.0), {"Duration": {"amount": 2.0}}), + ("tool", Hammer(3), {"hammer": {"weight": 3}}), + ("tool", Saw(3), {"saw": {"weight": 3}}), +) + + +@pytest.mark.parametrize(("field_name", "member", "stored"), FAMILY_MEMBERS) +def test_each_family_member_is_stored_under_its_tag_and_reads_back(field_name, member, stored): + envelope = replace(Envelope(Left(0), Length(0.0), Hammer(0)), **{field_name: member}) + document = to_jsonable(envelope) + + assert document[field_name] == stored + decoded = dataclass_from_mapping(Envelope, json.loads(json.dumps(document))) + assert decoded == envelope + assert type(getattr(decoded, field_name)) is type(member) + + +def test_members_with_identical_fields_decode_to_the_class_the_tag_names(): + document = to_jsonable(Envelope(Right(1), Duration(1.0), Saw(1), sides=(Left(1), Right(1)))) + + decoded = dataclass_from_mapping(Envelope, document) + + assert [type(side) for side in (decoded.side, *decoded.sides)] == [Right, Left, Right] + assert type(decoded.measurement) is Duration + assert type(decoded.tool) is Saw + + +def test_family_fields_inside_containers_and_optional_fields_carry_the_tag(): + envelope = Envelope( + Left(1), + Length(1.0), + Hammer(1), + measurements={"wait": Duration(5.0)}, + spare=Saw(2), + note=Right(7), + ) + document = to_jsonable(envelope) + + assert document["measurements"] == {"wait": {"Duration": {"amount": 5.0}}} + assert document["spare"] == {"saw": {"weight": 2}} + assert document["note"] == {"Right": {"value": 7}} + assert dataclass_from_mapping(Envelope, document) == envelope + untagged_arms = replace(envelope, spare=None, note="text") + assert to_jsonable(untagged_arms)["note"] == "text" + assert dataclass_from_mapping(Envelope, to_jsonable(untagged_arms)) == untagged_arms + + +@pytest.mark.parametrize( + ("field_name", "stored", "known"), + ( + ("side", {"Middle": {"value": 1}}, r"\['Left', 'Right'\]"), + ("measurement", {"Mass": {"amount": 1.0}}, r"\['Duration', 'Length'\]"), + ("tool", {"drill": {"weight": 1}}, r"\['hammer', 'saw'\]"), + ), +) +def test_unknown_tag_is_an_error_naming_the_known_members(field_name, stored, known): + document = {**to_jsonable(Envelope(Left(1), Length(1.0), Hammer(1))), field_name: stored} + + with pytest.raises(ValueError, match=f"{next(iter(stored))!r} is not .*{known}"): + dataclass_from_mapping(Envelope, document) + + +def test_untagged_fields_are_not_matched_to_a_member_by_their_shape(): + document = to_jsonable(Envelope(Left(1), Length(1.0), Hammer(1))) + document = {**document, "measurement": {"amount": 1.0}} + + with pytest.raises(ValueError, match="'amount' is not a member of Measurement"): + dataclass_from_mapping(Envelope, document) + with pytest.raises(ValueError, match="stored as {tag: fields}"): + dataclass_from_mapping(Envelope, {**document, "measurement": {"amount": 1.0, "x": 2}}) + + +@dataclass(frozen=True) +class Other: + value: int + + +Namesake = dataclass(frozen=True)(type("Left", (), {"__annotations__": {"value": int}})) + + +@dataclass(frozen=True) +class Pair: + side: Left | Namesake + + +def test_a_value_outside_the_declared_family_is_not_encoded(): + with pytest.raises(TypeError, match=r"Other is not a member of Left \| Right"): + to_jsonable(Envelope(Other(1), Length(1.0), Hammer(1))) + + +def test_two_members_with_one_tag_are_rejected(): + with pytest.raises(TypeError, match="two members tagged 'Left'"): + to_jsonable(Pair(Left(1))) + + +class Probe(metaclass=AutoRegisterMeta): + __registry_key__ = "depth" + depth = None + + +@dataclass(frozen=True) +class DeepProbe(Probe): + depth = 3 + reach: int + + +to_jsonable.register(Probe, lambda probe: {"reach": probe.reach, "depth": probe.depth}) + + +@dataclass(frozen=True) +class Survey: + probe: Probe + + +def test_a_member_with_its_own_projection_keeps_its_stored_form(): + assert to_jsonable(Survey(DeepProbe(9))) == {"probe": {"reach": 9, "depth": 3}} + + +@dataclass(frozen=True) +class Unresolved: + value: NotImportedAtRunTime # noqa: F821 + + +def test_a_dataclass_whose_field_types_do_not_resolve_is_not_projected(): + with pytest.raises(TypeError, match="Unresolved cannot be projected.*NotImportedAtRunTime"): + to_jsonable(Unresolved(1))