def get_decoded_message()

in azurefunctions-extensions-bindings-servicebus/azurefunctions/extensions/bindings/servicebus/utils.py [0:0]

• 12 lines of code
• 3 McCabe index (conditional complexity)

def get_decoded_message(content: bytes):
    """
    First, find the end of the lock token. Then,
    get the lock token UUID and create the delivery
    annotations dictionary. Finally, get the amqp message
    and set the delivery annotations. Once the delivery
    annotations have been set, the amqp message is ready to
    return.
    """
    if content:
        try:
            index = content.find(_X_OPT_LOCK_TOKEN)

            lock_token = get_lock_token(content, index)
            delivery_anno_dict = {_X_OPT_LOCK_TOKEN: lock_token}

            decoded_message = get_amqp_message(content, index)
            decoded_message.delivery_annotations = delivery_anno_dict
            return decoded_message
        except Exception as e:
            raise ValueError(f"Failed to decode ServiceBus content: {e}") from e
    return None