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

How to use _serialize_body_params method in localstack

Best Python code snippet using localstack_python

serialize.py

Source:serialize.py

...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'...

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.