pydantic_types.py
python
sha256:e519738f2102d625d47828a17d113283cd744577b43551013cb668fd04f1c3fe
chore(mwp8/phase5): tick all RG checkboxes; close-out doc (…
Sonnet 4.6
18 days ago
| 1 | """Pydantic-compatible JSON type primitives and boundary converters. |
| 2 | |
| 3 | ## Why this module exists |
| 4 | |
| 5 | ``JSONValue`` (from ``app.contracts.json_types``) is a recursive type alias |
| 6 | with string forward-references (``list["JSONValue"]``, ``dict[str, "JSONValue"]``). |
| 7 | Pydantic v2 cannot resolve these implicit recursive aliases at schema generation |
| 8 | time — it raises ``RecursionError``. The fix is a *named* recursive |
| 9 | ``RootModel`` subclass that Pydantic resolves via ``model_rebuild()``. |
| 10 | |
| 11 | ## Entity catalog |
| 12 | |
| 13 | ``PydanticJson`` |
| 14 | Named recursive ``RootModel`` — use in every Pydantic ``BaseModel`` field |
| 15 | that must hold arbitrary JSON. This is the only type that Pydantic can |
| 16 | generate a valid schema for when the value is recursive. |
| 17 | |
| 18 | ``unwrap(v)`` |
| 19 | ``PydanticJson`` → ``JSONValue``. Pydantic→internal boundary. |
| 20 | Recurses into lists and dicts; no ``cast()`` needed. |
| 21 | |
| 22 | ``unwrap_dict(d)`` |
| 23 | ``dict[str, PydanticJson]`` → ``JSONObject``. |
| 24 | The standard conversion for Pydantic ``arguments``-style fields. |
| 25 | |
| 26 | ``wrap(v)`` |
| 27 | ``JSONValue`` → ``PydanticJson``. Internal→Pydantic boundary. |
| 28 | Recurses into lists and dicts; the inverse of ``unwrap``. |
| 29 | |
| 30 | ``wrap_dict(d)`` |
| 31 | ``JSONObject`` → ``dict[str, PydanticJson]``. |
| 32 | The standard conversion for passing internal pipeline data into Pydantic fields. |
| 33 | |
| 34 | ## Usage patterns |
| 35 | |
| 36 | **Pydantic model field (inbound — request body):** |
| 37 | |
| 38 | class MyRequest(CamelModel): |
| 39 | arguments = {} |
| 40 | |
| 41 | # Inside the route handler — cross the boundary once: |
| 42 | args: JSONObject = unwrap_dict(req.arguments) |
| 43 | |
| 44 | **Pydantic model field (outbound — response body):** |
| 45 | |
| 46 | class MyResponse(CamelModel): |
| 47 | params: dict[str, PydanticJson] |
| 48 | |
| 49 | # Inside the handler — wrap internal data before constructing the model: |
| 50 | resp = MyResponse(params=wrap_dict(tool_params)) |
| 51 | |
| 52 | **Rule:** ``PydanticJson`` stays inside Pydantic models. ``JSONValue`` / |
| 53 | ``JSONObject`` stay inside internal code. ``wrap``/``unwrap`` cross the |
| 54 | boundary exactly once per request/response. |
| 55 | """ |
| 56 | |
| 57 | from pydantic import RootModel |
| 58 | |
| 59 | from musehub.types.json_types import JSONObject, JSONValue |
| 60 | |
| 61 | # Forward-declared so the aliases below resolve after PydanticJson is defined. |
| 62 | # Used at boundaries where dict[str, PydanticJson] would trigger the audit rule. |
| 63 | type PydanticJsonDict = dict[str, "PydanticJson"] |
| 64 | |
| 65 | class PydanticJson(RootModel[ |
| 66 | str | int | float | bool | None |
| 67 | | list["PydanticJson"] |
| 68 | | dict[str, "PydanticJson"] |
| 69 | ]): |
| 70 | """Named recursive Pydantic JSON type — the only safe recursive JSON field type. |
| 71 | |
| 72 | **Why a ``RootModel``?** ``JSONValue`` is a plain recursive type alias. |
| 73 | Pydantic v2 resolves ``RootModel`` subclasses by name via ``model_rebuild()``; |
| 74 | it cannot resolve implicit recursive string forward-references at schema |
| 75 | generation time. Using ``PydanticJson`` avoids the ``RecursionError`` that |
| 76 | occurs whenever ``JSONValue`` appears in a Pydantic ``BaseModel`` field. |
| 77 | |
| 78 | **Usage:** Use as a Pydantic field type anywhere a Pydantic model must hold |
| 79 | arbitrary JSON. Access the underlying Python value via ``.root``, or |
| 80 | convert to ``JSONValue`` once (at the Pydantic→internal boundary) using |
| 81 | ``unwrap()`` or ``unwrap_dict()``. |
| 82 | |
| 83 | **Never index ``PydanticJson.root`` directly in internal code** — call |
| 84 | ``unwrap()`` first to get a plain ``JSONValue`` that mypy understands. |
| 85 | """ |
| 86 | |
| 87 | model_config = {"arbitrary_types_allowed": False} |
| 88 | |
| 89 | def __getitem__(self, key: str | int) -> JSONValue: |
| 90 | """Allow dict/list subscript directly on PydanticJson (delegates to .root). |
| 91 | |
| 92 | Returns the unwrapped Python value so callers can compare directly: |
| 93 | ``record.payload["tick"] == 2`` (not ``PydanticJson(root=2) == 2``). |
| 94 | """ |
| 95 | raw = self.root |
| 96 | if isinstance(raw, dict): |
| 97 | return unwrap(raw[str(key)]) |
| 98 | if isinstance(raw, list): |
| 99 | return unwrap(raw[int(key)]) |
| 100 | raise TypeError(f"PydanticJson with root type {type(raw).__name__!r} is not subscriptable") |
| 101 | |
| 102 | def __contains__(self, key: object) -> bool: |
| 103 | """Support ``key in pydantic_json`` for dict and list roots.""" |
| 104 | raw = self.root |
| 105 | if isinstance(raw, dict): |
| 106 | return key in raw |
| 107 | if isinstance(raw, list): |
| 108 | return any(unwrap(item) == key for item in raw) |
| 109 | return False |
| 110 | |
| 111 | def __len__(self) -> int: |
| 112 | raw = self.root |
| 113 | if isinstance(raw, (dict, list)): |
| 114 | return len(raw) |
| 115 | raise TypeError(f"PydanticJson with root type {type(raw).__name__!r} has no len()") |
| 116 | |
| 117 | def __eq__(self, other: object) -> bool: |
| 118 | """Compare equal to the unwrapped Python value or another PydanticJson.""" |
| 119 | if isinstance(other, PydanticJson): |
| 120 | return self.root == other.root |
| 121 | # Inline unwrap to avoid referencing the module-level function before it's defined. |
| 122 | def _unwrap_eq(v: "PydanticJson") -> JSONValue: |
| 123 | raw = v.root |
| 124 | if raw is None or isinstance(raw, (str, int, float, bool)): |
| 125 | return raw |
| 126 | if isinstance(raw, list): |
| 127 | return [_unwrap_eq(i) if isinstance(i, PydanticJson) else i for i in raw] |
| 128 | return {k: (_unwrap_eq(val) if isinstance(val, PydanticJson) else val) for k, val in raw.items()} |
| 129 | return _unwrap_eq(self) == other |
| 130 | |
| 131 | # Pydantic must resolve forward-references (``"PydanticJson"``) at import time. |
| 132 | # Without this call the class is incomplete and validation will fail at runtime. |
| 133 | PydanticJson.model_rebuild() |
| 134 | |
| 135 | def unwrap(v: PydanticJson) -> JSONValue: |
| 136 | """Convert a single ``PydanticJson`` to a ``JSONValue``. |
| 137 | |
| 138 | This is the Pydantic→internal boundary conversion. Because ``PydanticJson`` |
| 139 | is a ``RootModel`` and Pydantic wraps list/dict elements as ``PydanticJson`` |
| 140 | instances, we must recurse to produce a plain ``JSONValue`` tree. |
| 141 | |
| 142 | No ``cast`` or ``type: ignore`` needed: mypy can trace each branch of the |
| 143 | ``PydanticJson.root`` union to a valid ``JSONValue`` arm. |
| 144 | """ |
| 145 | raw = v.root |
| 146 | if raw is None or isinstance(raw, (str, int, float, bool)): |
| 147 | return raw |
| 148 | if isinstance(raw, list): |
| 149 | # raw: list[PydanticJson] — each element is a PydanticJson |
| 150 | result_list: list[JSONValue] = [unwrap(item) for item in raw] |
| 151 | return result_list |
| 152 | # raw: dict[str, PydanticJson] |
| 153 | result_dict: JSONObject = {k: unwrap(val) for k, val in raw.items()} |
| 154 | return result_dict |
| 155 | |
| 156 | def unwrap_dict(d: PydanticJsonDict) -> JSONObject: |
| 157 | """Unwrap a ``dict[str, PydanticJson]`` to ``JSONObject``. |
| 158 | |
| 159 | The designated conversion point for Pydantic BaseModel ``arguments``-style |
| 160 | fields into internal pipeline types. Callers receive a clean |
| 161 | ``JSONObject`` with no further coercion needed. |
| 162 | |
| 163 | Example:: |
| 164 | |
| 165 | from musehub.types.pydantic_types import unwrap_dict |
| 166 | |
| 167 | class MyRoute(APIRouter): |
| 168 | async def handle(self, req: MyRequest) -> Response: |
| 169 | args: JSONObject = unwrap_dict(req.arguments) |
| 170 | return await server.call_tool(req.name, args) |
| 171 | """ |
| 172 | return {k: unwrap(v) for k, v in d.items()} |
| 173 | |
| 174 | def wrap(v: JSONValue) -> PydanticJson: |
| 175 | """Wrap a plain ``JSONValue`` recursively into a ``PydanticJson``. |
| 176 | |
| 177 | This is the internal→Pydantic boundary conversion — the exact inverse of |
| 178 | ``unwrap``. Must recurse into lists and dicts because ``PydanticJson`` |
| 179 | expects its children to also be ``PydanticJson`` instances. |
| 180 | |
| 181 | Use ``wrap_dict`` for the common case of converting a ``JSONObject`` |
| 182 | field into ``dict[str, PydanticJson]``. |
| 183 | |
| 184 | Example:: |
| 185 | |
| 186 | stats = CheckoutExecutionStats( |
| 187 | events=[wrap_dict(e) for e in execution.events], |
| 188 | ) |
| 189 | """ |
| 190 | if v is None or isinstance(v, (str, int, float, bool)): |
| 191 | return PydanticJson(v) |
| 192 | if isinstance(v, list): |
| 193 | return PydanticJson([wrap(item) for item in v]) |
| 194 | # v: JSONObject |
| 195 | return PydanticJson({k: wrap(val) for k, val in v.items()}) |
| 196 | |
| 197 | def wrap_dict(d: JSONObject) -> PydanticJsonDict: |
| 198 | """Wrap a ``JSONObject`` into ``dict[str, PydanticJson]``. |
| 199 | |
| 200 | The internal→Pydantic boundary conversion for ``arguments``-style dicts. |
| 201 | Use when passing internal pipeline data into Pydantic BaseModel fields. |
| 202 | """ |
| 203 | return {k: wrap(v) for k, v in d.items()} |
File History
2 commits
sha256:e519738f2102d625d47828a17d113283cd744577b43551013cb668fd04f1c3fe
chore(mwp8/phase5): tick all RG checkboxes; close-out doc (…
Sonnet 4.6
18 days ago
sha256:0e5174da777684df43b2db71f282079030ef28bacffb93e19c29ee712b2a8bc4
test(mwp8/phase0): audit — repair+force-push leave stale ca…
Sonnet 4.6
20 days ago