|
2 | 2 |
|
3 | 3 | from __future__ import annotations |
4 | 4 |
|
| 5 | +from collections.abc import Mapping |
5 | 6 | from typing import Any, Protocol, TypeVar |
6 | 7 |
|
7 | | -from pydantic import BaseModel |
| 8 | +from pydantic import BaseModel, ValidationError |
8 | 9 |
|
9 | 10 | from jacobian.canonical import CanonicalizationError, encode_strict_json |
10 | 11 | from jacobian.capability_errors import CapabilityInvocationError |
|
19 | 20 | PreparedT = TypeVar("PreparedT") |
20 | 21 |
|
21 | 22 |
|
| 23 | +def _restore_json_tuple_paths( |
| 24 | + value: Any, |
| 25 | + paths: tuple[tuple[Any, ...], ...], |
| 26 | +) -> Any: |
| 27 | + """Restore tuple-shaped model fields after JSON parsing. |
| 28 | +
|
| 29 | + JSON arrays are the wire representation for both lists and tuples. Pydantic |
| 30 | + strict JSON validation accepts the former, but rejects some constrained |
| 31 | + tuple annotations even though their JSON representation is valid. Only the |
| 32 | + paths identified by that precise validation error are adapted; all scalar |
| 33 | + values remain subject to strict Python-mode validation below. |
| 34 | + """ |
| 35 | + |
| 36 | + if any(not path for path in paths): |
| 37 | + return tuple(value) if isinstance(value, list) else value |
| 38 | + |
| 39 | + if isinstance(value, Mapping): |
| 40 | + restored_mapping = dict(value) |
| 41 | + for key in restored_mapping: |
| 42 | + child_paths = tuple(path[1:] for path in paths if path and path[0] == key) |
| 43 | + if child_paths: |
| 44 | + restored_mapping[key] = _restore_json_tuple_paths( |
| 45 | + restored_mapping[key], child_paths |
| 46 | + ) |
| 47 | + return restored_mapping |
| 48 | + |
| 49 | + if isinstance(value, (list, tuple)): |
| 50 | + restored_sequence = list(value) |
| 51 | + for index in range(len(restored_sequence)): |
| 52 | + child_paths = tuple(path[1:] for path in paths if path and path[0] == index) |
| 53 | + if child_paths: |
| 54 | + restored_sequence[index] = _restore_json_tuple_paths( |
| 55 | + restored_sequence[index], child_paths |
| 56 | + ) |
| 57 | + return ( |
| 58 | + tuple(restored_sequence) if isinstance(value, tuple) else restored_sequence |
| 59 | + ) |
| 60 | + |
| 61 | + return value |
| 62 | + |
| 63 | + |
| 64 | +def _validate_strict_json_model[ModelT: BaseModel]( |
| 65 | + model: type[ModelT], encoded: bytes, wire_payload: object |
| 66 | +) -> ModelT: |
| 67 | + """Validate JSON strictly while honoring tuple fields' array wire form.""" |
| 68 | + |
| 69 | + try: |
| 70 | + parsed = model.model_validate_json(encoded, strict=True) |
| 71 | + except ValidationError as exc: |
| 72 | + # Pydantic's strict JSON path treats some constrained tuple fields as |
| 73 | + # Python-only tuples, although JSON arrays are their valid wire form. |
| 74 | + # Normalize only paths reported as tuple mismatches and retry the whole |
| 75 | + # model in strict Python mode. Nested tuple fields are reported only |
| 76 | + # after their containing tuple has been normalized, so repeat until the |
| 77 | + # model validates or a non-tuple error remains. |
| 78 | + normalized_payload = wire_payload |
| 79 | + for _ in range(64): |
| 80 | + errors = exc.errors() |
| 81 | + tuple_paths = tuple( |
| 82 | + tuple(error["loc"]) for error in errors if error["type"] == "tuple_type" |
| 83 | + ) |
| 84 | + if not tuple_paths: |
| 85 | + raise exc |
| 86 | + updated_payload = _restore_json_tuple_paths( |
| 87 | + normalized_payload, |
| 88 | + tuple_paths, |
| 89 | + ) |
| 90 | + if updated_payload == normalized_payload: |
| 91 | + raise exc |
| 92 | + normalized_payload = updated_payload |
| 93 | + try: |
| 94 | + parsed = model.model_validate(normalized_payload, strict=True) |
| 95 | + except ValidationError as retry_exc: |
| 96 | + exc = retry_exc |
| 97 | + else: |
| 98 | + return parsed |
| 99 | + raise exc |
| 100 | + return parsed |
| 101 | + |
| 102 | + |
22 | 103 | def parse_capability_input[ModelT: BaseModel]( |
23 | 104 | model: type[ModelT], payload: dict[str, Any] |
24 | 105 | ) -> ModelT: |
@@ -57,7 +138,7 @@ def parse_capability_input[ModelT: BaseModel]( |
57 | 138 | ), |
58 | 139 | ) |
59 | 140 | ) from exc |
60 | | - parsed = model.model_validate_json(encoded, strict=True) |
| 141 | + parsed = _validate_strict_json_model(model, encoded, wire_payload) |
61 | 142 | if typed_values: |
62 | 143 | # Parse all ordinary fields through the strict JSON boundary, then |
63 | 144 | # restore the already-validated port values. This preserves identity |
|
0 commit comments