turbo_broccoli.custom.embedded
Embedded JSON documents.
Serializing a EmbeddedDict
or a EmbeddedList
will (unconditionally) result
in its own JSON artefact being created and referenced by the main JSON
document.
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()
class
EmbeddedDict(builtins.dict):
22class EmbeddedDict(dict): 23 """See module documentation""" 24 25 _tb_artifact_id: str | None = None
See module documentation
Inherited Members
- builtins.dict
- get
- setdefault
- pop
- popitem
- keys
- items
- values
- update
- fromkeys
- clear
- copy
class
EmbeddedList(builtins.list):
28class EmbeddedList(list): 29 """See module documentation""" 30 31 _tb_artifact_id: str | None = None
See module documentation
Inherited Members
- builtins.list
- list
- clear
- copy
- append
- insert
- extend
- pop
- remove
- index
- count
- reverse
- sort
108def from_json(dct: dict, ctx: Context) -> Any: 109 decoders = { 110 "embedded.dict": _json_to_embedded_dict, 111 "embedded.list": _json_to_embedded_list, 112 } 113 try: 114 type_name = dct["__type__"] 115 return decoders[type_name](dct, ctx) 116 except KeyError as exc: 117 raise DeserializationError() from exc
120def to_json(obj: Any, ctx: Context) -> dict: 121 """ 122 Serializes a `EmbeddedDict` or an `EmbeddedList` into JSON. The return dict 123 has the following structure 124 125 ```py 126 { 127 "__type__": "embedded.dict", 128 "__version__": 1, 129 "id": <uuid4>, 130 } 131 ``` 132 133 or 134 135 ```py 136 { 137 "__type__": "embedded.list", 138 "__version__": 1, 139 "id": <uuid4>, 140 } 141 ``` 142 143 where the UUID points to the artefact containing the actual data. 144 """ 145 encoders: list[Tuple[type, Callable[[Any, Context], dict]]] = [ 146 (EmbeddedDict, _embedded_dict_to_json), 147 (EmbeddedList, _embedded_list_to_json), 148 ] 149 for t, f in encoders: 150 if isinstance(obj, t): 151 return f(obj, ctx) 152 raise TypeNotSupported()
Serializes a EmbeddedDict
or an EmbeddedList
into JSON. The return dict
has the following structure
{
"__type__": "embedded.dict",
"__version__": 1,
"id": <uuid4>,
}
or
{
"__type__": "embedded.list",
"__version__": 1,
"id": <uuid4>,
}
where the UUID points to the artefact containing the actual data.