AioPikaParser

@staticmethod
async def decode_message(
    msg: StreamMessage[aio_pika.IncomingMessage],
) -> DecodedMessage:
    """Decode a message.

    Args:
        msg: The message to decode.

    Returns:
        The decoded message.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    return decode_message(msg)

@staticmethod
def encode_message(
    message: AioPikaSendableMessage,
    persist: bool = False,
    callback_queue: Optional[aio_pika.abc.AbstractRobustQueue] = None,
    reply_to: Optional[str] = None,
    **message_kwargs: Any,
) -> aio_pika.Message:
    """Encodes a message for sending using AioPika.

    Args:
        message (AioPikaSendableMessage): The message to encode.
        persist (bool, optional): Whether to persist the message. Defaults to False.
        callback_queue (aio_pika.abc.AbstractRobustQueue, optional): The callback queue to use for replies. Defaults to None.
        reply_to (str, optional): The reply-to queue to use for replies. Defaults to None.
        **message_kwargs (Any): Additional keyword arguments to include in the encoded message.

    Returns:
        aio_pika.Message: The encoded message.

    Raises:
        NotImplementedError: If the message is not an instance of aio_pika.Message.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    if not isinstance(message, aio_pika.Message):
        message, content_type = encode_message(message)

        delivery_mode = (
            DeliveryMode.PERSISTENT if persist else DeliveryMode.NOT_PERSISTENT
        )

        message = aio_pika.Message(
            message,
            **{
                "delivery_mode": delivery_mode,
                "content_type": content_type,
                "reply_to": callback_queue or reply_to,
                "correlation_id": str(uuid4()),
                **message_kwargs,
            },
        )

    return message

@staticmethod
async def parse_message(
    message: aio_pika.IncomingMessage,
) -> StreamMessage[aio_pika.IncomingMessage]:
    """Parses an incoming message and returns a RabbitMessage object.

    Args:
        message: The incoming message to parse.

    Returns:
        A StreamMessage object representing the parsed message.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    handler = context.get_local("handler_")
    path: AnyDict = {}
    path_re: Optional[Pattern[str]]
    if (  # pragma: no branch
        handler
        and (path_re := handler.queue.path_regex) is not None
        and (match := path_re.match(message.routing_key or "")) is not None
    ):
        path = match.groupdict()

    return RabbitMessage(
        body=message.body,
        headers=message.headers,
        reply_to=message.reply_to or "",
        content_type=message.content_type,
        message_id=message.message_id or str(uuid4()),
        correlation_id=message.correlation_id or str(uuid4()),
        path=path,
        raw_message=message,
    )