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

How to use _partition_members method in localstack

Best Python code snippet using localstack_python

serializer.py

Source:serializer.py

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

terminal.py

Source:terminal.py

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

gen.py

Source:gen.py

...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    """)...

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.