Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _serialize_body_params_to_xml method in localstack

Best Python code snippet using localstack_python

serializer.py

Source:serializer.py

...326            fault_tag.text = "Sender"327    def _serialize_body_params(328        self, params: dict, shape: Shape, operation_model: OperationModel329    ) -> str:330        root = self._serialize_body_params_to_xml(params, shape, operation_model)331        self._prepare_additional_traits_in_xml(root)332        return ETree.tostring(root, encoding=self.DEFAULT_ENCODING)333    def _serialize_body_params_to_xml(334        self, params: dict, shape: Shape, operation_model: OperationModel335    ) -> Optional[ETree.Element]:336        if shape is None:337            return338        # The botocore serializer expects `shape.serialization["name"]`, but this isn't always present for responses339        root_name = shape.serialization.get("name", shape.name)340        pseudo_root = ETree.Element("")341        self._serialize(shape, params, pseudo_root, root_name)342        real_root = list(pseudo_root)[0]343        return real_root344    def _encode_payload(self, body: Union[bytes, str]) -> bytes:345        if isinstance(body, six.text_type):346            return body.encode(self.DEFAULT_ENCODING)347        return body348    def _serialize(self, shape: Shape, params: any, xmlnode: ETree.Element, name: str) -> None:349        """This method dynamically invokes the correct `_serialize_type_*` method for each shape type."""350        if shape is None:351            return352        # Some output shapes define a `resultWrapper` in their serialization spec.353        # While the name would imply that the result is _wrapped_, it is actually renamed.354        if shape.serialization.get("resultWrapper"):355            name = shape.serialization.get("resultWrapper")356        method = getattr(self, "_serialize_type_%s" % shape.type_name, self._default_serialize)357        method(xmlnode, params, shape, name)358    def _serialize_type_structure(359        self, xmlnode: ETree.Element, params: dict, shape: StructureShape, name: str360    ) -> None:361        structure_node = ETree.SubElement(xmlnode, name)362        if "xmlNamespace" in shape.serialization:363            namespace_metadata = shape.serialization["xmlNamespace"]364            attribute_name = "xmlns"365            if namespace_metadata.get("prefix"):366                attribute_name += ":%s" % namespace_metadata["prefix"]367            structure_node.attrib[attribute_name] = namespace_metadata["uri"]368        for key, value in params.items():369            member_shape = shape.members[key]370            member_name = member_shape.serialization.get("name", key)371            # We need to special case member shapes that are marked as an xmlAttribute.372            # Rather than serializing into an XML child node, we instead serialize the shape to373            # an XML attribute of the *current* node.374            if value is None:375                # Don't serialize any param whose value is None.376                continue377            if member_shape.serialization.get("xmlAttribute"):378                # xmlAttributes must have a serialization name.379                xml_attribute_name = member_shape.serialization["name"]380                structure_node.attrib[xml_attribute_name] = value381                continue382            self._serialize(member_shape, value, structure_node, member_name)383    def _serialize_type_list(384        self, xmlnode: ETree.Element, params: list, shape: ListShape, name: str385    ) -> None:386        member_shape = shape.member387        if shape.serialization.get("flattened"):388            # If the list is flattened, either take the member's "name" or the name of the usual name for the parent389            # element for the children.390            element_name = self._get_serialized_name(member_shape, name)391            list_node = xmlnode392        else:393            element_name = self._get_serialized_name(member_shape, "member")394            list_node = ETree.SubElement(xmlnode, name)395        for item in params:396            self._serialize(member_shape, item, list_node, element_name)397    def _serialize_type_map(398        self, xmlnode: ETree.Element, params: dict, shape: MapShape, name: str399    ) -> None:400        """401        Given the ``name`` of MyMap, an input of {"key1": "val1", "key2": "val2"}, and the ``flattened: False``402        we serialize this as:403          <MyMap>404            <entry>405              <key>key1</key>406              <value>val1</value>407            </entry>408            <entry>409              <key>key2</key>410              <value>val2</value>411            </entry>412          </MyMap>413        If it is flattened, it is serialized as follows:414          <MyMap>415            <key>key1</key>416            <value>val1</value>417          </MyMap>418          <MyMap>419            <key>key2</key>420            <value>val2</value>421          </MyMap>422        """423        if shape.serialization.get("flattened"):424            entries_node = xmlnode425            entry_node_name = name426        else:427            entries_node = ETree.SubElement(xmlnode, name)428            entry_node_name = "entry"429        for key, value in params.items():430            entry_node = ETree.SubElement(entries_node, entry_node_name)431            key_name = self._get_serialized_name(shape.key, default_name="key")432            val_name = self._get_serialized_name(shape.value, default_name="value")433            self._serialize(shape.key, key, entry_node, key_name)434            self._serialize(shape.value, value, entry_node, val_name)435    @staticmethod436    def _serialize_type_boolean(xmlnode: ETree.Element, params: bool, _, name: str) -> None:437        """438        For scalar types, the 'params' attr is actually just a scalar value representing the data439        we need to serialize as a boolean. It will either be 'true' or 'false'440        """441        node = ETree.SubElement(xmlnode, name)442        if params:443            str_value = "true"444        else:445            str_value = "false"446        node.text = str_value447    def _serialize_type_blob(448        self, xmlnode: ETree.Element, params: Union[str, bytes], _, name: str449    ) -> None:450        node = ETree.SubElement(xmlnode, name)451        node.text = self._get_base64(params)452    def _serialize_type_timestamp(453        self, xmlnode: ETree.Element, params: str, shape: Shape, name: str454    ) -> None:455        node = ETree.SubElement(xmlnode, name)456        node.text = self._convert_timestamp_to_str(457            params, shape.serialization.get("timestampFormat")458        )459    @staticmethod460    def _default_serialize(xmlnode: ETree.Element, params: str, _, name: str) -> None:461        node = ETree.SubElement(xmlnode, name)462        node.text = six.text_type(params)463    def _prepare_additional_traits_in_xml(self, root: Optional[ETree.Element]):464        """465        Prepares the XML root node before being serialized with additional traits (like the Response ID in the Query466        protocol).467        For some protocols (like rest-xml), the root can be None.468        """469        pass470class BaseRestResponseSerializer(ResponseSerializer, ABC):471    """472    The BaseRestResponseSerializer performs the basic logic for the ReST response serialization.473    In our case it basically only adds the request metadata to the HTTP header.474    """475    def _prepare_additional_traits_in_response(476        self, response: HttpResponse, operation_model: OperationModel477    ):478        """Adds the request ID to the headers (in contrast to the body - as in the Query protocol)."""479        response = super()._prepare_additional_traits_in_response(response, operation_model)480        response["headers"]["x-amz-request-id"] = gen_amzn_requestid_long()481        return response482class RestXMLResponseSerializer(BaseRestResponseSerializer, BaseXMLResponseSerializer):483    """484    The ``RestXMLResponseSerializer`` is responsible for the serialization of responses from services with the485    ``rest-xml`` protocol.486    It combines the ``BaseRestResponseSerializer`` (for the ReST specific logic) with the ``BaseXMLResponseSerializer``487    (for the XML body response serialization), and adds some minor logic to handle S3 specific peculiarities with the488    error response serialization.489    **Experimental:** This serializer is still experimental.490    When implementing services with this serializer, some edge cases might not work out-of-the-box.491    """492    def _serialize_error(493        self,494        error: ServiceException,495        code: str,496        sender_fault: bool,497        serialized: HttpResponse,498        shape: Shape,499        operation_model: OperationModel,500    ) -> None:501        # It wouldn't be a spec if there wouldn't be any exceptions.502        # S3 errors look differently than other service's errors.503        if operation_model.name == "s3":504            attr = (505                {"xmlns": operation_model.metadata.get("xmlNamespace")}506                if "xmlNamespace" in operation_model.metadata507                else None508            )509            root = ETree.Element("Error", attr)510            self._add_error_tags(code, error, root, sender_fault)511            request_id = ETree.SubElement(root, "RequestId")512            request_id.text = gen_amzn_requestid_long()513            serialized["body"] = self._encode_payload(514                ETree.tostring(root, encoding=self.DEFAULT_ENCODING)515            )516        else:517            super()._serialize_error(error, code, sender_fault, serialized, shape, operation_model)518class QueryResponseSerializer(BaseXMLResponseSerializer):519    """520    The ``QueryResponseSerializer`` is responsible for the serialization of responses from services which use the521    ``query`` protocol. The responses of these services also use XML, but with a few subtle differences to the522    ``rest-xml`` protocol.523    **Experimental:** This serializer is still experimental.524    When implementing services with this serializer, some edge cases might not work out-of-the-box.525    """526    def _serialize_body_params_to_xml(527        self, params: dict, shape: Shape, operation_model: OperationModel528    ) -> ETree.Element:529        # The Query protocol responses have a root element which is not contained in the specification file.530        # Therefore we first call the super function to perform the normal XML serialization, and afterwards wrap the531        # result in a root element based on the operation name.532        node = super()._serialize_body_params_to_xml(params, shape, operation_model)533        # Check if we need to add a namespace534        attr = (535            {"xmlns": operation_model.metadata.get("xmlNamespace")}536            if "xmlNamespace" in operation_model.metadata537            else None538        )539        # Create the root element and add the result of the XML serializer as a child node540        root = ETree.Element(f"{operation_model.name}Response", attr)541        if node is not None:542            root.append(node)543        return root544    def _prepare_additional_traits_in_xml(self, root: Optional[ETree.Element]):545        # Add the response metadata here (it's not defined in the specs)546        # For the ec2 and the query protocol, the root cannot be None at this time....

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.