import json from typing import Any, Dict, List from urllib.parse import unquote from moto.core.responses import TYPE_RESPONSE, BaseResponse from moto.utilities.utils import merge_multiple_dicts from .exceptions import InvalidRequestInput from .models import APIGatewayBackend, apigateway_backends from .utils import deserialize_body API_KEY_SOURCES = ["AUTHORIZER", "HEADER"] AUTHORIZER_TYPES = ["TOKEN", "REQUEST", "COGNITO_USER_POOLS"] ENDPOINT_CONFIGURATION_TYPES = ["PRIVATE", "EDGE", "REGIONAL"] class APIGatewayResponse(BaseResponse): def __init__(self) -> None: super().__init__(service_name="apigateway") def error(self, type_: str, message: str, status: int = 400) -> TYPE_RESPONSE: headers = self.response_headers or {} headers["X-Amzn-Errortype"] = type_ return (status, headers, json.dumps({"__type": type_, "message": message})) @property def backend(self) -> APIGatewayBackend: return apigateway_backends[self.current_account][self.region] def __validate_api_key_source(self, api_key_source: str) -> TYPE_RESPONSE: # type: ignore[return] if api_key_source and api_key_source not in API_KEY_SOURCES: return self.error( "ValidationException", ( "1 validation error detected: " "Value '{api_key_source}' at 'createRestApiInput.apiKeySource' failed " "to satisfy constraint: Member must satisfy enum value set: " "[AUTHORIZER, HEADER]" ).format(api_key_source=api_key_source), ) def __validate_endpoint_configuration( # type: ignore[return] self, endpoint_configuration: Dict[str, str] ) -> TYPE_RESPONSE: if endpoint_configuration and "types" in endpoint_configuration: invalid_types = list( set(endpoint_configuration["types"]) - set(ENDPOINT_CONFIGURATION_TYPES) ) if invalid_types: return self.error( "ValidationException", ( "1 validation error detected: Value '{endpoint_type}' " "at 'createRestApiInput.endpointConfiguration.types' failed " "to satisfy constraint: Member must satisfy enum value set: " "[PRIVATE, EDGE, REGIONAL]" ).format(endpoint_type=invalid_types[0]), ) def restapis( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) if self.method == "GET": apis = self.backend.list_apis() return 200, {}, json.dumps({"item": [api.to_dict() for api in apis]}) elif self.method == "POST": api_doc = deserialize_body(self.body) if api_doc: fail_on_warnings = self._get_bool_param("failonwarnings") or False rest_api = self.backend.import_rest_api(api_doc, fail_on_warnings) return 200, {}, json.dumps(rest_api.to_dict()) name = self._get_param("name") description = self._get_param("description") api_key_source = self._get_param("apiKeySource") endpoint_configuration = self._get_param("endpointConfiguration") tags = self._get_param("tags") policy = self._get_param("policy") minimum_compression_size = self._get_param("minimumCompressionSize") disable_execute_api_endpoint = self._get_param("disableExecuteApiEndpoint") # Param validation response = self.__validate_api_key_source(api_key_source) if response is not None: return response response = self.__validate_endpoint_configuration(endpoint_configuration) if response is not None: return response rest_api = self.backend.create_rest_api( name, description, api_key_source=api_key_source, endpoint_configuration=endpoint_configuration, tags=tags, policy=policy, minimum_compression_size=minimum_compression_size, disable_execute_api_endpoint=disable_execute_api_endpoint, ) return 200, {}, json.dumps(rest_api.to_dict()) def __validte_rest_patch_operations( # type: ignore[return] self, patch_operations: List[Dict[str, str]] ) -> TYPE_RESPONSE: for op in patch_operations: path = op["path"] if "apiKeySource" in path: value = op["value"] return self.__validate_api_key_source(value) def restapis_individual( self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) function_id = self.path.replace("/restapis/", "", 1).split("/")[0] if self.method == "GET": rest_api = self.backend.get_rest_api(function_id) elif self.method == "DELETE": rest_api = self.backend.delete_rest_api(function_id) elif self.method == "PUT": mode = self._get_param("mode", "merge") fail_on_warnings = self._get_bool_param("failonwarnings") or False api_doc = deserialize_body(self.body) rest_api = self.backend.put_rest_api( function_id, api_doc, mode=mode, fail_on_warnings=fail_on_warnings ) elif self.method == "PATCH": patch_operations = self._get_param("patchOperations") response = self.__validte_rest_patch_operations(patch_operations) if response is not None: return response rest_api = self.backend.update_rest_api(function_id, patch_operations) return 200, {}, json.dumps(rest_api.to_dict()) def resources( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) function_id = self.path.replace("/restapis/", "", 1).split("/")[0] if self.method == "GET": resources = self.backend.get_resources(function_id) return ( 200, {}, json.dumps({"item": [resource.to_dict() for resource in resources]}), ) def gateway_response( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) if request.method == "PUT": return self.put_gateway_response() elif request.method == "GET": return self.get_gateway_response() elif request.method == "DELETE": return self.delete_gateway_response() def gateway_responses( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) if request.method == "GET": return self.get_gateway_responses() def resource_individual( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) function_id = self.path.replace("/restapis/", "", 1).split("/")[0] resource_id = self.path.split("/")[-1] if self.method == "GET": resource = self.backend.get_resource(function_id, resource_id) return 200, {}, json.dumps(resource.to_dict()) elif self.method == "POST": path_part = self._get_param("pathPart") resource = self.backend.create_resource(function_id, resource_id, path_part) return 201, {}, json.dumps(resource.to_dict()) elif self.method == "DELETE": resource = self.backend.delete_resource(function_id, resource_id) return 202, {}, json.dumps(resource.to_dict()) def resource_methods( self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") function_id = url_path_parts[2] resource_id = url_path_parts[4] method_type = url_path_parts[6] if self.method == "GET": method = self.backend.get_method(function_id, resource_id, method_type) return 200, {}, json.dumps(method.to_json()) elif self.method == "PUT": authorization_type = self._get_param("authorizationType") api_key_required = self._get_param("apiKeyRequired") request_models = self._get_param("requestModels") operation_name = self._get_param("operationName") authorizer_id = self._get_param("authorizerId") authorization_scopes = self._get_param("authorizationScopes") request_validator_id = self._get_param("requestValidatorId") request_parameters = self._get_param("requestParameters") method = self.backend.put_method( function_id, resource_id, method_type, authorization_type, api_key_required, request_models=request_models, request_parameters=request_parameters, operation_name=operation_name, authorizer_id=authorizer_id, authorization_scopes=authorization_scopes, request_validator_id=request_validator_id, ) return 201, {}, json.dumps(method.to_json()) elif self.method == "DELETE": self.backend.delete_method(function_id, resource_id, method_type) return 204, {}, "" return 200, {}, "" def resource_method_responses( self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") function_id = url_path_parts[2] resource_id = url_path_parts[4] method_type = url_path_parts[6] response_code = url_path_parts[8] if self.method == "GET": method_response = self.backend.get_method_response( function_id, resource_id, method_type, response_code ) return 200, {}, json.dumps(method_response.to_json()) # type: ignore[union-attr] elif self.method == "PUT": response_models = self._get_param("responseModels") response_parameters = self._get_param("responseParameters") method_response = self.backend.put_method_response( function_id, resource_id, method_type, response_code, response_models, response_parameters, ) return 201, {}, json.dumps(method_response.to_json()) elif self.method == "DELETE": method_response = self.backend.delete_method_response( function_id, resource_id, method_type, response_code ) return 204, {}, json.dumps(method_response.to_json()) # type: ignore[union-attr] raise Exception(f'Unexpected HTTP method "{self.method}"') def restapis_authorizers( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") restapi_id = url_path_parts[2] if self.method == "POST": name = self._get_param("name") authorizer_type = self._get_param("type") provider_arns = self._get_param("providerARNs") auth_type = self._get_param("authType") authorizer_uri = self._get_param("authorizerUri") authorizer_credentials = self._get_param("authorizerCredentials") identity_source = self._get_param("identitySource") identiy_validation_expression = self._get_param( "identityValidationExpression" ) authorizer_result_ttl = self._get_param( "authorizerResultTtlInSeconds", if_none=300 ) # Param validation if authorizer_type and authorizer_type not in AUTHORIZER_TYPES: return self.error( "ValidationException", ( "1 validation error detected: " "Value '{authorizer_type}' at 'createAuthorizerInput.type' failed " "to satisfy constraint: Member must satisfy enum value set: " "[TOKEN, REQUEST, COGNITO_USER_POOLS]" ).format(authorizer_type=authorizer_type), ) authorizer_response = self.backend.create_authorizer( restapi_id=restapi_id, name=name, authorizer_type=authorizer_type, provider_arns=provider_arns, auth_type=auth_type, authorizer_uri=authorizer_uri, authorizer_credentials=authorizer_credentials, identity_source=identity_source, identiy_validation_expression=identiy_validation_expression, authorizer_result_ttl=authorizer_result_ttl, ) return 201, {}, json.dumps(authorizer_response.to_json()) elif self.method == "GET": authorizers = self.backend.get_authorizers(restapi_id) return 200, {}, json.dumps({"item": [a.to_json() for a in authorizers]}) def request_validators( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") restapi_id = url_path_parts[2] if self.method == "GET": validators = self.backend.get_request_validators(restapi_id) res = json.dumps( {"item": [validator.to_dict() for validator in validators]} ) return 200, {}, res if self.method == "POST": name = self._get_param("name") body = self._get_bool_param("validateRequestBody") params = self._get_bool_param("validateRequestParameters") validator = self.backend.create_request_validator( restapi_id, name, body, params ) return 201, {}, json.dumps(validator.to_dict()) def request_validator_individual( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") restapi_id = url_path_parts[2] validator_id = url_path_parts[4] if self.method == "GET": validator = self.backend.get_request_validator(restapi_id, validator_id) return 200, {}, json.dumps(validator.to_dict()) if self.method == "DELETE": self.backend.delete_request_validator(restapi_id, validator_id) return 202, {}, "" if self.method == "PATCH": patch_operations = self._get_param("patchOperations") validator = self.backend.update_request_validator( restapi_id, validator_id, patch_operations ) return 200, {}, json.dumps(validator.to_dict()) def authorizers( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") restapi_id = url_path_parts[2] authorizer_id = url_path_parts[4] if self.method == "GET": authorizer_response = self.backend.get_authorizer(restapi_id, authorizer_id) return 200, {}, json.dumps(authorizer_response.to_json()) elif self.method == "PATCH": patch_operations = self._get_param("patchOperations") authorizer_response = self.backend.update_authorizer( restapi_id, authorizer_id, patch_operations ) return 200, {}, json.dumps(authorizer_response.to_json()) elif self.method == "DELETE": self.backend.delete_authorizer(restapi_id, authorizer_id) return 202, {}, "{}" def restapis_stages( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") function_id = url_path_parts[2] if self.method == "POST": stage_name = self._get_param("stageName") deployment_id = self._get_param("deploymentId") stage_variables = self._get_param("variables", if_none={}) description = self._get_param("description", if_none="") cacheClusterEnabled = self._get_param("cacheClusterEnabled", if_none=False) cacheClusterSize = self._get_param("cacheClusterSize") tags = self._get_param("tags") tracing_enabled = self._get_param("tracingEnabled") stage_response = self.backend.create_stage( function_id, stage_name, deployment_id, variables=stage_variables, description=description, cacheClusterEnabled=cacheClusterEnabled, cacheClusterSize=cacheClusterSize, tags=tags, tracing_enabled=tracing_enabled, ) return 201, {}, json.dumps(stage_response.to_json()) elif self.method == "GET": stages = self.backend.get_stages(function_id) return 200, {}, json.dumps({"item": [s.to_json() for s in stages]}) def restapis_stages_tags( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") function_id = url_path_parts[4] stage_name = url_path_parts[6] if self.method == "PUT": tags = self._get_param("tags") if tags: stage = self.backend.get_stage(function_id, stage_name) stage.tags = merge_multiple_dicts(stage.tags or {}, tags) return 200, {}, json.dumps({"item": tags}) if self.method == "DELETE": stage = self.backend.get_stage(function_id, stage_name) for tag in (stage.tags or {}).copy(): if tag in (self.querystring.get("tagKeys") or {}): stage.tags.pop(tag, None) # type: ignore[union-attr] return 200, {}, json.dumps({"item": ""}) def stages( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") function_id = url_path_parts[2] stage_name = url_path_parts[4] if self.method == "GET": stage_response = self.backend.get_stage(function_id, stage_name) return 200, {}, json.dumps(stage_response.to_json()) elif self.method == "PATCH": patch_operations = self._get_param("patchOperations") stage_response = self.backend.update_stage( function_id, stage_name, patch_operations ) return 200, {}, json.dumps(stage_response.to_json()) elif self.method == "DELETE": self.backend.delete_stage(function_id, stage_name) return 202, {}, "{}" def export( self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") rest_api_id = url_path_parts[-5] export_type = url_path_parts[-1] body = self.backend.export_api(rest_api_id, export_type) now = body["info"]["version"] filename = f"swagger_{now}Z.json" headers = { "Content-Type": "application/octet-stream", "Content-Disposition": f'attachment; filename="{filename}"', } return 200, headers, json.dumps(body).encode("utf-8") def integrations( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") function_id = url_path_parts[2] resource_id = url_path_parts[4] method_type = url_path_parts[6] if self.method == "GET": integration_response = self.backend.get_integration( function_id, resource_id, method_type ) if integration_response: return 200, {}, json.dumps(integration_response.to_json()) return 200, {}, "{}" elif self.method == "PUT": integration_type = self._get_param("type") uri = self._get_param("uri") credentials = self._get_param("credentials") request_templates = self._get_param("requestTemplates") passthrough_behavior = self._get_param("passthroughBehavior") tls_config = self._get_param("tlsConfig") cache_namespace = self._get_param("cacheNamespace") timeout_in_millis = self._get_param("timeoutInMillis") request_parameters = self._get_param("requestParameters") content_handling = self._get_param("contentHandling") connection_type = self._get_param("connectionType") self.backend.get_method(function_id, resource_id, method_type) integration_http_method = self._get_param( "httpMethod" ) # default removed because it's a required parameter integration_response = self.backend.put_integration( function_id, resource_id, method_type, integration_type, uri, credentials=credentials, integration_method=integration_http_method, request_templates=request_templates, passthrough_behavior=passthrough_behavior, tls_config=tls_config, cache_namespace=cache_namespace, timeout_in_millis=timeout_in_millis, request_parameters=request_parameters, content_handling=content_handling, connection_type=connection_type, ) return 201, {}, json.dumps(integration_response.to_json()) elif self.method == "DELETE": integration_response = self.backend.delete_integration( function_id, resource_id, method_type ) return 204, {}, json.dumps(integration_response.to_json()) def integration_responses( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") function_id = url_path_parts[2] resource_id = url_path_parts[4] method_type = url_path_parts[6] status_code = url_path_parts[9] if self.method == "GET": integration_response = self.backend.get_integration_response( function_id, resource_id, method_type, status_code ) return 200, {}, json.dumps(integration_response.to_json()) elif self.method == "PUT": if not self.body: raise InvalidRequestInput() selection_pattern = self._get_param("selectionPattern") response_templates = self._get_param("responseTemplates") response_parameters = self._get_param("responseParameters") content_handling = self._get_param("contentHandling") integration_response = self.backend.put_integration_response( function_id, resource_id, method_type, status_code, selection_pattern, response_templates, response_parameters, content_handling, ) return 201, {}, json.dumps(integration_response.to_json()) elif self.method == "DELETE": integration_response = self.backend.delete_integration_response( function_id, resource_id, method_type, status_code ) return 204, {}, json.dumps(integration_response.to_json()) def deployments( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) function_id = self.path.replace("/restapis/", "", 1).split("/")[0] if self.method == "GET": deployments = self.backend.get_deployments(function_id) return 200, {}, json.dumps({"item": [d.to_json() for d in deployments]}) elif self.method == "POST": name = self._get_param("stageName") description = self._get_param("description") stage_variables = self._get_param("variables", if_none={}) deployment = self.backend.create_deployment( function_id, name, description, stage_variables ) return 201, {}, json.dumps(deployment.to_json()) def individual_deployment( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") function_id = url_path_parts[2] deployment_id = url_path_parts[4] if self.method == "GET": deployment = self.backend.get_deployment(function_id, deployment_id) return 200, {}, json.dumps(deployment.to_json()) elif self.method == "DELETE": deployment = self.backend.delete_deployment(function_id, deployment_id) return 202, {}, json.dumps(deployment.to_json()) def apikeys( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) if self.method == "POST": apikey_response = self.backend.create_api_key(json.loads(self.body)) return 201, {}, json.dumps(apikey_response.to_json()) elif self.method == "GET": include_values = self._get_bool_param("includeValues") or False apikeys_response = self.backend.get_api_keys() resp = [a.to_json() for a in apikeys_response] if not include_values: for key in resp: key.pop("value") return 200, {}, json.dumps({"item": resp}) def apikey_individual( self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") apikey = url_path_parts[2] if self.method == "GET": include_value = self._get_bool_param("includeValue") or False apikey_resp = self.backend.get_api_key(apikey).to_json() if not include_value: apikey_resp.pop("value") elif self.method == "PATCH": patch_operations = self._get_param("patchOperations") apikey_resp = self.backend.update_api_key( apikey, patch_operations ).to_json() elif self.method == "DELETE": self.backend.delete_api_key(apikey) return 202, {}, "{}" return 200, {}, json.dumps(apikey_resp) def usage_plans( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) if self.method == "POST": usage_plan_response = self.backend.create_usage_plan(json.loads(self.body)) return 201, {}, json.dumps(usage_plan_response.to_json()) elif self.method == "GET": api_key_id = self.querystring.get("keyId", [None])[0] usage_plans_response = self.backend.get_usage_plans(api_key_id=api_key_id) return ( 200, {}, json.dumps({"item": [u.to_json() for u in usage_plans_response]}), ) def usage_plan_individual( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") usage_plan = url_path_parts[2] if self.method == "GET": usage_plan_response = self.backend.get_usage_plan(usage_plan) return 200, {}, json.dumps(usage_plan_response.to_json()) elif self.method == "DELETE": self.backend.delete_usage_plan(usage_plan) return 202, {}, "{}" elif self.method == "PATCH": patch_operations = self._get_param("patchOperations") usage_plan_response = self.backend.update_usage_plan( usage_plan, patch_operations ) return 200, {}, json.dumps(usage_plan_response.to_json()) def usage_plan_keys( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") usage_plan_id = url_path_parts[2] if self.method == "POST": usage_plan_response = self.backend.create_usage_plan_key( usage_plan_id, json.loads(self.body) ) return 201, {}, json.dumps(usage_plan_response.to_json()) elif self.method == "GET": usage_plans_response = self.backend.get_usage_plan_keys(usage_plan_id) return ( 200, {}, json.dumps({"item": [u.to_json() for u in usage_plans_response]}), ) def usage_plan_key_individual( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") usage_plan_id = url_path_parts[2] key_id = url_path_parts[4] if self.method == "GET": usage_plan_response = self.backend.get_usage_plan_key(usage_plan_id, key_id) return 200, {}, json.dumps(usage_plan_response.to_json()) elif self.method == "DELETE": self.backend.delete_usage_plan_key(usage_plan_id, key_id) return 202, {}, "{}" def domain_names( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) if self.method == "GET": domain_names = self.backend.get_domain_names() return 200, {}, json.dumps({"item": [d.to_json() for d in domain_names]}) elif self.method == "POST": domain_name = self._get_param("domainName") certificate_name = self._get_param("certificateName") tags = self._get_param("tags") certificate_arn = self._get_param("certificateArn") certificate_body = self._get_param("certificateBody") certificate_private_key = self._get_param("certificatePrivateKey") certificate_chain = self._get_param("certificateChain") regional_certificate_name = self._get_param("regionalCertificateName") regional_certificate_arn = self._get_param("regionalCertificateArn") endpoint_configuration = self._get_param("endpointConfiguration") security_policy = self._get_param("securityPolicy") domain_name_resp = self.backend.create_domain_name( domain_name, certificate_name, tags, certificate_arn, certificate_body, certificate_private_key, certificate_chain, regional_certificate_name, regional_certificate_arn, endpoint_configuration, security_policy, ) return 201, {}, json.dumps(domain_name_resp.to_json()) def domain_name_induvidual( self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") domain_name = url_path_parts[2] if self.method == "GET": if domain_name is not None: domain_names = self.backend.get_domain_name(domain_name) return 200, {}, json.dumps(domain_names.to_json()) return 200, {}, "{}" elif self.method == "DELETE": if domain_name is not None: self.backend.delete_domain_name(domain_name) return 202, {}, json.dumps({}) else: msg = f'Method "{self.method}" for API GW domain names not implemented' return 404, {}, json.dumps({"error": msg}) def models( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) rest_api_id = self.path.replace("/restapis/", "", 1).split("/")[0] if self.method == "GET": models = self.backend.get_models(rest_api_id) return 200, {}, json.dumps({"item": [m.to_json() for m in models]}) elif self.method == "POST": name = self._get_param("name") description = self._get_param("description") schema = self._get_param("schema") content_type = self._get_param("contentType") model = self.backend.create_model( rest_api_id, name, content_type, description, schema, ) return 201, {}, json.dumps(model.to_json()) def model_induvidual( self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") rest_api_id = url_path_parts[2] model_name = url_path_parts[4] if self.method == "GET": model_info = self.backend.get_model(rest_api_id, model_name) return 200, {}, json.dumps(model_info.to_json()) return 200, {}, "{}" def base_path_mappings( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") domain_name = url_path_parts[2] if self.method == "GET": base_path_mappings = self.backend.get_base_path_mappings(domain_name) return ( 200, {}, json.dumps({"item": [m.to_json() for m in base_path_mappings]}), ) elif self.method == "POST": base_path = self._get_param("basePath") rest_api_id = self._get_param("restApiId") stage = self._get_param("stage") base_path_mapping_resp = self.backend.create_base_path_mapping( domain_name, rest_api_id, base_path, stage ) return 201, {}, json.dumps(base_path_mapping_resp.to_json()) def base_path_mapping_individual( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") domain_name = url_path_parts[2] base_path = unquote(url_path_parts[4]) if self.method == "GET": base_path_mapping = self.backend.get_base_path_mapping( domain_name, base_path ) return 200, {}, json.dumps(base_path_mapping.to_json()) elif self.method == "DELETE": self.backend.delete_base_path_mapping(domain_name, base_path) return 202, {}, "" elif self.method == "PATCH": patch_operations = self._get_param("patchOperations") base_path_mapping = self.backend.update_base_path_mapping( domain_name, base_path, patch_operations ) return 200, {}, json.dumps(base_path_mapping.to_json()) def vpc_link( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") vpc_link_id = url_path_parts[-1] if self.method == "DELETE": self.backend.delete_vpc_link(vpc_link_id=vpc_link_id) return 202, {}, "{}" if self.method == "GET": vpc_link = self.backend.get_vpc_link(vpc_link_id=vpc_link_id) return 200, {}, json.dumps(vpc_link.to_json()) def vpc_links( # type: ignore[return] self, request: Any, full_url: str, headers: Dict[str, str] ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) if self.method == "GET": vpc_links = self.backend.get_vpc_links() return 200, {}, json.dumps({"item": [v.to_json() for v in vpc_links]}) if self.method == "POST": name = self._get_param("name") description = self._get_param("description") target_arns = self._get_param("targetArns") tags = self._get_param("tags") vpc_link = self.backend.create_vpc_link( name=name, description=description, target_arns=target_arns, tags=tags ) return 202, {}, json.dumps(vpc_link.to_json()) def put_gateway_response(self) -> TYPE_RESPONSE: rest_api_id = self.path.split("/")[-3] response_type = self.path.split("/")[-1] params = json.loads(self.body) status_code = params.get("statusCode") response_parameters = params.get("responseParameters") response_templates = params.get("responseTemplates") response = self.backend.put_gateway_response( rest_api_id=rest_api_id, response_type=response_type, status_code=status_code, response_parameters=response_parameters, response_templates=response_templates, ) return 201, {}, json.dumps(response.to_json()) def get_gateway_response(self) -> TYPE_RESPONSE: rest_api_id = self.path.split("/")[-3] response_type = self.path.split("/")[-1] response = self.backend.get_gateway_response( rest_api_id=rest_api_id, response_type=response_type ) return 200, {}, json.dumps(response.to_json()) def get_gateway_responses(self) -> TYPE_RESPONSE: rest_api_id = self.path.split("/")[-2] responses = self.backend.get_gateway_responses(rest_api_id=rest_api_id) return 200, {}, json.dumps(dict(item=[gw.to_json() for gw in responses])) def delete_gateway_response(self) -> TYPE_RESPONSE: rest_api_id = self.path.split("/")[-3] response_type = self.path.split("/")[-1] self.backend.delete_gateway_response( rest_api_id=rest_api_id, response_type=response_type ) return 202, {}, json.dumps(dict())