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

Best Python code snippet using localstack_python
serialize.py
Source:serialize.py
...120    def _timestamp_rfc822(self, value):121        if isinstance(value, datetime.datetime):122            value = self._timestamp_unixtimestamp(value)123        return formatdate(value, usegmt=True)124    def _convert_timestamp_to_str(self, value, timestamp_format=None):125        if timestamp_format is None:126            timestamp_format = self.TIMESTAMP_FORMAT127        timestamp_format = timestamp_format.lower()128        datetime_obj = parse_to_aware_datetime(value)129        converter = getattr(130            self, '_timestamp_%s' % timestamp_format)131        final_value = converter(datetime_obj)132        return final_value133    def _get_serialized_name(self, shape, default_name):134        # Returns the serialized name for the shape if it exists.135        # Otherwise it will return the passed in default_name.136        return shape.serialization.get('name', default_name)137    def _get_base64(self, value):138        # Returns the base64-encoded version of value, handling139        # both strings and bytes. The returned value is a string140        # via the default encoding.141        if isinstance(value, six.text_type):142            value = value.encode(self.DEFAULT_ENCODING)143        return base64.b64encode(value).strip().decode(144            self.DEFAULT_ENCODING)145    def _expand_host_prefix(self, parameters, operation_model):146        operation_endpoint = operation_model.endpoint147        if operation_endpoint is None:148            return None149        host_prefix_expression = operation_endpoint['hostPrefix']150        input_members = operation_model.input_shape.members151        host_labels = [152            member for member, shape in input_members.items()153            if shape.serialization.get('hostLabel')154        ]155        format_kwargs = dict((name, parameters[name]) for name in host_labels)156        return host_prefix_expression.format(**format_kwargs)157    def _prepare_additional_traits(self, request, operation_model):158        """Determine if additional traits are required for given model"""159        if operation_model.http_checksum_required:160            conditionally_calculate_md5(request)161        return request162class QuerySerializer(Serializer):163    TIMESTAMP_FORMAT = 'iso8601'164    def serialize_to_request(self, parameters, operation_model):165        shape = operation_model.input_shape166        serialized = self._create_default_request()167        serialized['method'] = operation_model.http.get('method',168                                                        self.DEFAULT_METHOD)169        serialized['headers'] = {170            'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8'171        }172        # The query serializer only deals with body params so173        # that's what we hand off the _serialize_* methods.174        body_params = self.MAP_TYPE()175        body_params['Action'] = operation_model.name176        body_params['Version'] = operation_model.metadata['apiVersion']177        if shape is not None:178            self._serialize(body_params, parameters, shape)179        serialized['body'] = body_params180        host_prefix = self._expand_host_prefix(parameters, operation_model)181        if host_prefix is not None:182            serialized['host_prefix'] = host_prefix183        serialized = self._prepare_additional_traits(184            serialized, operation_model185        )186        return serialized187    def _serialize(self, serialized, value, shape, prefix=''):188        # serialized: The dict that is incrementally added to with the189        #             final serialized parameters.190        # value: The current user input value.191        # shape: The shape object that describes the structure of the192        #        input.193        # prefix: The incrementally built up prefix for the serialized194        #         key (i.e Foo.bar.members.1).195        method = getattr(self, '_serialize_type_%s' % shape.type_name,196                         self._default_serialize)197        method(serialized, value, shape, prefix=prefix)198    def _serialize_type_structure(self, serialized, value, shape, prefix=''):199        members = shape.members200        for key, value in value.items():201            member_shape = members[key]202            member_prefix = self._get_serialized_name(member_shape, key)203            if prefix:204                member_prefix = '%s.%s' % (prefix, member_prefix)205            self._serialize(serialized, value, member_shape, member_prefix)206    def _serialize_type_list(self, serialized, value, shape, prefix=''):207        if not value:208            # The query protocol serializes empty lists.209            serialized[prefix] = ''210            return211        if self._is_shape_flattened(shape):212            list_prefix = prefix213            if shape.member.serialization.get('name'):214                name = self._get_serialized_name(shape.member, default_name='')215                # Replace '.Original' with '.{name}'.216                list_prefix = '.'.join(prefix.split('.')[:-1] + [name])217        else:218            list_name = shape.member.serialization.get('name', 'member')219            list_prefix = '%s.%s' % (prefix, list_name)220        for i, element in enumerate(value, 1):221            element_prefix = '%s.%s' % (list_prefix, i)222            element_shape = shape.member223            self._serialize(serialized, element, element_shape, element_prefix)224    def _serialize_type_map(self, serialized, value, shape, prefix=''):225        if self._is_shape_flattened(shape):226            full_prefix = prefix227        else:228            full_prefix = '%s.entry' % prefix229        template = full_prefix + '.{i}.{suffix}'230        key_shape = shape.key231        value_shape = shape.value232        key_suffix = self._get_serialized_name(key_shape, default_name='key')233        value_suffix = self._get_serialized_name(value_shape, 'value')234        for i, key in enumerate(value, 1):235            key_prefix = template.format(i=i, suffix=key_suffix)236            value_prefix = template.format(i=i, suffix=value_suffix)237            self._serialize(serialized, key, key_shape, key_prefix)238            self._serialize(serialized, value[key], value_shape, value_prefix)239    def _serialize_type_blob(self, serialized, value, shape, prefix=''):240        # Blob args must be base64 encoded.241        serialized[prefix] = self._get_base64(value)242    def _serialize_type_timestamp(self, serialized, value, shape, prefix=''):243        serialized[prefix] = self._convert_timestamp_to_str(244            value, shape.serialization.get('timestampFormat'))245    def _serialize_type_boolean(self, serialized, value, shape, prefix=''):246        if value:247            serialized[prefix] = 'true'248        else:249            serialized[prefix] = 'false'250    def _default_serialize(self, serialized, value, shape, prefix=''):251        serialized[prefix] = value252    def _is_shape_flattened(self, shape):253        return shape.serialization.get('flattened')254class EC2Serializer(QuerySerializer):255    """EC2 specific customizations to the query protocol serializers.256    The EC2 model is almost, but not exactly, similar to the query protocol257    serializer.  This class encapsulates those differences.  The model258    will have be marked with a ``protocol`` of ``ec2``, so you don't need259    to worry about wiring this class up correctly.260    """261    def _get_serialized_name(self, shape, default_name):262        # Returns the serialized name for the shape if it exists.263        # Otherwise it will return the passed in default_name.264        if 'queryName' in shape.serialization:265            return shape.serialization['queryName']266        elif 'name' in shape.serialization:267            # A locationName is always capitalized268            # on input for the ec2 protocol.269            name = shape.serialization['name']270            return name[0].upper() + name[1:]271        else:272            return default_name273    def _serialize_type_list(self, serialized, value, shape, prefix=''):274        for i, element in enumerate(value, 1):275            element_prefix = '%s.%s' % (prefix, i)276            element_shape = shape.member277            self._serialize(serialized, element, element_shape, element_prefix)278class JSONSerializer(Serializer):279    TIMESTAMP_FORMAT = 'unixtimestamp'280    def serialize_to_request(self, parameters, operation_model):281        target = '%s.%s' % (operation_model.metadata['targetPrefix'],282                            operation_model.name)283        json_version = operation_model.metadata['jsonVersion']284        serialized = self._create_default_request()285        serialized['method'] = operation_model.http.get('method',286                                                        self.DEFAULT_METHOD)287        serialized['headers'] = {288            'X-Amz-Target': target,289            'Content-Type': 'application/x-amz-json-%s' % json_version,290        }291        body = self.MAP_TYPE()292        input_shape = operation_model.input_shape293        if input_shape is not None:294            self._serialize(body, parameters, input_shape)295        serialized['body'] = json.dumps(body).encode(self.DEFAULT_ENCODING)296        host_prefix = self._expand_host_prefix(parameters, operation_model)297        if host_prefix is not None:298            serialized['host_prefix'] = host_prefix299        serialized = self._prepare_additional_traits(300            serialized, operation_model301        )302        return serialized303    def _serialize(self, serialized, value, shape, key=None):304        method = getattr(self, '_serialize_type_%s' % shape.type_name,305                         self._default_serialize)306        method(serialized, value, shape, key)307    def _serialize_type_structure(self, serialized, value, shape, key):308        if shape.is_document_type:309            serialized[key] = value310        else:311            if key is not None:312                # If a key is provided, this is a result of a recursive313                # call so we need to add a new child dict as the value314                # of the passed in serialized dict.  We'll then add315                # all the structure members as key/vals in the new serialized316                # dictionary we just created.317                new_serialized = self.MAP_TYPE()318                serialized[key] = new_serialized319                serialized = new_serialized320            members = shape.members321            for member_key, member_value in value.items():322                member_shape = members[member_key]323                if 'name' in member_shape.serialization:324                    member_key = member_shape.serialization['name']325                self._serialize(serialized, member_value, member_shape, member_key)326    def _serialize_type_map(self, serialized, value, shape, key):327        map_obj = self.MAP_TYPE()328        serialized[key] = map_obj329        for sub_key, sub_value in value.items():330            self._serialize(map_obj, sub_value, shape.value, sub_key)331    def _serialize_type_list(self, serialized, value, shape, key):332        list_obj = []333        serialized[key] = list_obj334        for list_item in value:335            wrapper = {}336            # The JSON list serialization is the only case where we aren't337            # setting a key on a dict.  We handle this by using338            # a __current__ key on a wrapper dict to serialize each339            # list item before appending it to the serialized list.340            self._serialize(wrapper, list_item, shape.member, "__current__")341            list_obj.append(wrapper["__current__"])342    def _default_serialize(self, serialized, value, shape, key):343        serialized[key] = value344    def _serialize_type_timestamp(self, serialized, value, shape, key):345        serialized[key] = self._convert_timestamp_to_str(346            value, shape.serialization.get('timestampFormat'))347    def _serialize_type_blob(self, serialized, value, shape, key):348        serialized[key] = self._get_base64(value)349class BaseRestSerializer(Serializer):350    """Base class for rest protocols.351    The only variance between the various rest protocols is the352    way that the body is serialized.  All other aspects (headers, uri, etc.)353    are the same and logic for serializing those aspects lives here.354    Subclasses must implement the ``_serialize_body_params`` method.355    """356    QUERY_STRING_TIMESTAMP_FORMAT = 'iso8601'357    HEADER_TIMESTAMP_FORMAT = 'rfc822'358    # This is a list of known values for the "location" key in the359    # serialization dict.  The location key tells us where on the request360    # to put the serialized value.361    KNOWN_LOCATIONS = ['uri', 'querystring', 'header', 'headers']362    def serialize_to_request(self, parameters, operation_model):363        serialized = self._create_default_request()364        serialized['method'] = operation_model.http.get('method',365                                                        self.DEFAULT_METHOD)366        shape = operation_model.input_shape367        if shape is None:368            serialized['url_path'] = operation_model.http['requestUri']369            return serialized370        shape_members = shape.members371        # While the ``serialized`` key holds the final serialized request372        # data, we need interim dicts for the various locations of the373        # request.  We need this for the uri_path_kwargs and the374        # query_string_kwargs because they are templated, so we need375        # to gather all the needed data for the string template,376        # then we render the template.  The body_kwargs is needed377        # because once we've collected them all, we run them through378        # _serialize_body_params, which for rest-json, creates JSON,379        # and for rest-xml, will create XML.  This is what the380        # ``partitioned`` dict below is for.381        partitioned = {382            'uri_path_kwargs': self.MAP_TYPE(),383            'query_string_kwargs': self.MAP_TYPE(),384            'body_kwargs': self.MAP_TYPE(),385            'headers': self.MAP_TYPE(),386        }387        for param_name, param_value in parameters.items():388            if param_value is None:389                # Don't serialize any parameter with a None value.390                continue391            self._partition_parameters(partitioned, param_name, param_value,392                                       shape_members)393        serialized['url_path'] = self._render_uri_template(394            operation_model.http['requestUri'],395            partitioned['uri_path_kwargs'])396        # Note that we lean on the http implementation to handle the case397        # where the requestUri path already has query parameters.398        # The bundled http client, requests, already supports this.399        serialized['query_string'] = partitioned['query_string_kwargs']400        if partitioned['headers']:401            serialized['headers'] = partitioned['headers']402        self._serialize_payload(partitioned, parameters,403                                serialized, shape, shape_members)404        self._serialize_content_type(serialized, shape, shape_members)405        host_prefix = self._expand_host_prefix(parameters, operation_model)406        if host_prefix is not None:407            serialized['host_prefix'] = host_prefix408        serialized = self._prepare_additional_traits(409            serialized, operation_model410        )411        return serialized412    def _render_uri_template(self, uri_template, params):413        # We need to handle two cases::414        #415        # /{Bucket}/foo416        # /{Key+}/bar417        # A label ending with '+' is greedy.  There can only418        # be one greedy key.419        encoded_params = {}420        for template_param in re.findall(r'{(.*?)}', uri_template):421            if template_param.endswith('+'):422                encoded_params[template_param] = percent_encode(423                    params[template_param[:-1]], safe='/~')424            else:425                encoded_params[template_param] = percent_encode(426                    params[template_param])427        return uri_template.format(**encoded_params)428    def _serialize_payload(self, partitioned, parameters,429                           serialized, shape, shape_members):430        # partitioned - The user input params partitioned by location.431        # parameters - The user input params.432        # serialized - The final serialized request dict.433        # shape - Describes the expected input shape434        # shape_members - The members of the input struct shape435        payload_member = shape.serialization.get('payload')436        if self._has_streaming_payload(payload_member, shape_members):437            # If it's streaming, then the body is just the438            # value of the payload.439            body_payload = parameters.get(payload_member, b'')440            body_payload = self._encode_payload(body_payload)441            serialized['body'] = body_payload442        elif payload_member is not None:443            # If there's a payload member, we serialized that444            # member to they body.445            body_params = parameters.get(payload_member)446            if body_params is not None:447                serialized['body'] = self._serialize_body_params(448                    body_params,449                    shape_members[payload_member])450            else:451                serialized['body'] = self._serialize_empty_body()452        elif partitioned['body_kwargs']:453            serialized['body'] = self._serialize_body_params(454                partitioned['body_kwargs'], shape)455        elif self._requires_empty_body(shape):456            serialized['body'] = self._serialize_empty_body()457    def _serialize_empty_body(self):458        return b''459    def _serialize_content_type(self, serialized, shape, shape_members):460        """461        Some protocols require varied Content-Type headers462        depending on user input. This allows subclasses to apply463        this conditionally.464        """465        pass466    def _requires_empty_body(self, shape):467        """468        Some protocols require a specific body to represent an empty469        payload. This allows subclasses to apply this conditionally.470        """471        return False472    def _has_streaming_payload(self, payload, shape_members):473        """Determine if payload is streaming (a blob or string)."""474        return (475            payload is not None and476            shape_members[payload].type_name in ['blob', 'string']477        )478    def _encode_payload(self, body):479        if isinstance(body, six.text_type):480            return body.encode(self.DEFAULT_ENCODING)481        return body482    def _partition_parameters(self, partitioned, param_name,483                              param_value, shape_members):484        # This takes the user provided input parameter (``param``)485        # and figures out where they go in the request dict.486        # Some params are HTTP headers, some are used in the URI, some487        # are in the request body.  This method deals with this.488        member = shape_members[param_name]489        location = member.serialization.get('location')490        key_name = member.serialization.get('name', param_name)491        if location == 'uri':492            partitioned['uri_path_kwargs'][key_name] = param_value493        elif location == 'querystring':494            if isinstance(param_value, dict):495                partitioned['query_string_kwargs'].update(param_value)496            elif isinstance(param_value, bool):497                bool_str = str(param_value).lower()498                partitioned['query_string_kwargs'][key_name] = bool_str499            elif member.type_name == 'timestamp':500                timestamp_format = member.serialization.get(501                    'timestampFormat', self.QUERY_STRING_TIMESTAMP_FORMAT)502                timestamp = self._convert_timestamp_to_str(503                    param_value, timestamp_format504                )505                partitioned['query_string_kwargs'][key_name] = timestamp506            else:507                partitioned['query_string_kwargs'][key_name] = param_value508        elif location == 'header':509            shape = shape_members[param_name]510            if not param_value and shape.type_name == 'list':511                # Empty lists should not be set on the headers512                return513            value = self._convert_header_value(shape, param_value)514            partitioned['headers'][key_name] = str(value)515        elif location == 'headers':516            # 'headers' is a bit of an oddball.  The ``key_name``517            # is actually really a prefix for the header names:518            header_prefix = key_name519            # The value provided by the user is a dict so we'll be520            # creating multiple header key/val pairs.  The key521            # name to use for each header is the header_prefix (``key_name``)522            # plus the key provided by the user.523            self._do_serialize_header_map(header_prefix,524                                          partitioned['headers'],525                                          param_value)526        else:527            partitioned['body_kwargs'][param_name] = param_value528    def _do_serialize_header_map(self, header_prefix, headers, user_input):529        for key, val in user_input.items():530            full_key = header_prefix + key531            headers[full_key] = val532    def _serialize_body_params(self, params, shape):533        raise NotImplementedError('_serialize_body_params')534    def _convert_header_value(self, shape, value):535        if shape.type_name == 'timestamp':536            datetime_obj = parse_to_aware_datetime(value)537            timestamp = calendar.timegm(datetime_obj.utctimetuple())538            timestamp_format = shape.serialization.get(539                'timestampFormat', self.HEADER_TIMESTAMP_FORMAT)540            return self._convert_timestamp_to_str(timestamp, timestamp_format)541        elif shape.type_name == 'list':542            converted_value = [543                self._convert_header_value(shape.member, v)544                for v in value if v is not None545            ]546            return ",".join(converted_value)547        elif is_json_value_header(shape):548            # Serialize with no spaces after separators to save space in549            # the header.550            return self._get_base64(json.dumps(value, separators=(',', ':')))551        else:552            return value553class RestJSONSerializer(BaseRestSerializer, JSONSerializer):554    def _serialize_empty_body(self):555        return b'{}'556    def _requires_empty_body(self, shape):557        """558        Serialize an empty JSON object whenever the shape has559        members not targeting a location.560        """561        for member, val in shape.members.items():562            if 'location' not in val.serialization:563                return True564        return False565    def _serialize_content_type(self, serialized, shape, shape_members):566        """Set Content-Type to application/json for all structured bodies."""567        payload = shape.serialization.get('payload')568        if self._has_streaming_payload(payload, shape_members):569            # Don't apply content-type to streaming bodies570            return571        has_body = serialized['body'] != b''572        has_content_type = has_header('Content-Type', serialized['headers'])573        if has_body and not has_content_type:574            serialized['headers']['Content-Type'] = 'application/json'575    def _serialize_body_params(self, params, shape):576        serialized_body = self.MAP_TYPE()577        self._serialize(serialized_body, params, shape)578        return json.dumps(serialized_body).encode(self.DEFAULT_ENCODING)579class RestXMLSerializer(BaseRestSerializer):580    TIMESTAMP_FORMAT = 'iso8601'581    def _serialize_body_params(self, params, shape):582        root_name = shape.serialization['name']583        pseudo_root = ElementTree.Element('')584        self._serialize(shape, params, pseudo_root, root_name)585        real_root = list(pseudo_root)[0]586        return ElementTree.tostring(real_root, encoding=self.DEFAULT_ENCODING)587    def _serialize(self, shape, params, xmlnode, name):588        method = getattr(self, '_serialize_type_%s' % shape.type_name,589                         self._default_serialize)590        method(xmlnode, params, shape, name)591    def _serialize_type_structure(self, xmlnode, params, shape, name):592        structure_node = ElementTree.SubElement(xmlnode, name)593        if 'xmlNamespace' in shape.serialization:594            namespace_metadata = shape.serialization['xmlNamespace']595            attribute_name = 'xmlns'596            if namespace_metadata.get('prefix'):597                attribute_name += ':%s' % namespace_metadata['prefix']598            structure_node.attrib[attribute_name] = namespace_metadata['uri']599        for key, value in params.items():600            member_shape = shape.members[key]601            member_name = member_shape.serialization.get('name', key)602            # We need to special case member shapes that are marked as an603            # xmlAttribute.  Rather than serializing into an XML child node,604            # we instead serialize the shape to an XML attribute of the605            # *current* node.606            if value is None:607                # Don't serialize any param whose value is None.608                return609            if member_shape.serialization.get('xmlAttribute'):610                # xmlAttributes must have a serialization name.611                xml_attribute_name = member_shape.serialization['name']612                structure_node.attrib[xml_attribute_name] = value613                continue614            self._serialize(member_shape, value, structure_node, member_name)615    def _serialize_type_list(self, xmlnode, params, shape, name):616        member_shape = shape.member617        if shape.serialization.get('flattened'):618            element_name = name619            list_node = xmlnode620        else:621            element_name = member_shape.serialization.get('name', 'member')622            list_node = ElementTree.SubElement(xmlnode, name)623        for item in params:624            self._serialize(member_shape, item, list_node, element_name)625    def _serialize_type_map(self, xmlnode, params, shape, name):626        # Given the ``name`` of MyMap, and input of {"key1": "val1"}627        # we serialize this as:628        #   <MyMap>629        #     <entry>630        #       <key>key1</key>631        #       <value>val1</value>632        #     </entry>633        #  </MyMap>634        node = ElementTree.SubElement(xmlnode, name)635        # TODO: handle flattened maps.636        for key, value in params.items():637            entry_node = ElementTree.SubElement(node, 'entry')638            key_name = self._get_serialized_name(shape.key, default_name='key')639            val_name = self._get_serialized_name(shape.value,640                                                 default_name='value')641            self._serialize(shape.key, key, entry_node, key_name)642            self._serialize(shape.value, value, entry_node, val_name)643    def _serialize_type_boolean(self, xmlnode, params, shape, name):644        # For scalar types, the 'params' attr is actually just a scalar645        # value representing the data we need to serialize as a boolean.646        # It will either be 'true' or 'false'647        node = ElementTree.SubElement(xmlnode, name)648        if params:649            str_value = 'true'650        else:651            str_value = 'false'652        node.text = str_value653    def _serialize_type_blob(self, xmlnode, params, shape, name):654        node = ElementTree.SubElement(xmlnode, name)655        node.text = self._get_base64(params)656    def _serialize_type_timestamp(self, xmlnode, params, shape, name):657        node = ElementTree.SubElement(xmlnode, name)658        node.text = self._convert_timestamp_to_str(659            params, shape.serialization.get('timestampFormat'))660    def _default_serialize(self, xmlnode, params, shape, name):661        node = ElementTree.SubElement(xmlnode, name)662        node.text = six.text_type(params)663SERIALIZERS = {664    'ec2': EC2Serializer,665    'query': QuerySerializer,666    'json': JSONSerializer,667    'rest-json': RestJSONSerializer,668    'rest-xml': RestXMLSerializer,...
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.