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

How to use _serialize_content_type method in localstack

Best Python code snippet using localstack_python

serialize.py

Source:serialize.py

...406            serialized['headers'] = partitioned['headers']407        self._serialize_payload(408            partitioned, parameters, serialized, shape, shape_members409        )410        self._serialize_content_type(serialized, shape, shape_members)411        host_prefix = self._expand_host_prefix(parameters, operation_model)412        if host_prefix is not None:413            serialized['host_prefix'] = host_prefix414        return serialized415    def _render_uri_template(self, uri_template, params):416        # We need to handle two cases::417        #418        # /{Bucket}/foo419        # /{Key+}/bar420        # A label ending with '+' is greedy.  There can only421        # be one greedy key.422        encoded_params = {}423        for template_param in re.findall(r'{(.*?)}', uri_template):424            if template_param.endswith('+'):425                encoded_params[template_param] = percent_encode(426                    params[template_param[:-1]], safe='/~'427                )428            else:429                encoded_params[template_param] = percent_encode(430                    params[template_param]431                )432        return uri_template.format(**encoded_params)433    def _serialize_payload(434        self, partitioned, parameters, serialized, shape, shape_members435    ):436        # partitioned - The user input params partitioned by location.437        # parameters - The user input params.438        # serialized - The final serialized request dict.439        # shape - Describes the expected input shape440        # shape_members - The members of the input struct shape441        payload_member = shape.serialization.get('payload')442        if self._has_streaming_payload(payload_member, shape_members):443            # If it's streaming, then the body is just the444            # value of the payload.445            body_payload = parameters.get(payload_member, b'')446            body_payload = self._encode_payload(body_payload)447            serialized['body'] = body_payload448        elif payload_member is not None:449            # If there's a payload member, we serialized that450            # member to they body.451            body_params = parameters.get(payload_member)452            if body_params is not None:453                serialized['body'] = self._serialize_body_params(454                    body_params, shape_members[payload_member]455                )456            else:457                serialized['body'] = self._serialize_empty_body()458        elif partitioned['body_kwargs']:459            serialized['body'] = self._serialize_body_params(460                partitioned['body_kwargs'], shape461            )462        elif self._requires_empty_body(shape):463            serialized['body'] = self._serialize_empty_body()464    def _serialize_empty_body(self):465        return b''466    def _serialize_content_type(self, serialized, shape, shape_members):467        """468        Some protocols require varied Content-Type headers469        depending on user input. This allows subclasses to apply470        this conditionally.471        """472        pass473    def _requires_empty_body(self, shape):474        """475        Some protocols require a specific body to represent an empty476        payload. This allows subclasses to apply this conditionally.477        """478        return False479    def _has_streaming_payload(self, payload, shape_members):480        """Determine if payload is streaming (a blob or string)."""481        return payload is not None and shape_members[payload].type_name in (482            'blob',483            'string',484        )485    def _encode_payload(self, body):486        if isinstance(body, str):487            return body.encode(self.DEFAULT_ENCODING)488        return body489    def _partition_parameters(490        self, partitioned, param_name, param_value, shape_members491    ):492        # This takes the user provided input parameter (``param``)493        # and figures out where they go in the request dict.494        # Some params are HTTP headers, some are used in the URI, some495        # are in the request body.  This method deals with this.496        member = shape_members[param_name]497        location = member.serialization.get('location')498        key_name = member.serialization.get('name', param_name)499        if location == 'uri':500            partitioned['uri_path_kwargs'][key_name] = param_value501        elif location == 'querystring':502            if isinstance(param_value, dict):503                partitioned['query_string_kwargs'].update(param_value)504            elif isinstance(param_value, bool):505                bool_str = str(param_value).lower()506                partitioned['query_string_kwargs'][key_name] = bool_str507            elif member.type_name == 'timestamp':508                timestamp_format = member.serialization.get(509                    'timestampFormat', self.QUERY_STRING_TIMESTAMP_FORMAT510                )511                timestamp = self._convert_timestamp_to_str(512                    param_value, timestamp_format513                )514                partitioned['query_string_kwargs'][key_name] = timestamp515            else:516                partitioned['query_string_kwargs'][key_name] = param_value517        elif location == 'header':518            shape = shape_members[param_name]519            if not param_value and shape.type_name == 'list':520                # Empty lists should not be set on the headers521                return522            value = self._convert_header_value(shape, param_value)523            partitioned['headers'][key_name] = str(value)524        elif location == 'headers':525            # 'headers' is a bit of an oddball.  The ``key_name``526            # is actually really a prefix for the header names:527            header_prefix = key_name528            # The value provided by the user is a dict so we'll be529            # creating multiple header key/val pairs.  The key530            # name to use for each header is the header_prefix (``key_name``)531            # plus the key provided by the user.532            self._do_serialize_header_map(533                header_prefix, partitioned['headers'], param_value534            )535        else:536            partitioned['body_kwargs'][param_name] = param_value537    def _do_serialize_header_map(self, header_prefix, headers, user_input):538        for key, val in user_input.items():539            full_key = header_prefix + key540            headers[full_key] = val541    def _serialize_body_params(self, params, shape):542        raise NotImplementedError('_serialize_body_params')543    def _convert_header_value(self, shape, value):544        if shape.type_name == 'timestamp':545            datetime_obj = parse_to_aware_datetime(value)546            timestamp = calendar.timegm(datetime_obj.utctimetuple())547            timestamp_format = shape.serialization.get(548                'timestampFormat', self.HEADER_TIMESTAMP_FORMAT549            )550            return self._convert_timestamp_to_str(timestamp, timestamp_format)551        elif shape.type_name == 'list':552            converted_value = [553                self._convert_header_value(shape.member, v)554                for v in value555                if v is not None556            ]557            return ",".join(converted_value)558        elif is_json_value_header(shape):559            # Serialize with no spaces after separators to save space in560            # the header.561            return self._get_base64(json.dumps(value, separators=(',', ':')))562        else:563            return value564class RestJSONSerializer(BaseRestSerializer, JSONSerializer):565    def _serialize_empty_body(self):566        return b'{}'567    def _requires_empty_body(self, shape):568        """569        Serialize an empty JSON object whenever the shape has570        members not targeting a location.571        """572        for member, val in shape.members.items():573            if 'location' not in val.serialization:574                return True575        return False576    def _serialize_content_type(self, serialized, shape, shape_members):577        """Set Content-Type to application/json for all structured bodies."""578        payload = shape.serialization.get('payload')579        if self._has_streaming_payload(payload, shape_members):580            # Don't apply content-type to streaming bodies581            return582        has_body = serialized['body'] != b''583        has_content_type = has_header('Content-Type', serialized['headers'])584        if has_body and not has_content_type:585            serialized['headers']['Content-Type'] = 'application/json'586    def _serialize_body_params(self, params, shape):587        serialized_body = self.MAP_TYPE()588        self._serialize(serialized_body, params, shape)589        return json.dumps(serialized_body).encode(self.DEFAULT_ENCODING)590class RestXMLSerializer(BaseRestSerializer):...

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.