How to use to_documentation_part_response_json method in localstack

Best Python code snippet using localstack_python

provider.py

Source:provider.py

...198        region_details = APIGatewayRegion.get()199        # This function returns either a list or a single entity (depending on the path)200        api_id = request["restApiId"]201        auth_list = region_details.documentation_parts.get(api_id) or []202        result = [to_documentation_part_response_json(api_id, a) for a in auth_list]203        result = {"item": result}204        return result205    def get_documentation_part(206        self, context: RequestContext, rest_api_id: String, documentation_part_id: String207    ) -> DocumentationPart:208        entity = find_api_subentity_by_id(rest_api_id, documentation_part_id, "documentation_parts")209        if entity is None:210            raise NotFoundException(f"Documentation part not found: {documentation_part_id}")211        return to_documentation_part_response_json(rest_api_id, entity)212    def create_documentation_part(213        self,214        context: RequestContext,215        rest_api_id: String,216        location: DocumentationPartLocation,217        properties: String,218    ) -> DocumentationPart:219        region_details = APIGatewayRegion.get()220        entity_id = short_uid()[:6]  # length 6 for AWS parity / Terraform compatibility221        entry = {222            "id": entity_id,223            "restApiId": rest_api_id,224            "location": location,225            "properties": properties,226        }227        region_details.documentation_parts.setdefault(rest_api_id, []).append(entry)228        result = to_documentation_part_response_json(rest_api_id, entry)229        return DocumentationPart(**result)230    def update_documentation_part(231        self,232        context: RequestContext,233        rest_api_id: String,234        documentation_part_id: String,235        patch_operations: ListOfPatchOperation = None,236    ) -> DocumentationPart:237        region_details = APIGatewayRegion.get()238        entity = find_api_subentity_by_id(rest_api_id, documentation_part_id, "documentation_parts")239        if entity is None:240            raise NotFoundException(f"Documentation part not found: {documentation_part_id}")241        result = apply_json_patch_safe(entity, patch_operations)242        auth_list = region_details.documentation_parts[rest_api_id]243        for i in range(len(auth_list)):244            if auth_list[i]["id"] == documentation_part_id:245                auth_list[i] = result246        result = to_documentation_part_response_json(rest_api_id, result)247        return DocumentationPart(**result)248    def delete_documentation_part(249        self, context: RequestContext, rest_api_id: String, documentation_part_id: String250    ) -> None:251        region_details = APIGatewayRegion.get()252        auth_list = region_details.documentation_parts[rest_api_id]253        for i in range(len(auth_list)):254            if auth_list[i]["id"] == documentation_part_id:255                del auth_list[i]256                break257    # base path mappings258    def get_base_path_mappings(259        self,260        context: RequestContext,261        domain_name: String,262        position: String = None,263        limit: NullableInteger = None,264    ) -> BasePathMappings:265        region_details = APIGatewayRegion.get()266        mappings_list = region_details.base_path_mappings.get(domain_name) or []267        result = [268            to_base_mapping_response_json(domain_name, m["basePath"], m) for m in mappings_list269        ]270        return BasePathMappings(items=result)271    def get_base_path_mapping(272        self, context: RequestContext, domain_name: String, base_path: String273    ) -> BasePathMapping:274        region_details = APIGatewayRegion.get()275        mappings_list = region_details.base_path_mappings.get(domain_name) or []276        mapping = ([m for m in mappings_list if m["basePath"] == base_path] or [None])[0]277        if mapping is None:278            raise NotFoundException(f"Base path mapping not found: {domain_name} - {base_path}")279        result = to_base_mapping_response_json(domain_name, base_path, mapping)280        return BasePathMapping(**result)281    def create_base_path_mapping(282        self,283        context: RequestContext,284        domain_name: String,285        rest_api_id: String,286        base_path: String = None,287        stage: String = None,288    ) -> BasePathMapping:289        region_details = APIGatewayRegion.get()290        # Note: "(none)" is a special value in API GW:291        # https://docs.aws.amazon.com/apigateway/api-reference/link-relation/basepathmapping-by-base-path292        base_path = base_path or "(none)"293        entry = {294            "domainName": domain_name,295            "restApiId": rest_api_id,296            "basePath": base_path,297            "stage": stage,298        }299        region_details.base_path_mappings.setdefault(domain_name, []).append(entry)300        result = to_base_mapping_response_json(domain_name, base_path, entry)301        return BasePathMapping(**result)302    def update_base_path_mapping(303        self,304        context: RequestContext,305        domain_name: String,306        base_path: String,307        patch_operations: ListOfPatchOperation = None,308    ) -> BasePathMapping:309        region_details = APIGatewayRegion.get()310        mappings_list = region_details.base_path_mappings.get(domain_name) or []311        mapping = ([m for m in mappings_list if m["basePath"] == base_path] or [None])[0]312        if mapping is None:313            raise NotFoundException(314                f"Not found: mapping for domain name {domain_name}, "315                f"base path {base_path} in list {mappings_list}"316            )317        patch_operations = ensure_list(patch_operations)318        for operation in patch_operations:319            if operation["path"] == "/restapiId":320                operation["path"] = "/restApiId"321        result = apply_json_patch_safe(mapping, patch_operations)322        for i in range(len(mappings_list)):323            if mappings_list[i]["basePath"] == base_path:324                mappings_list[i] = result325        result = to_base_mapping_response_json(domain_name, base_path, result)326        return BasePathMapping(**result)327    def delete_base_path_mapping(328        self, context: RequestContext, domain_name: String, base_path: String329    ) -> None:330        region_details = APIGatewayRegion.get()331        mappings_list = region_details.base_path_mappings.get(domain_name) or []332        for i in range(len(mappings_list)):333            if mappings_list[i]["basePath"] == base_path:334                del mappings_list[i]335                return336        raise NotFoundException(f"Base path mapping {base_path} for domain {domain_name} not found")337    # client certificates338    def get_client_certificate(339        self, context: RequestContext, client_certificate_id: String340    ) -> ClientCertificate:341        region_details = APIGatewayRegion.get()342        result = region_details.client_certificates.get(client_certificate_id)343        if result is None:344            raise NotFoundException(f"Client certificate ID {client_certificate_id} not found")345        return ClientCertificate(**result)346    def get_client_certificates(347        self, context: RequestContext, position: String = None, limit: NullableInteger = None348    ) -> ClientCertificates:349        region_details = APIGatewayRegion.get()350        result = list(region_details.client_certificates.values())351        return ClientCertificates(items=result)352    def generate_client_certificate(353        self, context: RequestContext, description: String = None, tags: MapOfStringToString = None354    ) -> ClientCertificate:355        region_details = APIGatewayRegion.get()356        cert_id = short_uid()357        creation_time = now_utc()358        entry = {359            "description": description,360            "tags": tags,361            "clientCertificateId": cert_id,362            "createdDate": creation_time,363            "expirationDate": creation_time + 60 * 60 * 24 * 30,  # assume 30 days validity364            "pemEncodedCertificate": "testcert-123",  # TODO return proper certificate!365        }366        region_details.client_certificates[cert_id] = entry367        result = to_client_cert_response_json(entry)368        return ClientCertificate(**result)369    def update_client_certificate(370        self,371        context: RequestContext,372        client_certificate_id: String,373        patch_operations: ListOfPatchOperation = None,374    ) -> ClientCertificate:375        region_details = APIGatewayRegion.get()376        entity = region_details.client_certificates.get(client_certificate_id)377        if entity is None:378            raise NotFoundException(f'Client certificate ID "{client_certificate_id}" not found')379        result = apply_json_patch_safe(entity, patch_operations)380        result = to_client_cert_response_json(result)381        return ClientCertificate(**result)382    def delete_client_certificate(383        self, context: RequestContext, client_certificate_id: String384    ) -> None:385        region_details = APIGatewayRegion.get()386        entity = region_details.client_certificates.pop(client_certificate_id, None)387        if entity is None:388            raise NotFoundException(f'VPC link ID "{client_certificate_id}" not found for deletion')389    # VPC links390    def create_vpc_link(391        self,392        context: RequestContext,393        name: String,394        target_arns: ListOfString,395        description: String = None,396        tags: MapOfStringToString = None,397    ) -> VpcLink:398        region_details = APIGatewayRegion.get()399        link_id = short_uid()400        entry = {"id": link_id, "status": "AVAILABLE"}401        region_details.vpc_links[link_id] = entry402        result = to_vpc_link_response_json(entry)403        return VpcLink(**result)404    def get_vpc_links(405        self, context: RequestContext, position: String = None, limit: NullableInteger = None406    ) -> VpcLinks:407        region_details = APIGatewayRegion.get()408        result = region_details.vpc_links.values()409        result = [to_vpc_link_response_json(r) for r in result]410        result = {"items": result}411        return result412    def get_vpc_link(self, context: RequestContext, vpc_link_id: String) -> VpcLink:413        region_details = APIGatewayRegion.get()414        vpc_link = region_details.vpc_links.get(vpc_link_id)415        if vpc_link is None:416            raise NotFoundException(f'VPC link ID "{vpc_link_id}" not found')417        result = to_vpc_link_response_json(vpc_link)418        return VpcLink(**result)419    def update_vpc_link(420        self,421        context: RequestContext,422        vpc_link_id: String,423        patch_operations: ListOfPatchOperation = None,424    ) -> VpcLink:425        region_details = APIGatewayRegion.get()426        vpc_link = region_details.vpc_links.get(vpc_link_id)427        if vpc_link is None:428            raise NotFoundException(f'VPC link ID "{vpc_link_id}" not found')429        result = apply_json_patch_safe(vpc_link, patch_operations)430        result = to_vpc_link_response_json(result)431        return VpcLink(**result)432    def delete_vpc_link(self, context: RequestContext, vpc_link_id: String) -> None:433        region_details = APIGatewayRegion.get()434        vpc_link = region_details.vpc_links.pop(vpc_link_id, None)435        if vpc_link is None:436            raise NotFoundException(f'VPC link ID "{vpc_link_id}" not found for deletion')437    # request validators438    def get_request_validators(439        self,440        context: RequestContext,441        rest_api_id: String,442        position: String = None,443        limit: NullableInteger = None,444    ) -> RequestValidators:445        region_details = APIGatewayRegion.get()446        auth_list = region_details.validators.get(rest_api_id) or []447        result = [to_validator_response_json(rest_api_id, a) for a in auth_list]448        return RequestValidators(items=result)449    def get_request_validator(450        self, context: RequestContext, rest_api_id: String, request_validator_id: String451    ) -> RequestValidator:452        region_details = APIGatewayRegion.get()453        auth_list = region_details.validators.get(rest_api_id) or []454        validator = ([a for a in auth_list if a["id"] == request_validator_id] or [None])[0]455        if validator is None:456            raise NotFoundException(457                f"Validator {request_validator_id} for API Gateway {rest_api_id} not found"458            )459        result = to_validator_response_json(rest_api_id, validator)460        return RequestValidator(**result)461    def create_request_validator(462        self,463        context: RequestContext,464        rest_api_id: String,465        name: String = None,466        validate_request_body: Boolean = None,467        validate_request_parameters: Boolean = None,468    ) -> RequestValidator:469        region_details = APIGatewayRegion.get()470        # length 6 for AWS parity and TF compatibility471        validator_id = short_uid()[:6]472        entry = {473            "id": validator_id,474            "name": name,475            "restApiId": rest_api_id,476            "validateRequestBody": validate_request_body,477            "validateRequestPparameters": validate_request_parameters,478        }479        region_details.validators.setdefault(rest_api_id, []).append(entry)480        return RequestValidator(**entry)481    def update_request_validator(482        self,483        context: RequestContext,484        rest_api_id: String,485        request_validator_id: String,486        patch_operations: ListOfPatchOperation = None,487    ) -> RequestValidator:488        region_details = APIGatewayRegion.get()489        auth_list = region_details.validators.get(rest_api_id) or []490        validator = ([a for a in auth_list if a["id"] == request_validator_id] or [None])[0]491        if validator is None:492            raise NotFoundException(493                f"Validator {request_validator_id} for API Gateway {rest_api_id} not found"494            )495        result = apply_json_patch_safe(validator, patch_operations)496        entry_list = region_details.validators[rest_api_id]497        for i in range(len(entry_list)):498            if entry_list[i]["id"] == request_validator_id:499                entry_list[i] = result500        result = to_validator_response_json(rest_api_id, result)501        return RequestValidator(**result)502    def delete_request_validator(503        self, context: RequestContext, rest_api_id: String, request_validator_id: String504    ) -> None:505        region_details = APIGatewayRegion.get()506        auth_list = region_details.validators.get(rest_api_id, [])507        for i in range(len(auth_list)):508            if auth_list[i]["id"] == request_validator_id:509                del auth_list[i]510                return511        raise NotFoundException(512            f"Validator {request_validator_id} for API Gateway {rest_api_id} not found"513        )514    # tags515    def get_tags(516        self,517        context: RequestContext,518        resource_arn: String,519        position: String = None,520        limit: NullableInteger = None,521    ) -> Tags:522        result = APIGatewayRegion.TAGS.get(resource_arn, {})523        return Tags(tags=result)524    def tag_resource(525        self, context: RequestContext, resource_arn: String, tags: MapOfStringToString526    ) -> None:527        resource_tags = APIGatewayRegion.TAGS.setdefault(resource_arn, {})528        resource_tags.update(tags)529    def untag_resource(530        self, context: RequestContext, resource_arn: String, tag_keys: ListOfString531    ) -> None:532        resource_tags = APIGatewayRegion.TAGS.setdefault(resource_arn, {})533        for key in tag_keys:534            resource_tags.pop(key, None)535# ---------------536# UTIL FUNCTIONS537# ---------------538def normalize_authorizer(data):539    is_list = isinstance(data, list)540    entries = ensure_list(data)541    for i in range(len(entries)):542        entry = deepcopy(entries[i])543        # terraform sends this as a string in patch, so convert to int544        entry["authorizerResultTtlInSeconds"] = int(entry.get("authorizerResultTtlInSeconds", 300))545        entries[i] = entry546    return entries if is_list else entries[0]547def to_authorizer_response_json(api_id, data):548    return to_response_json("authorizer", data, api_id=api_id)549def to_validator_response_json(api_id, data):550    return to_response_json("validator", data, api_id=api_id)551def to_documentation_part_response_json(api_id, data):552    return to_response_json("documentationpart", data, api_id=api_id)553def to_base_mapping_response_json(domain_name, base_path, data):554    self_link = "/domainnames/%s/basepathmappings/%s" % (domain_name, base_path)555    return to_response_json("basepathmapping", data, self_link=self_link)556def to_account_response_json(data):557    return to_response_json("account", data, self_link="/account")558def to_vpc_link_response_json(data):559    return to_response_json("vpclink", data)560def to_client_cert_response_json(data):561    return to_response_json("clientcertificate", data, id_attr="clientCertificateId")562def to_response_json(model_type, data, api_id=None, self_link=None, id_attr=None):563    if isinstance(data, list) and len(data) == 1:564        data = data[0]565    id_attr = id_attr or "id"...

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.