Stanislav Hejny
77 lines
2.1 KiB
Python
77 lines
2.1 KiB
Python
"""
|
|
Helper functions used to aid with (de)serialization of messages sent over AMQP.
|
|
"""
|
|
|
|
import struct
|
|
from io import BytesIO
|
|
from typing import Dict, List, Union
|
|
|
|
|
|
def long_to_bytes(x: int) -> bytes:
|
|
return struct.pack("q", x)
|
|
|
|
|
|
def bytes_to_long(b: bytes) -> int:
|
|
return struct.unpack("q", b)[0]
|
|
|
|
|
|
def read_long(bis: BytesIO) -> int:
|
|
try:
|
|
return struct.unpack("q", bis.read(8))[0]
|
|
except struct.error:
|
|
return 0
|
|
|
|
|
|
def object_or_list_as_string(value: Union[str, List[str]]) -> str:
|
|
if isinstance(value, list):
|
|
return f"[{','.join(value)}]"
|
|
return str(value)
|
|
|
|
|
|
def map_as_string(map_: Dict[str, str]) -> str:
|
|
return ",".join(f"{k}={object_or_list_as_string(v)}" for k, v in map_.items())
|
|
|
|
|
|
def map_with_list_as_string(map_: Dict[str, List[str]]) -> str:
|
|
return ",".join(f"{k}={object_or_list_as_string(v)}" for k, v in map_.items())
|
|
|
|
|
|
def parse_map_string(input_str: str) -> Dict[str, str]:
|
|
map_ = {}
|
|
if input_str.startswith("{") and input_str.endswith("}"):
|
|
for entry in str(input_str[1:-1]).replace("\n", "").split(","):
|
|
if len(entry) > 1:
|
|
if entry.startswith('"'):
|
|
key, value = entry.split(":")
|
|
map_[key.strip('"')] = value.strip('"')
|
|
else:
|
|
key, value = entry.split("=")
|
|
map_[key] = value
|
|
return map_
|
|
|
|
|
|
def csv_as_list(input_str: str) -> List[str]:
|
|
return [item.strip() for item in input_str.strip("[]").split(",")]
|
|
|
|
|
|
def add_to_map(map_: Dict[str, List[str]], token: str) -> None:
|
|
try:
|
|
eq_idx = token.index("=")
|
|
except ValueError:
|
|
eq_idx = -1 # set to -1 when not found
|
|
if eq_idx > 0:
|
|
key = token[:eq_idx]
|
|
map_[key] = csv_as_list(token[eq_idx + 1 :])
|
|
|
|
|
|
def parse_map_list_string(input_str: str) -> Dict[str, List[str]]:
|
|
map_ = {}
|
|
if input_str.startswith("{") and input_str.endswith("}"):
|
|
for token in input_str[1:-1].split("],"):
|
|
add_to_map(map_, token)
|
|
return map_
|
|
|
|
|
|
def str_to_bool(value: str) -> bool:
|
|
return value.lower() in ("true", "yes", "1", "t", "y", "on")
|