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

How to use _create_default_response method in localstack

Best Python code snippet using localstack_python

serializer.py

Source:serializer.py

...99        :param operation_model: specification of the service & operation containing information about the shape of the100                                service's output / response101        :return: HttpResponse which can be sent to the calling client102        """103        serialized_response = self._create_default_response(operation_model)104        shape = operation_model.output_shape105        # The shape can also be none (for empty responses), but it still needs to be serialized (to add some metadata)106        shape_members = shape.members if shape is not None else None107        self._serialize_response(108            response, serialized_response, shape, shape_members, operation_model109        )110        serialized_response = self._prepare_additional_traits_in_response(111            serialized_response, operation_model112        )113        return serialized_response114    def serialize_error_to_response(115        self, error: ServiceException, operation_model: OperationModel116    ) -> HttpResponse:117        """118        Takes an error instance and serializes it to an actual HttpResponse.119        Therefore this method is used for errors which should be serialized and transmitted to the calling client.120        :param error: to serialize121        :param operation_model: specification of the service & operation containing information about the shape of the122                                service's output / response123        :return: HttpResponse which can be sent to the calling client124        """125        serialized_response = self._create_default_response(operation_model)126        if isinstance(error, CommonServiceException):127            # Not all possible exceptions are contained in the service's specification.128            # Therefore, service implementations can also throw a "CommonServiceException" to raise arbitrary /129            # non-specified exceptions (where the developer needs to define the data which would usually be taken from130            # the specification, like the "Code").131            code = error.code132            sender_fault = error.sender_fault133            status_code = error.status_code134            shape = None135        else:136            # It it's not a CommonServiceException, the exception is being serialized based on the specification137            # The shape name is equal to the class name (since the classes are generated from the shape's name)138            error_shape_name = error.__class__.__name__139            # Lookup the corresponding error shape in the operation model140            shape = next(141                shape for shape in operation_model.error_shapes if shape.name == error_shape_name142            )143            error_spec = shape.metadata.get("error", {})144            status_code = error_spec.get("httpStatusCode")145            # If the code is not explicitly set, it's typically the shape's name146            code = error_spec.get("code", shape.name)147            # The senderFault is only set if the "senderFault" is true148            # (there are no examples which show otherwise)149            sender_fault = error_spec.get("senderFault")150        # Some specifications do not contain the httpStatusCode field.151        # These errors typically have the http status code 400.152        serialized_response.status_code = status_code or 400153        self._serialize_error(154            error, code, sender_fault, serialized_response, shape, operation_model155        )156        serialized_response = self._prepare_additional_traits_in_response(157            serialized_response, operation_model158        )159        return serialized_response160    def _serialize_response(161        self,162        parameters: dict,163        response: HttpResponse,164        shape: Optional[Shape],165        shape_members: dict,166        operation_model: OperationModel,167    ) -> None: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        response: HttpResponse,176        shape: Shape,177        operation_model: OperationModel,178    ) -> None:179        raise NotImplementedError180    def _create_default_response(self, operation_model: OperationModel) -> HttpResponse:181        """182        Creates a boilerplate default response to be used by subclasses as starting points.183        Uses the default HTTP response status code defined in the operation model (if defined).184        :param operation_model: to extract the default HTTP status code185        :return: boilerplate HTTP response186        """187        return HttpResponse(response=b"", status=operation_model.http.get("responseCode", 200))188    # Some extra utility methods subclasses can use.189    @staticmethod190    def _timestamp_iso8601(value: datetime) -> str:191        if value.microsecond > 0:192            timestamp_format = ISO8601_MICRO193        else:194            timestamp_format = ISO8601195        return value.strftime(timestamp_format)196    @staticmethod197    def _timestamp_unixtimestamp(value: datetime) -> float:198        return round(value.timestamp(), 3)199    def _timestamp_rfc822(self, value: datetime) -> str:200        if isinstance(value, datetime):201            value = self._timestamp_unixtimestamp(value)202        return formatdate(value, usegmt=True)203    def _convert_timestamp_to_str(204        self, value: Union[int, str, datetime], timestamp_format=None205    ) -> str:206        if timestamp_format is None:207            timestamp_format = self.TIMESTAMP_FORMAT208        timestamp_format = timestamp_format.lower()209        datetime_obj = parse_to_aware_datetime(value)210        converter = getattr(self, "_timestamp_%s" % timestamp_format)211        final_value = converter(datetime_obj)212        return final_value213    @staticmethod214    def _get_serialized_name(shape: Shape, default_name: str) -> str:215        """216        Returns the serialized name for the shape if it exists.217        Otherwise it will return the passed in default_name.218        """219        return shape.serialization.get("name", default_name)220    def _get_base64(self, value: Union[str, bytes]):221        """222        Returns the base64-encoded version of value, handling223        both strings and bytes. The returned value is a string224        via the default encoding.225        """226        if isinstance(value, six.text_type):227            value = value.encode(self.DEFAULT_ENCODING)228        return base64.b64encode(value).strip().decode(self.DEFAULT_ENCODING)229    def _prepare_additional_traits_in_response(230        self, response: HttpResponse, operation_model: OperationModel231    ):232        """Applies additional traits on the raw response for a given model or protocol."""233        if operation_model.http_checksum_required:234            add_md5_header(response)235        return response236def add_md5_header(response: HttpResponse):237    """Add a Content-MD5 header if not yet there. Adapted from botocore.utils"""238    headers = response.headers239    body = response.data240    if body is not None and "Content-MD5" not in headers:241        md5_digest = calculate_md5(body)242        headers["Content-MD5"] = md5_digest243class BaseXMLResponseSerializer(ResponseSerializer):244    """245    The BaseXMLResponseSerializer performs the basic logic for the XML response serialization.246    It is slightly adapted by the QueryResponseSerializer.247    While the botocore's RestXMLSerializer is quite similar, there are some subtle differences (since botocore's248    implementation handles the serialization of the requests from the client to the service, not the responses from the249    service to the client).250    **Experimental:** This serializer is still experimental.251    When implementing services with this serializer, some edge cases might not work out-of-the-box.252    """253    def _serialize_response(254        self,255        parameters: dict,256        response: HttpResponse,257        shape: Optional[Shape],258        shape_members: dict,259        operation_model: OperationModel,260    ) -> None:261        """262        Serializes the given parameters as XML.263        :param parameters: The user input params264        :param response: The final serialized HttpResponse265        :param shape: Describes the expected output shape (can be None in case of an "empty" response)266        :param shape_members: The members of the output struct shape267        :param operation_model: The specification of the operation of which the response is serialized here268        :return: None - the given `serialized` dict is modified269        """270        # TODO implement the handling of location traits (where "location" is "header", "headers")271        payload_member = shape.serialization.get("payload") if shape is not None else None272        if payload_member is not None and shape_members[payload_member].type_name in [273            "blob",274            "string",275        ]:276            # If it's streaming, then the body is just the value of the payload.277            body_payload = parameters.get(payload_member, b"")278            body_payload = self._encode_payload(body_payload)279            response.data = body_payload280        elif payload_member is not None:281            # If there's a payload member, we serialized that member to the body.282            body_params = parameters.get(payload_member)283            if body_params is not None:284                response.data = self._encode_payload(285                    self._serialize_body_params(286                        body_params, shape_members[payload_member], operation_model287                    )288                )289        else:290            # Otherwise, we use the "traditional" way of serializing the whole parameters dict recursively.291            response.data = self._encode_payload(292                self._serialize_body_params(parameters, shape, operation_model)293            )294    def _serialize_error(295        self,296        error: ServiceException,297        code: str,298        sender_fault: bool,299        response: HttpResponse,300        shape: Shape,301        operation_model: OperationModel,302    ) -> None:303        # TODO handle error shapes with members304        # Check if we need to add a namespace305        attr = (306            {"xmlns": operation_model.metadata.get("xmlNamespace")}307            if "xmlNamespace" in operation_model.metadata308            else {}309        )310        root = ETree.Element("ErrorResponse", attr)311        error_tag = ETree.SubElement(root, "Error")312        self._add_error_tags(code, error, error_tag, sender_fault)313        request_id = ETree.SubElement(root, "RequestId")314        request_id.text = gen_amzn_requestid_long()315        response.data = self._encode_payload(self._xml_to_string(root))316    def _add_error_tags(317        self, code: str, error: ServiceException, error_tag: ETree.Element, sender_fault: bool318    ) -> None:319        code_tag = ETree.SubElement(error_tag, "Code")320        code_tag.text = code321        message = str(error)322        if len(message) > 0:323            self._default_serialize(error_tag, message, None, "Message")324        if sender_fault:325            # The sender fault is either not set or "Sender"326            self._default_serialize(error_tag, "Sender", None, "Fault")327    def _serialize_body_params(328        self, params: dict, shape: Shape, operation_model: OperationModel329    ) -> Optional[str]:330        root = self._serialize_body_params_to_xml(params, shape, operation_model)331        self._prepare_additional_traits_in_xml(root)332        return self._xml_to_string(root)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            if value is None:370                # Don't serialize any param whose value is None.371                continue372            member_shape = shape.members[key]373            member_name = member_shape.serialization.get("name", key)374            # We need to special case member shapes that are marked as an xmlAttribute.375            # Rather than serializing into an XML child node, we instead serialize the shape to376            # an XML attribute of the *current* node.377            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        if params is None:387            # Don't serialize any param whose value is None.388            return389        member_shape = shape.member390        if shape.serialization.get("flattened"):391            # If the list is flattened, either take the member's "name" or the name of the usual name for the parent392            # element for the children.393            element_name = self._get_serialized_name(member_shape, name)394            list_node = xmlnode395        else:396            element_name = self._get_serialized_name(member_shape, "member")397            list_node = ETree.SubElement(xmlnode, name)398        for item in params:399            # Don't serialize any item which is None400            if item is not None:401                self._serialize(member_shape, item, list_node, element_name)402    def _serialize_type_map(403        self, xmlnode: ETree.Element, params: dict, shape: MapShape, name: str404    ) -> None:405        """406        Given the ``name`` of MyMap, an input of {"key1": "val1", "key2": "val2"}, and the ``flattened: False``407        we serialize this as:408          <MyMap>409            <entry>410              <key>key1</key>411              <value>val1</value>412            </entry>413            <entry>414              <key>key2</key>415              <value>val2</value>416            </entry>417          </MyMap>418        If it is flattened, it is serialized as follows:419          <MyMap>420            <key>key1</key>421            <value>val1</value>422          </MyMap>423          <MyMap>424            <key>key2</key>425            <value>val2</value>426          </MyMap>427        """428        if params is None:429            # Don't serialize a non-existing map430            return431        if shape.serialization.get("flattened"):432            entries_node = xmlnode433            entry_node_name = name434        else:435            entries_node = ETree.SubElement(xmlnode, name)436            entry_node_name = "entry"437        for key, value in params.items():438            if value is None:439                # Don't serialize any param whose value is None.440                continue441            entry_node = ETree.SubElement(entries_node, entry_node_name)442            key_name = self._get_serialized_name(shape.key, default_name="key")443            val_name = self._get_serialized_name(shape.value, default_name="value")444            self._serialize(shape.key, key, entry_node, key_name)445            self._serialize(shape.value, value, entry_node, val_name)446    @staticmethod447    def _serialize_type_boolean(xmlnode: ETree.Element, params: bool, _, name: str) -> None:448        """449        For scalar types, the 'params' attr is actually just a scalar value representing the data450        we need to serialize as a boolean. It will either be 'true' or 'false'451        """452        node = ETree.SubElement(xmlnode, name)453        if params:454            str_value = "true"455        else:456            str_value = "false"457        node.text = str_value458    def _serialize_type_blob(459        self, xmlnode: ETree.Element, params: Union[str, bytes], _, name: str460    ) -> None:461        node = ETree.SubElement(xmlnode, name)462        node.text = self._get_base64(params)463    def _serialize_type_timestamp(464        self, xmlnode: ETree.Element, params: str, shape: Shape, name: str465    ) -> None:466        node = ETree.SubElement(xmlnode, name)467        node.text = self._convert_timestamp_to_str(468            params, shape.serialization.get("timestampFormat")469        )470    def _default_serialize(self, xmlnode: ETree.Element, params: str, _, name: str) -> None:471        node = ETree.SubElement(xmlnode, name)472        node.text = six.text_type(params)473    def _prepare_additional_traits_in_xml(self, root: Optional[ETree.Element]):474        """475        Prepares the XML root node before being serialized with additional traits (like the Response ID in the Query476        protocol).477        For some protocols (like rest-xml), the root can be None.478        """479        pass480    def _create_default_response(self, operation_model: OperationModel) -> HttpResponse:481        response = super()._create_default_response(operation_model)482        response.headers["Content-Type"] = "text/xml"483        return response484    def _xml_to_string(self, root: Optional[ETree.ElementTree]) -> Optional[str]:485        """Generates the string representation of the given XML element."""486        if root is not None:487            return ETree.tostring(488                element=root, encoding=self.DEFAULT_ENCODING, xml_declaration=True489            )490class BaseRestResponseSerializer(ResponseSerializer, ABC):491    """492    The BaseRestResponseSerializer performs the basic logic for the ReST response serialization.493    In our case it basically only adds the request metadata to the HTTP header.494    """495    def _prepare_additional_traits_in_response(...

application.py

Source:application.py

...57                    elif isinstance(response, str):58                        return self._create_response(59                            status_code=route.status_code, text=route.endpoint(**kwargs)60                        )61        return self._create_default_response()62    def _get_route(self, path: str) -> typing.Optional[typing.Tuple[Route, Kwargs]]:63        for route_path in self._routes.keys():64            if params := parse(route_path, path):65                return self._routes[route_path], params.named66        return None67    def _get_template(self, name: str, context: typing.Optional[Kwargs] = None):68        context = context or {}69        return self._templates.get_template(name=name).render(**context).encode()70    def _create_template_response(self, template: Template) -> Response:71        response = Response()72        response.status_code = 20073        response.body = self._get_template(name=template.name, context=template.context)74        return response75    @staticmethod76    def _create_response(status_code: int, text: str) -> Response:77        response = Response()78        response.status_code = status_code79        response.text = text80        return response81    @staticmethod82    def _create_default_response() -> Response:...

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.