pydantic_types.py
python
| 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]`` → ``dict[str, JSONValue]``. |
| 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 | ``dict[str, JSONValue]`` → ``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: dict[str, PydanticJson] = {} |
| 40 | |
| 41 | # Inside the route handler — cross the boundary once: |
| 42 | args: dict[str, JSONValue] = 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 __future__ import annotations |
| 58 | |
| 59 | from pydantic import RootModel |
| 60 | |
| 61 | from musehub.contracts.json_types import JSONValue |
| 62 | |
| 63 | |
| 64 | class PydanticJson(RootModel[ |
| 65 | str | int | float | bool | None |
| 66 | | list["PydanticJson"] |
| 67 | | dict[str, "PydanticJson"] |
| 68 | ]): |
| 69 | """Named recursive Pydantic JSON type — the only safe recursive JSON field type. |
| 70 | |
| 71 | **Why a ``RootModel``?** ``JSONValue`` is a plain recursive type alias. |
| 72 | Pydantic v2 resolves ``RootModel`` subclasses by name via ``model_rebuild()``; |
| 73 | it cannot resolve implicit recursive string forward-references at schema |
| 74 | generation time. Using ``PydanticJson`` avoids the ``RecursionError`` that |
| 75 | occurs whenever ``JSONValue`` appears in a Pydantic ``BaseModel`` field. |
| 76 | |
| 77 | **Usage:** Use as a Pydantic field type anywhere a Pydantic model must hold |
| 78 | arbitrary JSON. Access the underlying Python value via ``.root``, or |
| 79 | convert to ``JSONValue`` once (at the Pydantic→internal boundary) using |
| 80 | ``unwrap()`` or ``unwrap_dict()``. |
| 81 | |
| 82 | **Never index ``PydanticJson.root`` directly in internal code** — call |
| 83 | ``unwrap()`` first to get a plain ``JSONValue`` that mypy understands. |
| 84 | """ |
| 85 | |
| 86 | model_config = {"arbitrary_types_allowed": False} |
| 87 | |
| 88 | |
| 89 | # Pydantic must resolve forward-references (``"PydanticJson"``) at import time. |
| 90 | # Without this call the class is incomplete and validation will fail at runtime. |
| 91 | PydanticJson.model_rebuild() |
| 92 | |
| 93 | |
| 94 | def unwrap(v: PydanticJson) -> JSONValue: |
| 95 | """Convert a single ``PydanticJson`` to a ``JSONValue``. |
| 96 | |
| 97 | This is the Pydantic→internal boundary conversion. Because ``PydanticJson`` |
| 98 | is a ``RootModel`` and Pydantic wraps list/dict elements as ``PydanticJson`` |
| 99 | instances, we must recurse to produce a plain ``JSONValue`` tree. |
| 100 | |
| 101 | No ``cast`` or ``type: ignore`` needed: mypy can trace each branch of the |
| 102 | ``PydanticJson.root`` union to a valid ``JSONValue`` arm. |
| 103 | """ |
| 104 | raw = v.root |
| 105 | if raw is None or isinstance(raw, (str, int, float, bool)): |
| 106 | return raw |
| 107 | if isinstance(raw, list): |
| 108 | # raw: list[PydanticJson] — each element is a PydanticJson |
| 109 | result_list: list[JSONValue] = [unwrap(item) for item in raw] |
| 110 | return result_list |
| 111 | # raw: dict[str, PydanticJson] |
| 112 | result_dict: dict[str, JSONValue] = {k: unwrap(val) for k, val in raw.items()} |
| 113 | return result_dict |
| 114 | |
| 115 | |
| 116 | def unwrap_dict(d: dict[str, PydanticJson]) -> dict[str, JSONValue]: |
| 117 | """Unwrap a ``dict[str, PydanticJson]`` to ``dict[str, JSONValue]``. |
| 118 | |
| 119 | The designated conversion point for Pydantic BaseModel ``arguments``-style |
| 120 | fields into internal pipeline types. Callers receive a clean |
| 121 | ``dict[str, JSONValue]`` with no further coercion needed. |
| 122 | |
| 123 | Example:: |
| 124 | |
| 125 | from musehub.contracts.pydantic_types import unwrap_dict |
| 126 | |
| 127 | class MyRoute(APIRouter): |
| 128 | async def handle(self, req: MyRequest) -> Response: |
| 129 | args: dict[str, JSONValue] = unwrap_dict(req.arguments) |
| 130 | return await server.call_tool(req.name, args) |
| 131 | """ |
| 132 | return {k: unwrap(v) for k, v in d.items()} |
| 133 | |
| 134 | |
| 135 | def wrap(v: JSONValue) -> PydanticJson: |
| 136 | """Wrap a plain ``JSONValue`` recursively into a ``PydanticJson``. |
| 137 | |
| 138 | This is the internal→Pydantic boundary conversion — the exact inverse of |
| 139 | ``unwrap``. Must recurse into lists and dicts because ``PydanticJson`` |
| 140 | expects its children to also be ``PydanticJson`` instances. |
| 141 | |
| 142 | Use ``wrap_dict`` for the common case of converting a ``dict[str, JSONValue]`` |
| 143 | field into ``dict[str, PydanticJson]``. |
| 144 | |
| 145 | Example:: |
| 146 | |
| 147 | stats = CheckoutExecutionStats( |
| 148 | events=[wrap_dict(e) for e in execution.events], |
| 149 | ) |
| 150 | """ |
| 151 | if v is None or isinstance(v, (str, int, float, bool)): |
| 152 | return PydanticJson(v) |
| 153 | if isinstance(v, list): |
| 154 | return PydanticJson([wrap(item) for item in v]) |
| 155 | # v: dict[str, JSONValue] |
| 156 | return PydanticJson({k: wrap(val) for k, val in v.items()}) |
| 157 | |
| 158 | |
| 159 | def wrap_dict(d: dict[str, JSONValue]) -> dict[str, PydanticJson]: |
| 160 | """Wrap a ``dict[str, JSONValue]`` into ``dict[str, PydanticJson]``. |
| 161 | |
| 162 | The internal→Pydantic boundary conversion for ``arguments``-style dicts. |
| 163 | Use when passing internal pipeline data into Pydantic BaseModel fields. |
| 164 | """ |
| 165 | return {k: wrap(v) for k, v in d.items()} |