How to use parse_message_attributes method in localstack

Best Python code snippet using localstack_python

utils.py

Source:utils.py

...6from localstack.utils.urls import path_from_url7def is_sqs_queue_url(url):8    path = path_from_url(url).partition("?")[0]9    return re.match(r"^/(queue|%s)/[a-zA-Z0-9_-]+(.fifo)?$" % get_aws_account_id(), path)10def parse_message_attributes(11    querystring, key="MessageAttribute", base="", value_namespace="Value."12):13    message_attributes = {}14    index = 115    while True:16        # Loop through looking for message attributes17        name_key = base + "{0}.{1}.Name".format(key, index)18        name = querystring.get(name_key)19        if not name:20            # Found all attributes21            break22        data_type_key = base + "{0}.{1}.{2}DataType".format(key, index, value_namespace)23        data_type = querystring.get(data_type_key)24        if not data_type:25            raise MessageAttributesInvalid(26                "The message attribute '{0}' must contain non-empty message attribute value.".format(27                    name[0]28                )29            )30        data_type_parts = data_type[0].split(".")31        if data_type_parts[0] not in [32            "String",33            "Binary",34            "Number",35        ]:36            raise MessageAttributesInvalid(37                "The message attribute '{0}' has an invalid message attribute type, the set of supported type prefixes is Binary, Number, and String.".format(38                    name[0]39                )40            )41        type_prefix = "String"42        if data_type_parts[0] == "Binary":43            type_prefix = "Binary"44        value_key = base + "{0}.{1}.{2}{3}Value".format(key, index, value_namespace, type_prefix)45        value = querystring.get(value_key)46        if not value:47            raise MessageAttributesInvalid(48                "The message attribute '{0}' must contain non-empty message attribute value for message attribute type '{1}'.".format(49                    name[0], data_type[0]50                )51            )52        message_attributes[name[0]] = {53            "data_type": data_type[0],54            type_prefix.lower() + "_value": value[0],55        }56        index += 157    return message_attributes58def get_message_attributes_md5(req_data):59    req_data = clone(req_data)60    orig_types = {}61    for key, entry in dict(req_data).items():62        # Fix an issue in moto where data types like 'Number.java.lang.Integer' are63        # not supported: Keep track of the original data type, and temporarily change64        # it to the short form (e.g., 'Number'), before changing it back again.65        if key.endswith("DataType"):66            parts = entry.split(".")67            if len(parts) > 2:68                short_type_name = parts[0]69                full_type_name = entry70                attr_num = key.split(".")[1]71                attr_name = req_data["MessageAttribute.%s.Name" % attr_num]72                orig_types[attr_name] = full_type_name73                req_data[key] = [short_type_name]74                if full_type_name not in TRANSPORT_TYPE_ENCODINGS:75                    TRANSPORT_TYPE_ENCODINGS[full_type_name] = TRANSPORT_TYPE_ENCODINGS[76                        short_type_name77                    ]78    # moto parse_message_attributes(..) expects params to be passed as dict of lists79    req_data_lists = {k: [v] for k, v in req_data.items()}80    moto_message = Message("dummy_msg_id", "dummy_body")81    moto_message.message_attributes = parse_message_attributes(req_data_lists)82    for key, data_type in orig_types.items():83        moto_message.message_attributes[key]["data_type"] = data_type84    message_attr_hash = moto_message.attribute_md5...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.