turbo_broccoli.custom.embedded

  1"""
  2Embedded JSON documents.
  3
  4Serializing a `EmbeddedDict` or a `EmbeddedList` will (unconditionally) result
  5in its own JSON artefact being created and referenced by the main JSON
  6document.
  7"""
  8
  9# pylint: disable=cyclic-import
 10# pylint: disable=import-outside-toplevel  # to avoid actual circular imports
 11# pylint: disable=protected-access
 12
 13
 14from pathlib import Path
 15from typing import Any, Callable, Tuple
 16
 17from turbo_broccoli.context import Context
 18from turbo_broccoli.exceptions import DeserializationError, TypeNotSupported
 19
 20
 21class EmbeddedDict(dict):
 22    """See module documentation"""
 23
 24    _tb_artifact_id: str | None = None
 25
 26
 27class EmbeddedList(list):
 28    """See module documentation"""
 29
 30    _tb_artifact_id: str | None = None
 31
 32
 33def _get_artifact_path(
 34    obj: EmbeddedDict | EmbeddedList, ctx: Context
 35) -> tuple[Path, str]:
 36    """
 37    If the object has an `_tb_artifact_id`, return the corresponding path and
 38    id. Otherwise, generate new ones.
 39
 40    Warning:
 41        Does not NOT set the `_tb_artifact_id` attribute.
 42    """
 43    if obj._tb_artifact_id is None:
 44        return ctx.new_artifact_path(extension="json")
 45    name = obj._tb_artifact_id
 46    path = ctx.id_to_artifact_path(name, extension="json")
 47    return path, name
 48
 49
 50def _embedded_dict_to_json(obj: EmbeddedDict, ctx: Context) -> dict:
 51    from turbo_broccoli.turbo_broccoli import to_json as _to_json
 52
 53    path, name = _get_artifact_path(obj, ctx)
 54    with path.open("w", encoding="utf-8") as fp:
 55        fp.write(_to_json(dict(obj), ctx))
 56    obj._tb_artifact_id = name
 57    return {"__type__": "embedded.dict", "__version__": 1, "id": name}
 58
 59
 60# TODO: deduplicate with _embedded_dict_to_json
 61def _embedded_list_to_json(obj: EmbeddedList, ctx: Context) -> dict:
 62    from turbo_broccoli.turbo_broccoli import to_json as _to_json
 63
 64    path, name = _get_artifact_path(obj, ctx)
 65    with path.open("w", encoding="utf-8") as fp:
 66        fp.write(_to_json(list(obj), ctx))
 67    obj._tb_artifact_id = name
 68    return {"__type__": "embedded.list", "__version__": 1, "id": name}
 69
 70
 71def _json_to_embedded_dict(dct: dict, ctx: Context) -> EmbeddedDict:
 72    decoders = {
 73        1: _json_to_embedded_dict_v1,
 74    }
 75    return decoders[dct["__version__"]](dct, ctx)
 76
 77
 78def _json_to_embedded_dict_v1(dct: dict, ctx: Context) -> EmbeddedDict:
 79    from turbo_broccoli.turbo_broccoli import from_json as _from_json
 80
 81    path = ctx.id_to_artifact_path(dct["id"], extension="json")
 82    with path.open("r", encoding="utf-8") as fp:
 83        obj = EmbeddedDict(_from_json(fp.read(), ctx))
 84    obj._tb_artifact_id = dct["id"]
 85    return obj
 86
 87
 88def _json_to_embedded_list(dct: dict, ctx: Context) -> EmbeddedList:
 89    decoders = {
 90        1: _json_to_embedded_list_v1,
 91    }
 92    return decoders[dct["__version__"]](dct, ctx)
 93
 94
 95# TODO: deduplicate with _json_to_embedded_dict_v1
 96def _json_to_embedded_list_v1(dct: dict, ctx: Context) -> EmbeddedList:
 97    from turbo_broccoli.turbo_broccoli import from_json as _from_json
 98
 99    path = ctx.id_to_artifact_path(dct["id"], extension="json")
100    with path.open("r", encoding="utf-8") as fp:
101        obj = EmbeddedList(_from_json(fp.read(), ctx))
102    obj._tb_artifact_id = dct["id"]
103    return obj
104
105
106# pylint: disable=missing-function-docstring
107def from_json(dct: dict, ctx: Context) -> Any:
108    decoders = {
109        "embedded.dict": _json_to_embedded_dict,
110        "embedded.list": _json_to_embedded_list,
111    }
112    try:
113        type_name = dct["__type__"]
114        return decoders[type_name](dct, ctx)
115    except KeyError as exc:
116        raise DeserializationError() from exc
117
118
119def to_json(obj: Any, ctx: Context) -> dict:
120    """
121    Serializes a `EmbeddedDict` or an `EmbeddedList` into JSON. The return dict
122    has the following structure
123
124    ```py
125    {
126        "__type__": "embedded.dict",
127        "__version__": 1,
128        "id": <uuid4>,
129    }
130    ```
131
132    or
133
134    ```py
135    {
136        "__type__": "embedded.list",
137        "__version__": 1,
138        "id": <uuid4>,
139    }
140    ```
141
142    where the UUID points to the artefact containing the actual data.
143    """
144    encoders: list[Tuple[type, Callable[[Any, Context], dict]]] = [
145        (EmbeddedDict, _embedded_dict_to_json),
146        (EmbeddedList, _embedded_list_to_json),
147    ]
148    for t, f in encoders:
149        if isinstance(obj, t):
150            return f(obj, ctx)
151    raise TypeNotSupported()