KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _default_serialize method in localstack

Best Python code snippet using localstack_python

serialize.py

Source:serialize.py

...208        if value:209            serialized[prefix] = 'true'210        else:211            serialized[prefix] = 'false'212    def _default_serialize(self, serialized, value, shape, prefix=''):213        serialized[prefix] = value214    def _is_shape_flattened(self, shape):215        return shape.serialization.get('flattened')216class EC2Serializer(QuerySerializer):217    """EC2 specific customizations to the query protocol serializers.218    The EC2 model is almost, but not exactly, similar to the query protocol219    serializer.  This class encapsulates those differences.  The model220    will have be marked with a ``protocol`` of ``ec2``, so you don't need221    to worry about wiring this class up correctly.222    """223    def _get_serialized_name(self, shape, default_name):224        # Returns the serialized name for the shape if it exists.225        # Otherwise it will return the passed in default_name.226        if 'queryName' in shape.serialization:227            return shape.serialization['queryName']228        elif 'name' in shape.serialization:229            # A locationName is always capitalized230            # on input for the ec2 protocol.231            name = shape.serialization['name']232            return name[0].upper() + name[1:]233        else:234            return default_name235    def _serialize_type_list(self, serialized, value, shape, prefix=''):236        for i, element in enumerate(value, 1):237            element_prefix = '%s.%s' % (prefix, i)238            element_shape = shape.member239            self._serialize(serialized, element, element_shape, element_prefix)240class JSONSerializer(Serializer):241    TIMESTAMP_FORMAT = 'unixtimestamp'242    def serialize_to_request(self, parameters, operation_model):243        target = '%s.%s' % (operation_model.metadata['targetPrefix'],244                            operation_model.name)245        json_version = operation_model.metadata['jsonVersion']246        serialized = self._create_default_request()247        serialized['method'] = operation_model.http.get('method',248                                                        self.DEFAULT_METHOD)249        serialized['headers'] = {250            'X-Amz-Target': target,251            'Content-Type': 'application/x-amz-json-%s' % json_version,252        }253        body = {}254        input_shape = operation_model.input_shape255        if input_shape is not None:256            self._serialize(body, parameters, input_shape)257        serialized['body'] = json.dumps(body).encode(self.DEFAULT_ENCODING)258        return serialized259    def _serialize(self, serialized, value, shape, key=None):260        method = getattr(self, '_serialize_type_%s' % shape.type_name,261                         self._default_serialize)262        method(serialized, value, shape, key)263    def _serialize_type_structure(self, serialized, value, shape, key):264        if key is not None:265            # If a key is provided, this is a result of a recursive266            # call so we need to add a new child dict as the value267            # of the passed in serialized dict.  We'll then add268            # all the structure members as key/vals in the new serialized269            # dictionary we just created.270            new_serialized = self.MAP_TYPE()271            serialized[key] = new_serialized272            serialized = new_serialized273        members = shape.members274        for member_key, member_value in value.items():275            member_shape = members[member_key]276            if 'name' in member_shape.serialization:277                member_key = member_shape.serialization['name']278            self._serialize(serialized, member_value, member_shape, member_key)279    def _serialize_type_map(self, serialized, value, shape, key):280        map_obj = self.MAP_TYPE()281        serialized[key] = map_obj282        for sub_key, sub_value in value.items():283            self._serialize(map_obj, sub_value, shape.value, sub_key)284    def _serialize_type_list(self, serialized, value, shape, key):285        list_obj = []286        serialized[key] = list_obj287        for list_item in value:288            wrapper = {}289            # The JSON list serialization is the only case where we aren't290            # setting a key on a dict.  We handle this by using291            # a __current__ key on a wrapper dict to serialize each292            # list item before appending it to the serialized list.293            self._serialize(wrapper, list_item, shape.member, "__current__")294            list_obj.append(wrapper["__current__"])295    def _default_serialize(self, serialized, value, shape, key):296        serialized[key] = value297    def _serialize_type_timestamp(self, serialized, value, shape, key):298        serialized[key] = self._convert_timestamp_to_str(value)299    def _serialize_type_blob(self, serialized, value, shape, key):300        serialized[key] = self._get_base64(value)301class BaseRestSerializer(Serializer):302    """Base class for rest protocols.303    The only variance between the various rest protocols is the304    way that the body is serialized.  All other aspects (headers, uri, etc.)305    are the same and logic for serializing those aspects lives here.306    Subclasses must implement the ``_serialize_body_params`` method.307    """308    # This is a list of known values for the "location" key in the309    # serialization dict.  The location key tells us where on the request310    # to put the serialized value.311    KNOWN_LOCATIONS = ['uri', 'querystring', 'header', 'headers']312    def serialize_to_request(self, parameters, operation_model):313        serialized = self._create_default_request()314        serialized['method'] = operation_model.http.get('method',315                                                        self.DEFAULT_METHOD)316        shape = operation_model.input_shape317        if shape is None:318            serialized['url_path'] = operation_model.http['requestUri']319            return serialized320        shape_members = shape.members321        # While the ``serialized`` key holds the final serialized request322        # data, we need interim dicts for the various locations of the323        # request.  We need this for the uri_path_kwargs and the324        # query_string_kwargs because they are templated, so we need325        # to gather all the needed data for the string template,326        # then we render the template.  The body_kwargs is needed327        # because once we've collected them all, we run them through328        # _serialize_body_params, which for rest-json, creates JSON,329        # and for rest-xml, will create XML.  This is what the330        # ``partitioned`` dict below is for.331        partitioned = {332            'uri_path_kwargs': self.MAP_TYPE(),333            'query_string_kwargs': self.MAP_TYPE(),334            'body_kwargs': self.MAP_TYPE(),335            'headers': self.MAP_TYPE(),336        }337        for param_name, param_value in parameters.items():338            if param_value is None:339                # Don't serialize any parameter with a None value.340                continue341            self._partition_parameters(partitioned, param_name, param_value,342                                       shape_members)343        serialized['url_path'] = self._render_uri_template(344            operation_model.http['requestUri'],345            partitioned['uri_path_kwargs'])346        # Note that we lean on the http implementation to handle the case347        # where the requestUri path already has query parameters.348        # The bundled http client, requests, already supports this.349        serialized['query_string'] = partitioned['query_string_kwargs']350        if partitioned['headers']:351            serialized['headers'] = partitioned['headers']352        self._serialize_payload(partitioned, parameters,353                                serialized, shape, shape_members)354        return serialized355    def _render_uri_template(self, uri_template, params):356        # We need to handle two cases::357        #358        # /{Bucket}/foo359        # /{Key+}/bar360        # A label ending with '+' is greedy.  There can only361        # be one greedy key.362        encoded_params = {}363        for template_param in re.findall(r'{(.*?)}', uri_template):364            if template_param.endswith('+'):365                encoded_params[template_param] = percent_encode(366                    params[template_param[:-1]], safe='/~')367            else:368                encoded_params[template_param] = percent_encode(369                    params[template_param])370        return uri_template.format(**encoded_params)371    def _serialize_payload(self, partitioned, parameters,372                           serialized, shape, shape_members):373        # partitioned - The user input params partitioned by location.374        # parameters - The user input params.375        # serialized - The final serialized request dict.376        # shape - Describes the expected input shape377        # shape_members - The members of the input struct shape378        payload_member = shape.serialization.get('payload')379        if payload_member is not None and \380                shape_members[payload_member].type_name in ['blob', 'string']:381            # If it's streaming, then the body is just the382            # value of the payload.383            body_payload = parameters.get(payload_member, b'')384            body_payload = self._encode_payload(body_payload)385            serialized['body'] = body_payload386        elif payload_member is not None:387            # If there's a payload member, we serialized that388            # member to they body.389            body_params = parameters.get(payload_member)390            if body_params is not None:391                serialized['body'] = self._serialize_body_params(392                    body_params,393                    shape_members[payload_member])394        elif partitioned['body_kwargs']:395            serialized['body'] = self._serialize_body_params(396                partitioned['body_kwargs'], shape)397    def _encode_payload(self, body):398        if isinstance(body, six.text_type):399            return body.encode(self.DEFAULT_ENCODING)400        return body401    def _partition_parameters(self, partitioned, param_name,402                              param_value, shape_members):403        # This takes the user provided input parameter (``param``)404        # and figures out where they go in the request dict.405        # Some params are HTTP headers, some are used in the URI, some406        # are in the request body.  This method deals with this.407        member = shape_members[param_name]408        location = member.serialization.get('location')409        key_name = member.serialization.get('name', param_name)410        if location == 'uri':411            partitioned['uri_path_kwargs'][key_name] = param_value412        elif location == 'querystring':413            if isinstance(param_value, dict):414                partitioned['query_string_kwargs'].update(param_value)415            else:416                partitioned['query_string_kwargs'][key_name] = param_value417        elif location == 'header':418            shape = shape_members[param_name]419            value = self._convert_header_value(shape, param_value)420            partitioned['headers'][key_name] = str(value)421        elif location == 'headers':422            # 'headers' is a bit of an oddball.  The ``key_name``423            # is actually really a prefix for the header names:424            header_prefix = key_name425            # The value provided by the user is a dict so we'll be426            # creating multiple header key/val pairs.  The key427            # name to use for each header is the header_prefix (``key_name``)428            # plus the key provided by the user.429            self._do_serialize_header_map(header_prefix,430                                          partitioned['headers'],431                                          param_value)432        else:433            partitioned['body_kwargs'][param_name] = param_value434    def _do_serialize_header_map(self, header_prefix, headers, user_input):435        for key, val in user_input.items():436            full_key = header_prefix + key437            headers[full_key] = val438    def _serialize_body_params(self, params, shape):439        raise NotImplementedError('_serialize_body_params')440    def _convert_header_value(self, shape, value):441        if shape.type_name == 'timestamp':442            datetime_obj = parse_to_aware_datetime(value)443            timestamp = calendar.timegm(datetime_obj.utctimetuple())444            return self._timestamp_rfc822(timestamp)445        else:446            return value447class RestJSONSerializer(BaseRestSerializer, JSONSerializer):448    def _serialize_body_params(self, params, shape):449        serialized_body = self.MAP_TYPE()450        self._serialize(serialized_body, params, shape)451        return json.dumps(serialized_body).encode(self.DEFAULT_ENCODING)452class RestXMLSerializer(BaseRestSerializer):453    TIMESTAMP_FORMAT = 'iso8601'454    def _serialize_body_params(self, params, shape):455        root_name = shape.serialization['name']456        pseudo_root = ElementTree.Element('')457        self._serialize(shape, params, pseudo_root, root_name)458        real_root = list(pseudo_root)[0]459        return ElementTree.tostring(real_root, encoding=self.DEFAULT_ENCODING)460    def _serialize(self, shape, params, xmlnode, name):461        method = getattr(self, '_serialize_type_%s' % shape.type_name,462                         self._default_serialize)463        method(xmlnode, params, shape, name)464    def _serialize_type_structure(self, xmlnode, params, shape, name):465        structure_node = ElementTree.SubElement(xmlnode, name)466        if 'xmlNamespace' in shape.serialization:467            namespace_metadata = shape.serialization['xmlNamespace']468            attribute_name = 'xmlns'469            if namespace_metadata.get('prefix'):470                attribute_name += ':%s' % namespace_metadata['prefix']471            structure_node.attrib[attribute_name] = namespace_metadata['uri']472        for key, value in params.items():473            member_shape = shape.members[key]474            member_name = member_shape.serialization.get('name', key)475            # We need to special case member shapes that are marked as an476            # xmlAttribute.  Rather than serializing into an XML child node,477            # we instead serialize the shape to an XML attribute of the478            # *current* node.479            if value is None:480                # Don't serialize any param whose value is None.481                return482            if member_shape.serialization.get('xmlAttribute'):483                # xmlAttributes must have a serialization name.484                xml_attribute_name = member_shape.serialization['name']485                structure_node.attrib[xml_attribute_name] = value486                continue487            self._serialize(member_shape, value, structure_node, member_name)488    def _serialize_type_list(self, xmlnode, params, shape, name):489        member_shape = shape.member490        if shape.serialization.get('flattened'):491            element_name = name492            list_node = xmlnode493        else:494            element_name = member_shape.serialization.get('name', 'member')495            list_node = ElementTree.SubElement(xmlnode, name)496        for item in params:497            self._serialize(member_shape, item, list_node, element_name)498    def _serialize_type_map(self, xmlnode, params, shape, name):499        # Given the ``name`` of MyMap, and input of {"key1": "val1"}500        # we serialize this as:501        #   <MyMap>502        #     <entry>503        #       <key>key1</key>504        #       <value>val1</value>505        #     </entry>506        #  </MyMap>507        node = ElementTree.SubElement(xmlnode, name)508        # TODO: handle flattened maps.509        for key, value in params.items():510            entry_node = ElementTree.SubElement(node, 'entry')511            key_name = self._get_serialized_name(shape.key, default_name='key')512            val_name = self._get_serialized_name(shape.value,513                                                 default_name='value')514            self._serialize(shape.key, key, entry_node, key_name)515            self._serialize(shape.value, value, entry_node, val_name)516    def _serialize_type_boolean(self, xmlnode, params, shape, name):517        # For scalar types, the 'params' attr is actually just a scalar518        # value representing the data we need to serialize as a boolean.519        # It will either be 'true' or 'false'520        node = ElementTree.SubElement(xmlnode, name)521        if params:522            str_value = 'true'523        else:524            str_value = 'false'525        node.text = str_value526    def _serialize_type_blob(self, xmlnode, params, shape, name):527        node = ElementTree.SubElement(xmlnode, name)528        node.text = self._get_base64(params)529    def _serialize_type_timestamp(self, xmlnode, params, shape, name):530        node = ElementTree.SubElement(xmlnode, name)531        node.text = self._convert_timestamp_to_str(params)532    def _default_serialize(self, xmlnode, params, shape, name):533        node = ElementTree.SubElement(xmlnode, name)534        node.text = str(params)535SERIALIZERS = {536    'ec2': EC2Serializer,537    'query': QuerySerializer,538    'json': JSONSerializer,539    'rest-json': RestJSONSerializer,540    'rest-xml': RestXMLSerializer,...

config.py

Source:config.py

...34# client35class Client(txwac.Client):36    config = default_config37    @staticmethod38    def _default_serialize(o):39        if isinstance(o, datetime):40            return o.isoformat() + 'Z'41        raise TypeError(42            'Object of type {} with value of {} is not '43            'JSON serializable'.format(type(o), repr(o))44        )45    def _serialize(self, data):46        data = json.dumps(data, default=self._default_serialize)47        return 'application/json', data48    @staticmethod49    def _parse_deserialized(e):50        if isinstance(e, dict):51            for k in e.iterkeys():52                if k.endswith('_at') and isinstance(e[k], basestring):...

cache_serialization_decorator.py

Source:cache_serialization_decorator.py

...40    @staticmethod41    def _noop(x):42        return x43    @staticmethod44    def _default_serialize(obj) -> str:45        return json.dumps(serialize(obj, preserve_iterable_types=True))46    @staticmethod47    def _default_deserialize(data: Optional[str]):48        if data is None:49            return data...

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.