How to use _prepare_additional_traits_in_xml method in localstack

Best Python code snippet using localstack_python

serializer.py

Source:serializer.py Github

copy

Full Screen

...327 def _serialize_body_params(328 self, params: dict, shape: Shape, operation_model: OperationModel329 ) -> str:330 root = self._serialize_body_params_to_xml(params, shape, operation_model)331 self._prepare_additional_traits_in_xml(root)332 return ETree.tostring(root, encoding=self.DEFAULT_ENCODING)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 member_shape = shape.members[key]370 member_name = member_shape.serialization.get("name", key)371 # We need to special case member shapes that are marked as an xmlAttribute.372 # Rather than serializing into an XML child node, we instead serialize the shape to373 # an XML attribute of the *current* node.374 if value is None:375 # Don't serialize any param whose value is None.376 continue377 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 member_shape = shape.member387 if shape.serialization.get("flattened"):388 # If the list is flattened, either take the member's "name" or the name of the usual name for the parent389 # element for the children.390 element_name = self._get_serialized_name(member_shape, name)391 list_node = xmlnode392 else:393 element_name = self._get_serialized_name(member_shape, "member")394 list_node = ETree.SubElement(xmlnode, name)395 for item in params:396 self._serialize(member_shape, item, list_node, element_name)397 def _serialize_type_map(398 self, xmlnode: ETree.Element, params: dict, shape: MapShape, name: str399 ) -> None:400 """401 Given the ``name`` of MyMap, an input of {"key1": "val1", "key2": "val2"}, and the ``flattened: False``402 we serialize this as:403 <MyMap>404 <entry>405 <key>key1</key>406 <value>val1</value>407 </entry>408 <entry>409 <key>key2</key>410 <value>val2</value>411 </entry>412 </MyMap>413 If it is flattened, it is serialized as follows:414 <MyMap>415 <key>key1</key>416 <value>val1</value>417 </MyMap>418 <MyMap>419 <key>key2</key>420 <value>val2</value>421 </MyMap>422 """423 if shape.serialization.get("flattened"):424 entries_node = xmlnode425 entry_node_name = name426 else:427 entries_node = ETree.SubElement(xmlnode, name)428 entry_node_name = "entry"429 for key, value in params.items():430 entry_node = ETree.SubElement(entries_node, entry_node_name)431 key_name = self._get_serialized_name(shape.key, default_name="key")432 val_name = self._get_serialized_name(shape.value, default_name="value")433 self._serialize(shape.key, key, entry_node, key_name)434 self._serialize(shape.value, value, entry_node, val_name)435 @staticmethod436 def _serialize_type_boolean(xmlnode: ETree.Element, params: bool, _, name: str) -> None:437 """438 For scalar types, the 'params' attr is actually just a scalar value representing the data439 we need to serialize as a boolean. It will either be 'true' or 'false'440 """441 node = ETree.SubElement(xmlnode, name)442 if params:443 str_value = "true"444 else:445 str_value = "false"446 node.text = str_value447 def _serialize_type_blob(448 self, xmlnode: ETree.Element, params: Union[str, bytes], _, name: str449 ) -> None:450 node = ETree.SubElement(xmlnode, name)451 node.text = self._get_base64(params)452 def _serialize_type_timestamp(453 self, xmlnode: ETree.Element, params: str, shape: Shape, name: str454 ) -> None:455 node = ETree.SubElement(xmlnode, name)456 node.text = self._convert_timestamp_to_str(457 params, shape.serialization.get("timestampFormat")458 )459 @staticmethod460 def _default_serialize(xmlnode: ETree.Element, params: str, _, name: str) -> None:461 node = ETree.SubElement(xmlnode, name)462 node.text = six.text_type(params)463 def _prepare_additional_traits_in_xml(self, root: Optional[ETree.Element]):464 """465 Prepares the XML root node before being serialized with additional traits (like the Response ID in the Query466 protocol).467 For some protocols (like rest-xml), the root can be None.468 """469 pass470class BaseRestResponseSerializer(ResponseSerializer, ABC):471 """472 The BaseRestResponseSerializer performs the basic logic for the ReST response serialization.473 In our case it basically only adds the request metadata to the HTTP header.474 """475 def _prepare_additional_traits_in_response(476 self, response: HttpResponse, operation_model: OperationModel477 ):478 """Adds the request ID to the headers (in contrast to the body - as in the Query protocol)."""479 response = super()._prepare_additional_traits_in_response(response, operation_model)480 response["headers"]["x-amz-request-id"] = gen_amzn_requestid_long()481 return response482class RestXMLResponseSerializer(BaseRestResponseSerializer, BaseXMLResponseSerializer):483 """484 The ``RestXMLResponseSerializer`` is responsible for the serialization of responses from services with the485 ``rest-xml`` protocol.486 It combines the ``BaseRestResponseSerializer`` (for the ReST specific logic) with the ``BaseXMLResponseSerializer``487 (for the XML body response serialization), and adds some minor logic to handle S3 specific peculiarities with the488 error response serialization.489 **Experimental:** This serializer is still experimental.490 When implementing services with this serializer, some edge cases might not work out-of-the-box.491 """492 def _serialize_error(493 self,494 error: ServiceException,495 code: str,496 sender_fault: bool,497 serialized: HttpResponse,498 shape: Shape,499 operation_model: OperationModel,500 ) -> None:501 # It wouldn't be a spec if there wouldn't be any exceptions.502 # S3 errors look differently than other service's errors.503 if operation_model.name == "s3":504 attr = (505 {"xmlns": operation_model.metadata.get("xmlNamespace")}506 if "xmlNamespace" in operation_model.metadata507 else None508 )509 root = ETree.Element("Error", attr)510 self._add_error_tags(code, error, root, sender_fault)511 request_id = ETree.SubElement(root, "RequestId")512 request_id.text = gen_amzn_requestid_long()513 serialized["body"] = self._encode_payload(514 ETree.tostring(root, encoding=self.DEFAULT_ENCODING)515 )516 else:517 super()._serialize_error(error, code, sender_fault, serialized, shape, operation_model)518class QueryResponseSerializer(BaseXMLResponseSerializer):519 """520 The ``QueryResponseSerializer`` is responsible for the serialization of responses from services which use the521 ``query`` protocol. The responses of these services also use XML, but with a few subtle differences to the522 ``rest-xml`` protocol.523 **Experimental:** This serializer is still experimental.524 When implementing services with this serializer, some edge cases might not work out-of-the-box.525 """526 def _serialize_body_params_to_xml(527 self, params: dict, shape: Shape, operation_model: OperationModel528 ) -> ETree.Element:529 # The Query protocol responses have a root element which is not contained in the specification file.530 # Therefore we first call the super function to perform the normal XML serialization, and afterwards wrap the531 # result in a root element based on the operation name.532 node = super()._serialize_body_params_to_xml(params, shape, operation_model)533 # Check if we need to add a namespace534 attr = (535 {"xmlns": operation_model.metadata.get("xmlNamespace")}536 if "xmlNamespace" in operation_model.metadata537 else None538 )539 # Create the root element and add the result of the XML serializer as a child node540 root = ETree.Element(f"{operation_model.name}Response", attr)541 if node is not None:542 root.append(node)543 return root544 def _prepare_additional_traits_in_xml(self, root: Optional[ETree.Element]):545 # Add the response metadata here (it's not defined in the specs)546 # For the ec2 and the query protocol, the root cannot be None at this time.547 response_metadata = ETree.SubElement(root, "ResponseMetadata")548 request_id = ETree.SubElement(response_metadata, "RequestId")549 request_id.text = gen_amzn_requestid_long()550class EC2ResponseSerializer(QueryResponseSerializer):551 """552 The ``EC2ResponseSerializer`` is responsible for the serialization of responses from services which use the553 ``ec2`` protocol (basically the EC2 service). This protocol is basically equal to the ``query`` protocol with only554 a few subtle differences.555 **Experimental:** This serializer is still experimental.556 When implementing services with this serializer, some edge cases might not work out-of-the-box.557 """558 def _serialize_error(559 self,560 error: ServiceException,561 code: str,562 sender_fault: bool,563 serialized: HttpResponse,564 shape: Shape,565 operation_model: OperationModel,566 ) -> None:567 # EC2 errors look like:568 # <Response>569 # <Errors>570 # <Error>571 # <Code>InvalidInstanceID.Malformed</Code>572 # <Message>Invalid id: "1343124"</Message>573 # </Error>574 # </Errors>575 # <RequestID>12345</RequestID>576 # </Response>577 # This is different from QueryParser in that it's RequestID, not RequestId578 # and that the Error tag is in an enclosing Errors tag.579 attr = (580 {"xmlns": operation_model.metadata.get("xmlNamespace")}581 if "xmlNamespace" in operation_model.metadata582 else None583 )584 root = ETree.Element("Errors", attr)585 error_tag = ETree.SubElement(root, "Error")586 self._add_error_tags(code, error, error_tag, sender_fault)587 request_id = ETree.SubElement(root, "RequestID")588 request_id.text = gen_amzn_requestid_long()589 serialized["body"] = self._encode_payload(590 ETree.tostring(root, encoding=self.DEFAULT_ENCODING)591 )592 def _prepare_additional_traits_in_xml(self, root: Optional[ETree.Element]):593 # The EC2 protocol does not use the root output shape, therefore we need to remove the hierarchy level594 # below the root level595 output_node = root[0]596 for child in output_node:597 root.append(child)598 root.remove(output_node)599 # Add the requestId here (it's not defined in the specs)600 # For the ec2 and the query protocol, the root cannot be None at this time.601 request_id = ETree.SubElement(root, "requestId")602 request_id.text = gen_amzn_requestid_long()603class JSONResponseSerializer(ResponseSerializer):604 """605 The ``JSONResponseSerializer`` is responsible for the serialization of responses from services with the ``json``606 protocol. It implements the JSON response body serialization, which is also used by the...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run localstack automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful