"""
method to create, serialize and deserialize DataResponse message
"""
import base64
import re
from io import BytesIO
from amqp.adapter.serializer import map_as_string, parse_map_string
from amqp.model.model import AMQErrorMessage, DataMessageMagicByte, DataResponse
from amqp.model.snowflake_id import SnowflakeId
[docs]
class DataResponseFactory:
[docs]
@staticmethod
def create_async_response_message(_id: SnowflakeId) -> DataResponse:
"""
Helper method to create standard OK response.
:param _id: ID
:return: Response message
"""
return DataResponseFactory.of(
id=_id,
response_code=200,
content_type="application/text",
body="OK".encode("utf-8"),
trace_info={},
error="",
error_cause="",
)
[docs]
@staticmethod
def of(
id: SnowflakeId,
response_code: int,
content_type: str,
body: bytes,
trace_info: dict[str, str],
error: str | None = None,
error_cause: str | None = None,
) -> DataResponse:
"""
Create the DataResponse message supplying all key values
:param id: SnowflakeID
:param response_code: HTTP response code
:param content_type: MIME content type
:param body: payload
:param trace_info: Jaeger trace info (id)
:return: DataResponseMessage object
"""
return DataResponse(
magic=DataMessageMagicByte.HTTP_REQUEST.value,
id=id,
response_code=str(response_code),
content_type=content_type,
error=error,
error_cause=error_cause,
headers={},
trace_info=trace_info,
base64body=base64.b64encode(body),
)
[docs]
@staticmethod
def of_error_message(
id: SnowflakeId,
error_message: AMQErrorMessage,
trace_info: dict[str, str],
) -> DataResponse:
return DataResponseFactory.of(
id=id,
response_code=error_message.response_code,
content_type=error_message.content_type,
body=error_message.serialize(),
trace_info=trace_info,
error=error_message.error,
error_cause=error_message.detail,
)
[docs]
@staticmethod
def serialize(response: DataResponse) -> bytes:
"""
Serialize the message to byte array (to be sent over RabbitMQ)
:param response: Object to serialize
:return: byte array as serialized message
"""
id_bytes = response.id().as_string().encode("utf-8")
prolog = (
f"|r={response.response_code()}|c={response.content_type()}|e={response.error()}|f={response.error_cause()}|t={{{map_as_string(response.trace_info())}}}|b="
).encode("utf-8")
header_length = 1 + len(id_bytes) + len(prolog)
prolog_len = header_length.to_bytes(2, byteorder="big")
msg_length = 2 + header_length + len(response.base64body())
with BytesIO(bytearray(msg_length)) as baos:
baos.write(prolog_len)
baos.write(DataMessageMagicByte.HTTP_REQUEST.value.encode("utf-8"))
baos.write(id_bytes)
baos.write(prolog)
baos.write(response.base64body())
return baos.getvalue()
[docs]
@staticmethod
def from_bytes(input_bytes: bytes) -> DataResponse:
"""
Buil;ds the DataResponse object from its serialized form
:param input_bytes: byte array - serialized response
:return: DataResponse object
"""
prolog_len = (input_bytes[0] << 8) + input_bytes[1]
metadata = input_bytes[2 : prolog_len + 2].decode()
pattern = re.compile(
f"{DataMessageMagicByte.HTTP_REQUEST.value}"
r"([0-9A-Fa-f]+)\|"
r"r=([^|]*)\|"
r"c=([^|]*)\|"
r"e=([^|]*)\|"
r"f=([^|]*)\|"
r"t=(\{[^}]*\})\|"
r"b="
)
match = pattern.match(metadata)
if not match:
raise ValueError(
f"DataResponse format does not match expected format. "
f"Input string [{metadata}] does not match pattern: {pattern.pattern}"
)
id_str = match.group(1)
response_code = int(match.group(2))
content_type = match.group(3)
error = match.group(4)
error_cause = match.group(5)
trace_info_str = match.group(6)
body_b64 = input_bytes[prolog_len + 2 :]
trace_info = parse_map_string(trace_info_str)
return DataResponseFactory.of(
id_str,
response_code,
content_type,
base64.b64decode(body_b64),
trace_info,
error,
error_cause,
)
# Ensure backward compatibility
AMQResponseFactory = DataResponseFactory