How to use _partition_members method in localstack

Best Python code snippet using localstack_python

serializer.py

Source:serializer.py Github

copy

Full Screen

...545 shape: Optional[Shape],546 shape_members: dict,547 operation_model: OperationModel,548 ) -> None:549 header_params, payload_params = self._partition_members(parameters, shape)550 self._process_header_members(header_params, response, shape)551 # "HEAD" responeses are basically "GET" responses without the actual body.552 # Do not process the body payload in this case (setting a body could also manipulate the headers)553 if operation_model.http.get("method") != "HEAD":554 self._serialize_payload(payload_params, response, shape, shape_members, operation_model)555 self._serialize_content_type(response, shape, shape_members)556 self._prepare_additional_traits_in_response(response, operation_model)557 def _serialize_payload(558 self,559 parameters: dict,560 response: HttpResponse,561 shape: Optional[Shape],562 shape_members: dict,563 operation_model: OperationModel,564 ) -> None:565 """566 Serializes the given payload.567 :param parameters: The user input params568 :param response: The final serialized HttpResponse569 :param shape: Describes the expected output shape (can be None in case of an "empty" response)570 :param shape_members: The members of the output struct shape571 :param operation_model: The specification of the operation of which the response is serialized here572 :return: None - the given `serialized` dict is modified573 """574 if shape is None:575 return576 payload_member = shape.serialization.get("payload")577 if payload_member is not None and shape_members[payload_member].type_name in [578 "blob",579 "string",580 ]:581 # If it's streaming, then the body is just the value of the payload.582 body_payload = parameters.get(payload_member, b"")583 body_payload = self._encode_payload(body_payload)584 response.data = body_payload585 elif payload_member is not None:586 # If there's a payload member, we serialized that member to the body.587 body_params = parameters.get(payload_member)588 if body_params is not None:589 response.data = self._encode_payload(590 self._serialize_body_params(591 body_params, shape_members[payload_member], operation_model592 )593 )594 else:595 # Otherwise, we use the "traditional" way of serializing the whole parameters dict recursively.596 response.data = self._encode_payload(597 self._serialize_body_params(parameters, shape, operation_model)598 )599 def _serialize_content_type(self, serialized: HttpResponse, shape: Shape, shape_members: dict):600 """601 Some protocols require varied Content-Type headers602 depending on user input. This allows subclasses to apply603 this conditionally.604 """605 pass606 def _has_streaming_payload(self, payload: Optional[str], shape_members):607 """Determine if payload is streaming (a blob or string)."""608 return payload is not None and shape_members[payload].type_name in ["blob", "string"]609 def _prepare_additional_traits_in_response(610 self, response: HttpResponse, operation_model: OperationModel611 ):612 """Adds the request ID to the headers (in contrast to the body - as in the Query protocol)."""613 response = super()._prepare_additional_traits_in_response(response, operation_model)614 response.headers["x-amz-request-id"] = gen_amzn_requestid_long()615 return response616 def _process_header_members(self, parameters: dict, response: HttpResponse, shape: Shape):617 shape_members = shape.members if isinstance(shape, StructureShape) else []618 for name in shape_members:619 member_shape = shape_members[name]620 location = member_shape.serialization.get("location")621 if not location:622 continue623 if name not in parameters:624 # ignores optional keys625 continue626 key = member_shape.serialization.get("name", name)627 value = parameters[name]628 if value is None:629 continue630 if location == "header":631 response.headers[key] = self._serialize_header_value(member_shape, value)632 elif location == "headers":633 header_prefix = key634 self._serialize_header_map(header_prefix, response, value)635 elif location == "statusCode":636 response.status_code = int(value)637 def _serialize_header_map(self, prefix: str, response: HttpResponse, params: dict) -> None:638 """Serializes the header map for the location trait "headers"."""639 for key, val in params.items():640 actual_key = prefix + key641 response.headers[actual_key] = val642 def _serialize_header_value(self, shape: Shape, value: Any):643 """Serializes a value for the location trait "header"."""644 if shape.type_name == "timestamp":645 datetime_obj = parse_to_aware_datetime(value)646 timestamp_format = shape.serialization.get(647 "timestampFormat", self.HEADER_TIMESTAMP_FORMAT648 )649 return self._convert_timestamp_to_str(datetime_obj, timestamp_format)650 elif shape.type_name == "list":651 converted_value = [652 self._serialize_header_value(shape.member, v) for v in value if v is not None653 ]654 return ",".join(converted_value)655 elif shape.type_name == "boolean":656 # Set the header value to "true" if the given value is truthy, otherwise set the header value to "false".657 return "true" if value else "false"658 elif is_json_value_header(shape):659 # Serialize with no spaces after separators to save space in660 # the header.661 return self._get_base64(json.dumps(value, separators=(",", ":")))662 else:663 return value664 def _partition_members(self, parameters: dict, shape: Optional[Shape]) -> Tuple[dict, dict]:665 """Separates the top-level keys in the given parameters dict into header- and payload-located params."""666 if not isinstance(shape, StructureShape):667 # If the shape isn't a structure, we default to the whole response being parsed in the body.668 # Non-payload members are only loated in the top-level hierarchy and those are always structures.669 return {}, parameters670 header_params = {}671 payload_params = {}672 shape_members = shape.members673 for name in shape_members:674 member_shape = shape_members[name]675 if name not in parameters:676 continue677 location = member_shape.serialization.get("location")678 if location:...

Full Screen

Full Screen

terminal.py

Source:terminal.py Github

copy

Full Screen

...408 # Summarize contents.409 obj_dict = get_dict(objdoc, cfg["private"]) or {}410 if not cfg["imports"]:411 obj_dict = { n: o for n, o in obj_dict.items() if not is_ref(o) }412 partitions = _partition_members(obj_dict)413 partition = partitions.pop("modules", {})414 if len(partition) > 0:415 header("Modules")416 _print_members(inspector, partition, path, pr, False, imports=cfg["imports"])417 partition = partitions.pop("types", {})418 if len(partition) > 0:419 header("Types" if type_name == "module" else "Member Types")420 _print_members(inspector, partition, path, pr, False, imports=cfg["imports"])421 partition = partitions.pop("properties", {})422 if len(partition) > 0:423 header("Properties")424 _print_members(inspector, partition, path, pr, True, imports=cfg["imports"])425 partition = partitions.pop("functions", {})426 if len(partition) > 0:427 header("Functions" if type_name == "module" else "Methods")428 _print_members(inspector, partition, path, pr, True, imports=cfg["imports"])429 partition = partitions.pop("attributes", {})430 if len(partition) > 0:431 header("Attributes")432 _print_members(inspector, partition, path, pr, True, imports=cfg["imports"])433 assert len(partitions) == 0434# FIXME: Use paths.435_PARTITIONS = {436 "builtins.builtin_function_or_method" : "functions",437 "builtins.classmethod" : "functions",438 "builtins.classmethod_descriptor" : "functions",439 "builtins.function" : "functions",440 "builtins.method_descriptor" : "functions",441 "builtins.module" : "modules",442 "builtins.property" : "properties",443 "builtins.staticmethod" : "functions",444 "builtins.type" : "types",445 "builtins.wrapper_descriptor" : "functions",446 "_ctypes.PyCStructType" : "types",447}448def _partition_members(dict):449 partitions = {}450 for name, objdoc in dict.items():451 type = objdoc.get("type")452 if type is None:453 # Missing type...?454 partition_name = "attributes"455 else:456 type_path = ".".join(get_path(objdoc["type"]))457 partition_name = _PARTITIONS.get(str(type_path), "attributes")458 partitions.setdefault(partition_name, {})[name] = objdoc459 return partitions460 461def repr_is_uninteresting(objdoc):462 # FIXME...

Full Screen

Full Screen

gen.py

Source:gen.py Github

copy

Full Screen

...425 # Summarize contents.426 # FIXME427 private = True428 imports = True429 partitions = terminal._partition_members(terminal.get_dict(objdoc, private) or {})430 contents = doc << DIV(cls="contents")431 controls = contents << DIV(cls="controls")432 controls << BUTTON("Imported", id="cb-import", cls="toggle")433 # FIXME: Put this somewhere reasonable.434 controls << SCRIPT("""435 $(function () {436 $('.imported-name').toggle(false);437 // FIXME: This animation is cheesy.438 $('#cb-import').click(function (event) {439 $('.imported-name').toggle('fast');440 $('#cb-import').toggleClass('toggled');441 });442 });443 """)...

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