KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _prepare_additional_traits_in_response method in localstack

Best Python code snippet using localstack_python
serializer.py
Source:serializer.py
...105        shape_members = shape.members if shape is not None else None106        self._serialize_payload(107            response, serialized_response, shape, shape_members, operation_model108        )109        serialized_response = self._prepare_additional_traits_in_response(110            serialized_response, operation_model111        )112        return serialized_response113    def serialize_error_to_response(114        self, error: ServiceException, operation_model: OperationModel115    ) -> HttpResponse:116        """117        Takes an error instance and serializes it to an actual HttpResponse.118        Therefore this method is used for errors which should be serialized and transmitted to the calling client.119        :param error: to serialize120        :param operation_model: specification of the service & operation containing information about the shape of the121                                service's output / response122        :return: HttpResponse which can be sent to the calling client123        """124        serialized_response = self._create_default_response(operation_model)125        if isinstance(error, CommonServiceException):126            # Not all possible exceptions are contained in the service's specification.127            # Therefore, service implementations can also throw a "CommonServiceException" to raise arbitrary /128            # non-specified exceptions (where the developer needs to define the data which would usually be taken from129            # the specification, like the "Code").130            code = error.code131            sender_fault = error.sender_fault132            status_code = error.status_code133            shape = None134        else:135            # It it's not a CommonServiceException, the exception is being serialized based on the specification136            # The shape name is equal to the class name (since the classes are generated from the shape's name)137            error_shape_name = error.__class__.__name__138            # Lookup the corresponding error shape in the operation model139            shape = next(140                shape for shape in operation_model.error_shapes if shape.name == error_shape_name141            )142            error_spec = shape.metadata.get("error", {})143            status_code = error_spec.get("httpStatusCode")144            # If the code is not explicitly set, it's typically the shape's name145            code = error_spec.get("code", shape.name)146            # The senderFault is only set if the "senderFault" is true147            # (there are no examples which show otherwise)148            sender_fault = error_spec.get("senderFault")149        # Some specifications do not contain the httpStatusCode field.150        # These errors typically have the http status code 400.151        serialized_response["status_code"] = status_code or 400152        self._serialize_error(153            error, code, sender_fault, serialized_response, shape, operation_model154        )155        serialized_response = self._prepare_additional_traits_in_response(156            serialized_response, operation_model157        )158        return serialized_response159    def _serialize_payload(160        self,161        parameters: dict,162        serialized: HttpResponse,163        shape: Optional[Shape],164        shape_members: dict,165        operation_model: OperationModel,166    ) -> None:167        # TODO implement the handling of location traits (where "location" is "header", "headers", or "path")168        # TODO implement the handling of eventstreams (where "streaming" is True)169        raise NotImplementedError170    def _serialize_error(171        self,172        error: ServiceException,173        code: str,174        sender_fault: bool,175        serialized: HttpResponse,176        shape: Shape,177        operation_model: OperationModel,178    ) -> None:179        raise NotImplementedError180    @staticmethod181    def _create_default_response(operation_model: OperationModel) -> HttpResponse:182        """183        Creates a boilerplate default response dict to be used by subclasses as starting points.184        Uses the default HTTP response status code defined in the operation model (if defined).185        :param operation_model: to extract the default HTTP status code186        :return: boilerplate HTTP response187        """188        return HttpResponse(189            headers={}, body=b"", status_code=operation_model.http.get("responseCode", 200)190        )191    # Some extra utility methods subclasses can use.192    @staticmethod193    def _timestamp_iso8601(value: datetime) -> str:194        if value.microsecond > 0:195            timestamp_format = ISO8601_MICRO196        else:197            timestamp_format = ISO8601198        return value.strftime(timestamp_format)199    @staticmethod200    def _timestamp_unixtimestamp(value: datetime) -> int:201        return int(calendar.timegm(value.timetuple()))202    def _timestamp_rfc822(self, value: datetime) -> str:203        if isinstance(value, datetime):204            value = self._timestamp_unixtimestamp(value)205        return formatdate(value, usegmt=True)206    def _convert_timestamp_to_str(207        self, value: Union[int, str, datetime], timestamp_format=None208    ) -> str:209        if timestamp_format is None:210            timestamp_format = self.TIMESTAMP_FORMAT211        timestamp_format = timestamp_format.lower()212        datetime_obj = parse_to_aware_datetime(value)213        converter = getattr(self, "_timestamp_%s" % timestamp_format)214        final_value = converter(datetime_obj)215        return final_value216    @staticmethod217    def _get_serialized_name(shape: Shape, default_name: str) -> str:218        """219        Returns the serialized name for the shape if it exists.220        Otherwise it will return the passed in default_name.221        """222        return shape.serialization.get("name", default_name)223    def _get_base64(self, value: Union[str, bytes]):224        """225        Returns the base64-encoded version of value, handling226        both strings and bytes. The returned value is a string227        via the default encoding.228        """229        if isinstance(value, six.text_type):230            value = value.encode(self.DEFAULT_ENCODING)231        return base64.b64encode(value).strip().decode(self.DEFAULT_ENCODING)232    def _prepare_additional_traits_in_response(233        self, response: HttpResponse, operation_model: OperationModel234    ):235        """Applies additional traits on the raw response for a given model or protocol."""236        if operation_model.http_checksum_required:237            conditionally_calculate_md5(response)238        return response239class BaseXMLResponseSerializer(ResponseSerializer):240    """241    The BaseXMLResponseSerializer performs the basic logic for the XML response serialization.242    It is slightly adapted by the QueryResponseSerializer.243    While the botocore's RestXMLSerializer is quite similar, there are some subtle differences (since botocore's244    implementation handles the serialization of the requests from the client to the service, not the responses from the245    service to the client).246    **Experimental:** This serializer is still experimental.247    When implementing services with this serializer, some edge cases might not work out-of-the-box.248    """249    def _serialize_payload(250        self,251        parameters: dict,252        serialized: HttpResponse,253        shape: Optional[Shape],254        shape_members: dict,255        operation_model: OperationModel,256    ) -> None:257        """258        Serializes the given parameters as XML.259        :param parameters: The user input params260        :param serialized: The final serialized response dict261        :param shape: Describes the expected output shape (can be None in case of an "empty" response)262        :param shape_members: The members of the output struct shape263        :param operation_model: The specification of the operation of which the response is serialized here264        :return: None - the given `serialized` dict is modified265        """266        payload_member = shape.serialization.get("payload") if shape is not None else None267        if payload_member is not None and shape_members[payload_member].type_name in [268            "blob",269            "string",270        ]:271            # If it's streaming, then the body is just the value of the payload.272            body_payload = parameters.get(payload_member, b"")273            body_payload = self._encode_payload(body_payload)274            serialized["body"] = body_payload275        elif payload_member is not None:276            # If there's a payload member, we serialized that member to the body.277            body_params = parameters.get(payload_member)278            if body_params is not None:279                serialized["body"] = self._encode_payload(280                    self._serialize_body_params(281                        body_params, shape_members[payload_member], operation_model282                    )283                )284        else:285            # Otherwise we use the "traditional" way of serializing the whole parameters dict recursively.286            serialized["body"] = self._encode_payload(287                self._serialize_body_params(parameters, shape, operation_model)288            )289    def _serialize_error(290        self,291        error: ServiceException,292        code: str,293        sender_fault: bool,294        serialized: HttpResponse,295        shape: Shape,296        operation_model: OperationModel,297    ) -> None:298        # TODO handle error shapes with members299        # Check if we need to add a namespace300        attr = (301            {"xmlns": operation_model.metadata.get("xmlNamespace")}302            if "xmlNamespace" in operation_model.metadata303            else {}304        )305        root = ETree.Element("ErrorResponse", attr)306        error_tag = ETree.SubElement(root, "Error")307        self._add_error_tags(code, error, error_tag, sender_fault)308        request_id = ETree.SubElement(root, "RequestId")309        request_id.text = gen_amzn_requestid_long()310        serialized["body"] = self._encode_payload(311            ETree.tostring(root, encoding=self.DEFAULT_ENCODING)312        )313    @staticmethod314    def _add_error_tags(315        code: str, error: ServiceException, error_tag: ETree.Element, sender_fault: bool316    ) -> None:317        code_tag = ETree.SubElement(error_tag, "Code")318        code_tag.text = code319        message = str(error)320        if len(message) > 0:321            message_tag = ETree.SubElement(error_tag, "Message")322            message_tag.text = message323        if sender_fault:324            # The sender fault is either not set or "Sender"325            fault_tag = ETree.SubElement(error_tag, "Fault")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.547        response_metadata = ETree.SubElement(root, "ResponseMetadata")548        request_id = ETree.SubElement(response_metadata, "RequestId")549        request_id.text = gen_amzn_requestid_long()550class EC2ResponseSerializer(QueryResponseSerializer):551    """552    The ``EC2ResponseSerializer`` is responsible for the serialization of responses from services which use the553    ``ec2`` protocol (basically the EC2 service). This protocol is basically equal to the ``query`` protocol with only554    a few subtle differences.555    **Experimental:** This serializer is still experimental.556    When implementing services with this serializer, some edge cases might not work out-of-the-box.557    """558    def _serialize_error(559        self,560        error: ServiceException,561        code: str,562        sender_fault: bool,563        serialized: HttpResponse,564        shape: Shape,565        operation_model: OperationModel,566    ) -> None:567        # EC2 errors look like:568        # <Response>569        #   <Errors>570        #     <Error>571        #       <Code>InvalidInstanceID.Malformed</Code>572        #       <Message>Invalid id: "1343124"</Message>573        #     </Error>574        #   </Errors>575        #   <RequestID>12345</RequestID>576        # </Response>577        # This is different from QueryParser in that it's RequestID, not RequestId578        # and that the Error tag is in an enclosing Errors tag.579        attr = (580            {"xmlns": operation_model.metadata.get("xmlNamespace")}581            if "xmlNamespace" in operation_model.metadata582            else None583        )584        root = ETree.Element("Errors", attr)585        error_tag = ETree.SubElement(root, "Error")586        self._add_error_tags(code, error, error_tag, sender_fault)587        request_id = ETree.SubElement(root, "RequestID")588        request_id.text = gen_amzn_requestid_long()589        serialized["body"] = self._encode_payload(590            ETree.tostring(root, encoding=self.DEFAULT_ENCODING)591        )592    def _prepare_additional_traits_in_xml(self, root: Optional[ETree.Element]):593        # The EC2 protocol does not use the root output shape, therefore we need to remove the hierarchy level594        # below the root level595        output_node = root[0]596        for child in output_node:597            root.append(child)598        root.remove(output_node)599        # Add the requestId here (it's not defined in the specs)600        # For the ec2 and the query protocol, the root cannot be None at this time.601        request_id = ETree.SubElement(root, "requestId")602        request_id.text = gen_amzn_requestid_long()603class JSONResponseSerializer(ResponseSerializer):604    """605    The ``JSONResponseSerializer`` is responsible for the serialization of responses from services with the ``json``606    protocol. It implements the JSON response body serialization, which is also used by the607    ``RestJSONResponseSerializer``.608    **Experimental:** This serializer is still experimental.609    When implementing services with this serializer, some edge cases might not work out-of-the-box.610    """611    TIMESTAMP_FORMAT = "unixtimestamp"612    def _serialize_error(613        self,614        error: ServiceException,615        code: str,616        sender_fault: bool,617        serialized: HttpResponse,618        shape: Shape,619        operation_model: OperationModel,620    ) -> None:621        # TODO handle error shapes with members622        body = {"__type": code, "message": str(error)}623        serialized["body"] = json.dumps(body).encode(self.DEFAULT_ENCODING)624    def _serialize_payload(625        self,626        parameters: dict,627        serialized: HttpResponse,628        shape: Optional[Shape],629        shape_members: dict,630        operation_model: OperationModel,631    ) -> None:632        json_version = operation_model.metadata.get("jsonVersion")633        if json_version is not None:634            serialized["headers"] = {635                "Content-Type": "application/x-amz-json-%s" % json_version,636            }637        body = {}638        if shape is not None:639            self._serialize(body, parameters, shape)640        serialized["body"] = json.dumps(body).encode(self.DEFAULT_ENCODING)641    def _serialize(self, body: dict, value: any, shape, key: Optional[str] = None):642        """This method dynamically invokes the correct `_serialize_type_*` method for each shape type."""643        method = getattr(self, "_serialize_type_%s" % shape.type_name, self._default_serialize)644        method(body, value, shape, key)645    def _serialize_type_structure(self, body: dict, value: dict, shape: StructureShape, key: str):646        if shape.is_document_type:647            body[key] = value648        else:649            if key is not None:650                # If a key is provided, this is a result of a recursive651                # call so we need to add a new child dict as the value652                # of the passed in serialized dict.  We'll then add653                # all the structure members as key/vals in the new serialized654                # dictionary we just created.655                new_serialized = {}656                body[key] = new_serialized657                body = new_serialized658            members = shape.members659            for member_key, member_value in value.items():660                member_shape = members[member_key]661                if "name" in member_shape.serialization:662                    member_key = member_shape.serialization["name"]663                self._serialize(body, member_value, member_shape, member_key)664    def _serialize_type_map(self, body: dict, value: dict, shape: MapShape, key: str):665        map_obj = {}666        body[key] = map_obj667        for sub_key, sub_value in value.items():668            self._serialize(map_obj, sub_value, shape.value, sub_key)669    def _serialize_type_list(self, body: dict, value: list, shape: ListShape, key: str):670        list_obj = []671        body[key] = list_obj672        for list_item in value:673            wrapper = {}674            # The JSON list serialization is the only case where we aren't675            # setting a key on a dict.  We handle this by using676            # a __current__ key on a wrapper dict to serialize each677            # list item before appending it to the serialized list.678            self._serialize(wrapper, list_item, shape.member, "__current__")679            list_obj.append(wrapper["__current__"])680    @staticmethod681    def _default_serialize(body: dict, value: any, _, key: str):682        body[key] = value683    def _serialize_type_timestamp(self, body: dict, value: any, shape: Shape, key: str):684        body[key] = self._convert_timestamp_to_str(685            value, shape.serialization.get("timestampFormat")686        )687    def _serialize_type_blob(self, body: dict, value: Union[str, bytes], _, key: str):688        body[key] = self._get_base64(value)689    def _prepare_additional_traits_in_response(690        self, response: HttpResponse, operation_model: OperationModel691    ):692        response["headers"]["x-amzn-requestid"] = gen_amzn_requestid_long()693        response = super()._prepare_additional_traits_in_response(response, operation_model)694        return response695class RestJSONResponseSerializer(BaseRestResponseSerializer, JSONResponseSerializer):696    """697    The ``RestJSONResponseSerializer`` is responsible for the serialization of responses from services with the698    ``rest-json`` protocol.699    It combines the ``BaseRestResponseSerializer`` (for the ReST specific logic) with the ``JSONResponseSerializer``700    (for the JSOn body response serialization).701    **Experimental:** This serializer is still experimental.702    When implementing services with this serializer, some edge cases might not work out-of-the-box.703    """704    pass705def create_serializer(service: ServiceModel) -> ResponseSerializer:706    """707    Creates the right serializer for the given service model....
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.