1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
|
"""Generic Utility helper functions."""
from __future__ import annotations
import base64
import json
from typing import Any
from ..exceptions import UnparseableValue
def is_json_string(value: Any) -> bool:
"""Check if the provided string looks like json."""
# NOTE: we do not use json.loads here as it is not strict enough
return isinstance(value, str) and value.startswith("{") and value.endswith("}")
def convert_bytes_to_base64(data: bytes) -> str:
"""Convert bytes data to base64 for serialization."""
return base64.b64encode(data).decode("ascii")
def convert_base64_to_bytes(data: str) -> bytes:
"""Convert base64 data to bytes for deserialization."""
return base64.b64decode(data)
def parse_buffer(value: dict[str, Any] | str) -> str:
"""Parse value from a buffer data type."""
if isinstance(value, dict):
return parse_buffer_from_dict(value)
if is_json_string(value):
return parse_buffer_from_json(value)
return value
def parse_buffer_from_dict(value: dict[str, Any]) -> str:
"""Parse value dictionary from a buffer data type."""
if value.get("type") != "Buffer" or "data" not in value:
raise UnparseableValue(f"Unparseable value: {value}") from ValueError(
"JSON does not match expected schema"
)
return "".join([chr(x) for x in value["data"]])
def parse_buffer_from_json(value: str) -> str:
"""Parse value string from a buffer data type."""
try:
return parse_buffer_from_dict(json.loads(value))
except ValueError as err:
raise UnparseableValue(f"Unparseable value: {value}") from err
|
