1import json 2from urllib.parse import unquote 3 4from moto.utilities.utils import merge_multiple_dicts 5from moto.core.responses import BaseResponse 6from .models import apigateway_backends 7from .exceptions import ( 8 ApiKeyNotFoundException, 9 UsagePlanNotFoundException, 10 BadRequestException, 11 CrossAccountNotAllowed, 12 AuthorizerNotFoundException, 13 StageNotFoundException, 14 ApiKeyAlreadyExists, 15 DomainNameNotFound, 16 InvalidDomainName, 17 InvalidRestApiId, 18 InvalidModelName, 19 RestAPINotFound, 20 ModelNotFound, 21 ApiKeyValueMinLength, 22 InvalidRequestInput, 23 NoIntegrationDefined, 24 NoIntegrationResponseDefined, 25 NotFoundException, 26 ConflictException, 27) 28 29API_KEY_SOURCES = ["AUTHORIZER", "HEADER"] 30AUTHORIZER_TYPES = ["TOKEN", "REQUEST", "COGNITO_USER_POOLS"] 31ENDPOINT_CONFIGURATION_TYPES = ["PRIVATE", "EDGE", "REGIONAL"] 32 33 34class APIGatewayResponse(BaseResponse): 35 def error(self, type_, message, status=400): 36 headers = self.response_headers or {} 37 headers["X-Amzn-Errortype"] = type_ 38 return ( 39 status, 40 headers, 41 json.dumps({"__type": type_, "message": message}), 42 ) 43 44 @property 45 def backend(self): 46 return apigateway_backends[self.region] 47 48 def __validate_api_key_source(self, api_key_source): 49 if api_key_source and api_key_source not in API_KEY_SOURCES: 50 return self.error( 51 "ValidationException", 52 ( 53 "1 validation error detected: " 54 "Value '{api_key_source}' at 'createRestApiInput.apiKeySource' failed " 55 "to satisfy constraint: Member must satisfy enum value set: " 56 "[AUTHORIZER, HEADER]" 57 ).format(api_key_source=api_key_source), 58 ) 59 60 def __validate_endpoint_configuration(self, endpoint_configuration): 61 if endpoint_configuration and "types" in endpoint_configuration: 62 invalid_types = list( 63 set(endpoint_configuration["types"]) - set(ENDPOINT_CONFIGURATION_TYPES) 64 ) 65 if invalid_types: 66 return self.error( 67 "ValidationException", 68 ( 69 "1 validation error detected: Value '{endpoint_type}' " 70 "at 'createRestApiInput.endpointConfiguration.types' failed " 71 "to satisfy constraint: Member must satisfy enum value set: " 72 "[PRIVATE, EDGE, REGIONAL]" 73 ).format(endpoint_type=invalid_types[0]), 74 ) 75 76 def restapis(self, request, full_url, headers): 77 self.setup_class(request, full_url, headers) 78 79 if self.method == "GET": 80 apis = self.backend.list_apis() 81 return 200, {}, json.dumps({"item": [api.to_dict() for api in apis]}) 82 elif self.method == "POST": 83 name = self._get_param("name") 84 description = self._get_param("description") 85 api_key_source = self._get_param("apiKeySource") 86 endpoint_configuration = self._get_param("endpointConfiguration") 87 tags = self._get_param("tags") 88 policy = self._get_param("policy") 89 minimum_compression_size = self._get_param("minimumCompressionSize") 90 91 # Param validation 92 response = self.__validate_api_key_source(api_key_source) 93 if response is not None: 94 return response 95 96 response = self.__validate_endpoint_configuration(endpoint_configuration) 97 if response is not None: 98 return response 99 100 rest_api = self.backend.create_rest_api( 101 name, 102 description, 103 api_key_source=api_key_source, 104 endpoint_configuration=endpoint_configuration, 105 tags=tags, 106 policy=policy, 107 minimum_compression_size=minimum_compression_size, 108 ) 109 return 200, {}, json.dumps(rest_api.to_dict()) 110 111 def __validte_rest_patch_operations(self, patch_operations): 112 for op in patch_operations: 113 path = op["path"] 114 if "apiKeySource" in path: 115 value = op["value"] 116 return self.__validate_api_key_source(value) 117 118 def restapis_individual(self, request, full_url, headers): 119 self.setup_class(request, full_url, headers) 120 function_id = self.path.replace("/restapis/", "", 1).split("/")[0] 121 122 if self.method == "GET": 123 rest_api = self.backend.get_rest_api(function_id) 124 elif self.method == "DELETE": 125 rest_api = self.backend.delete_rest_api(function_id) 126 elif self.method == "PATCH": 127 patch_operations = self._get_param("patchOperations") 128 response = self.__validte_rest_patch_operations(patch_operations) 129 if response is not None: 130 return response 131 try: 132 rest_api = self.backend.update_rest_api(function_id, patch_operations) 133 except RestAPINotFound as error: 134 return ( 135 error.code, 136 {}, 137 '{{"message":"{0}","code":"{1}"}}'.format( 138 error.message, error.error_type 139 ), 140 ) 141 142 return 200, {}, json.dumps(rest_api.to_dict()) 143 144 def resources(self, request, full_url, headers): 145 self.setup_class(request, full_url, headers) 146 function_id = self.path.replace("/restapis/", "", 1).split("/")[0] 147 148 if self.method == "GET": 149 resources = self.backend.list_resources(function_id) 150 return ( 151 200, 152 {}, 153 json.dumps({"item": [resource.to_dict() for resource in resources]}), 154 ) 155 156 def resource_individual(self, request, full_url, headers): 157 self.setup_class(request, full_url, headers) 158 function_id = self.path.replace("/restapis/", "", 1).split("/")[0] 159 resource_id = self.path.split("/")[-1] 160 161 try: 162 if self.method == "GET": 163 resource = self.backend.get_resource(function_id, resource_id) 164 elif self.method == "POST": 165 path_part = self._get_param("pathPart") 166 resource = self.backend.create_resource( 167 function_id, resource_id, path_part 168 ) 169 elif self.method == "DELETE": 170 resource = self.backend.delete_resource(function_id, resource_id) 171 return 200, {}, json.dumps(resource.to_dict()) 172 except BadRequestException as e: 173 return self.error("BadRequestException", e.message) 174 175 def resource_methods(self, request, full_url, headers): 176 self.setup_class(request, full_url, headers) 177 url_path_parts = self.path.split("/") 178 function_id = url_path_parts[2] 179 resource_id = url_path_parts[4] 180 method_type = url_path_parts[6] 181 182 if self.method == "GET": 183 try: 184 method = self.backend.get_method(function_id, resource_id, method_type) 185 return 200, {}, json.dumps(method) 186 except NotFoundException as nfe: 187 return self.error("NotFoundException", nfe.message) 188 elif self.method == "PUT": 189 authorization_type = self._get_param("authorizationType") 190 api_key_required = self._get_param("apiKeyRequired") 191 request_models = self._get_param("requestModels") 192 operation_name = self._get_param("operationName") 193 authorizer_id = self._get_param("authorizerId") 194 authorization_scopes = self._get_param("authorizationScopes") 195 request_validator_id = self._get_param("requestValidatorId") 196 method = self.backend.create_method( 197 function_id, 198 resource_id, 199 method_type, 200 authorization_type, 201 api_key_required, 202 request_models=request_models, 203 operation_name=operation_name, 204 authorizer_id=authorizer_id, 205 authorization_scopes=authorization_scopes, 206 request_validator_id=request_validator_id, 207 ) 208 return 200, {}, json.dumps(method) 209 210 elif self.method == "DELETE": 211 self.backend.delete_method(function_id, resource_id, method_type) 212 return 200, {}, "" 213 214 elif self.method == "PATCH": 215 patch_operations = self._get_param("patchOperations") 216 self.backend.update_method( 217 function_id, resource_id, method_type, patch_operations 218 ) 219 220 return 200, {}, "" 221 222 def resource_method_responses(self, request, full_url, headers): 223 self.setup_class(request, full_url, headers) 224 url_path_parts = self.path.split("/") 225 function_id = url_path_parts[2] 226 resource_id = url_path_parts[4] 227 method_type = url_path_parts[6] 228 response_code = url_path_parts[8] 229 230 if self.method == "GET": 231 method_response = self.backend.get_method_response( 232 function_id, resource_id, method_type, response_code 233 ) 234 elif self.method == "PUT": 235 response_models = self._get_param("responseModels") 236 response_parameters = self._get_param("responseParameters") 237 method_response = self.backend.create_method_response( 238 function_id, 239 resource_id, 240 method_type, 241 response_code, 242 response_models, 243 response_parameters, 244 ) 245 elif self.method == "DELETE": 246 method_response = self.backend.delete_method_response( 247 function_id, resource_id, method_type, response_code 248 ) 249 elif self.method == "PATCH": 250 patch_operations = self._get_param("patchOperations") 251 method_response = self.backend.update_method_response( 252 function_id, resource_id, method_type, response_code, patch_operations 253 ) 254 else: 255 raise Exception('Unexpected HTTP method "%s"' % self.method) 256 return 200, {}, json.dumps(method_response) 257 258 def restapis_authorizers(self, request, full_url, headers): 259 self.setup_class(request, full_url, headers) 260 url_path_parts = self.path.split("/") 261 restapi_id = url_path_parts[2] 262 263 if self.method == "POST": 264 name = self._get_param("name") 265 authorizer_type = self._get_param("type") 266 267 provider_arns = self._get_param("providerARNs") 268 auth_type = self._get_param("authType") 269 authorizer_uri = self._get_param("authorizerUri") 270 authorizer_credentials = self._get_param("authorizerCredentials") 271 identity_source = self._get_param("identitySource") 272 identiy_validation_expression = self._get_param( 273 "identityValidationExpression" 274 ) 275 authorizer_result_ttl = self._get_param( 276 "authorizerResultTtlInSeconds", if_none=300 277 ) 278 279 # Param validation 280 if authorizer_type and authorizer_type not in AUTHORIZER_TYPES: 281 return self.error( 282 "ValidationException", 283 ( 284 "1 validation error detected: " 285 "Value '{authorizer_type}' at 'createAuthorizerInput.type' failed " 286 "to satisfy constraint: Member must satisfy enum value set: " 287 "[TOKEN, REQUEST, COGNITO_USER_POOLS]" 288 ).format(authorizer_type=authorizer_type), 289 ) 290 291 authorizer_response = self.backend.create_authorizer( 292 restapi_id, 293 name, 294 authorizer_type, 295 provider_arns=provider_arns, 296 auth_type=auth_type, 297 authorizer_uri=authorizer_uri, 298 authorizer_credentials=authorizer_credentials, 299 identity_source=identity_source, 300 identiy_validation_expression=identiy_validation_expression, 301 authorizer_result_ttl=authorizer_result_ttl, 302 ) 303 elif self.method == "GET": 304 authorizers = self.backend.get_authorizers(restapi_id) 305 return 200, {}, json.dumps({"item": authorizers}) 306 307 return 200, {}, json.dumps(authorizer_response) 308 309 def request_validators(self, request, full_url, headers): 310 self.setup_class(request, full_url, headers) 311 url_path_parts = self.path.split("/") 312 restapi_id = url_path_parts[2] 313 try: 314 315 if self.method == "GET": 316 validators = self.backend.get_request_validators(restapi_id) 317 res = json.dumps( 318 {"item": [validator.to_dict() for validator in validators]} 319 ) 320 return 200, {}, res 321 if self.method == "POST": 322 name = self._get_param("name") 323 body = self._get_bool_param("validateRequestBody") 324 params = self._get_bool_param("validateRequestParameters") 325 validator = self.backend.create_request_validator( 326 restapi_id, name, body, params 327 ) 328 return 200, {}, json.dumps(validator) 329 except BadRequestException as e: 330 return self.error("BadRequestException", e.message) 331 except CrossAccountNotAllowed as e: 332 return self.error("AccessDeniedException", e.message) 333 334 def request_validator_individual(self, request, full_url, headers): 335 self.setup_class(request, full_url, headers) 336 url_path_parts = self.path.split("/") 337 restapi_id = url_path_parts[2] 338 validator_id = url_path_parts[4] 339 try: 340 if self.method == "GET": 341 validator = self.backend.get_request_validator(restapi_id, validator_id) 342 return 200, {}, json.dumps(validator) 343 if self.method == "DELETE": 344 self.backend.delete_request_validator(restapi_id, validator_id) 345 return 202, {}, "" 346 if self.method == "PATCH": 347 patch_operations = self._get_param("patchOperations") 348 validator = self.backend.update_request_validator( 349 restapi_id, validator_id, patch_operations 350 ) 351 return 200, {}, json.dumps(validator) 352 except BadRequestException as e: 353 return self.error("BadRequestException", e.message) 354 except CrossAccountNotAllowed as e: 355 return self.error("AccessDeniedException", e.message) 356 357 def authorizers(self, request, full_url, headers): 358 self.setup_class(request, full_url, headers) 359 url_path_parts = self.path.split("/") 360 restapi_id = url_path_parts[2] 361 authorizer_id = url_path_parts[4] 362 363 if self.method == "GET": 364 try: 365 authorizer_response = self.backend.get_authorizer( 366 restapi_id, authorizer_id 367 ) 368 except AuthorizerNotFoundException as error: 369 return ( 370 error.code, 371 {}, 372 '{{"message":"{0}","code":"{1}"}}'.format( 373 error.message, error.error_type 374 ), 375 ) 376 elif self.method == "PATCH": 377 patch_operations = self._get_param("patchOperations") 378 authorizer_response = self.backend.update_authorizer( 379 restapi_id, authorizer_id, patch_operations 380 ) 381 elif self.method == "DELETE": 382 self.backend.delete_authorizer(restapi_id, authorizer_id) 383 return 202, {}, "{}" 384 return 200, {}, json.dumps(authorizer_response) 385 386 def restapis_stages(self, request, full_url, headers): 387 self.setup_class(request, full_url, headers) 388 url_path_parts = self.path.split("/") 389 function_id = url_path_parts[2] 390 391 if self.method == "POST": 392 stage_name = self._get_param("stageName") 393 deployment_id = self._get_param("deploymentId") 394 stage_variables = self._get_param("variables", if_none={}) 395 description = self._get_param("description", if_none="") 396 cacheClusterEnabled = self._get_param("cacheClusterEnabled", if_none=False) 397 cacheClusterSize = self._get_param("cacheClusterSize") 398 tags = self._get_param("tags") 399 tracing_enabled = self._get_param("tracingEnabled") 400 401 stage_response = self.backend.create_stage( 402 function_id, 403 stage_name, 404 deployment_id, 405 variables=stage_variables, 406 description=description, 407 cacheClusterEnabled=cacheClusterEnabled, 408 cacheClusterSize=cacheClusterSize, 409 tags=tags, 410 tracing_enabled=tracing_enabled, 411 ) 412 elif self.method == "GET": 413 stages = self.backend.get_stages(function_id) 414 return 200, {}, json.dumps({"item": stages}) 415 416 return 200, {}, json.dumps(stage_response) 417 418 def restapis_stages_tags(self, request, full_url, headers): 419 self.setup_class(request, full_url, headers) 420 url_path_parts = self.path.split("/") 421 function_id = url_path_parts[4] 422 stage_name = url_path_parts[6] 423 if self.method == "PUT": 424 tags = self._get_param("tags") 425 if tags: 426 stage = self.backend.get_stage(function_id, stage_name) 427 stage["tags"] = merge_multiple_dicts(stage.get("tags"), tags) 428 return 200, {}, json.dumps({"item": tags}) 429 if self.method == "DELETE": 430 stage = self.backend.get_stage(function_id, stage_name) 431 for tag in stage.get("tags").copy(): 432 if tag in self.querystring.get("tagKeys"): 433 stage["tags"].pop(tag, None) 434 return 200, {}, json.dumps({"item": ""}) 435 436 def stages(self, request, full_url, headers): 437 self.setup_class(request, full_url, headers) 438 url_path_parts = self.path.split("/") 439 function_id = url_path_parts[2] 440 stage_name = url_path_parts[4] 441 442 try: 443 if self.method == "GET": 444 stage_response = self.backend.get_stage(function_id, stage_name) 445 446 elif self.method == "PATCH": 447 patch_operations = self._get_param("patchOperations") 448 stage_response = self.backend.update_stage( 449 function_id, stage_name, patch_operations 450 ) 451 elif self.method == "DELETE": 452 self.backend.delete_stage(function_id, stage_name) 453 return 202, {}, "{}" 454 return 200, {}, json.dumps(stage_response) 455 except StageNotFoundException as error: 456 return error.code, {}, error.get_body() 457 458 def integrations(self, request, full_url, headers): 459 self.setup_class(request, full_url, headers) 460 url_path_parts = self.path.split("/") 461 function_id = url_path_parts[2] 462 resource_id = url_path_parts[4] 463 method_type = url_path_parts[6] 464 465 try: 466 integration_response = {} 467 468 if self.method == "GET": 469 integration_response = self.backend.get_integration( 470 function_id, resource_id, method_type 471 ) 472 elif self.method == "PUT": 473 integration_type = self._get_param("type") 474 uri = self._get_param("uri") 475 credentials = self._get_param("credentials") 476 request_templates = self._get_param("requestTemplates") 477 tls_config = self._get_param("tlsConfig") 478 cache_namespace = self._get_param("cacheNamespace") 479 self.backend.get_method(function_id, resource_id, method_type) 480 481 integration_http_method = self._get_param( 482 "httpMethod" 483 ) # default removed because it's a required parameter 484 485 integration_response = self.backend.create_integration( 486 function_id, 487 resource_id, 488 method_type, 489 integration_type, 490 uri, 491 credentials=credentials, 492 integration_method=integration_http_method, 493 request_templates=request_templates, 494 tls_config=tls_config, 495 cache_namespace=cache_namespace, 496 ) 497 elif self.method == "DELETE": 498 integration_response = self.backend.delete_integration( 499 function_id, resource_id, method_type 500 ) 501 502 return 200, {}, json.dumps(integration_response) 503 504 except BadRequestException as e: 505 return self.error("BadRequestException", e.message) 506 except CrossAccountNotAllowed as e: 507 return self.error("AccessDeniedException", e.message) 508 509 def integration_responses(self, request, full_url, headers): 510 self.setup_class(request, full_url, headers) 511 url_path_parts = self.path.split("/") 512 function_id = url_path_parts[2] 513 resource_id = url_path_parts[4] 514 method_type = url_path_parts[6] 515 status_code = url_path_parts[9] 516 517 try: 518 if self.method == "GET": 519 integration_response = self.backend.get_integration_response( 520 function_id, resource_id, method_type, status_code 521 ) 522 elif self.method == "PUT": 523 if not self.body: 524 raise InvalidRequestInput() 525 526 selection_pattern = self._get_param("selectionPattern") 527 response_templates = self._get_param("responseTemplates") 528 content_handling = self._get_param("contentHandling") 529 integration_response = self.backend.create_integration_response( 530 function_id, 531 resource_id, 532 method_type, 533 status_code, 534 selection_pattern, 535 response_templates, 536 content_handling, 537 ) 538 elif self.method == "DELETE": 539 integration_response = self.backend.delete_integration_response( 540 function_id, resource_id, method_type, status_code 541 ) 542 return 200, {}, json.dumps(integration_response) 543 except BadRequestException as e: 544 return self.error("BadRequestException", e.message) 545 except (NoIntegrationDefined, NoIntegrationResponseDefined) as e: 546 return self.error("NotFoundException", e.message) 547 548 def deployments(self, request, full_url, headers): 549 self.setup_class(request, full_url, headers) 550 function_id = self.path.replace("/restapis/", "", 1).split("/")[0] 551 552 try: 553 if self.method == "GET": 554 deployments = self.backend.get_deployments(function_id) 555 return 200, {}, json.dumps({"item": deployments}) 556 elif self.method == "POST": 557 name = self._get_param("stageName") 558 description = self._get_param("description", if_none="") 559 stage_variables = self._get_param("variables", if_none={}) 560 deployment = self.backend.create_deployment( 561 function_id, name, description, stage_variables 562 ) 563 return 200, {}, json.dumps(deployment) 564 except BadRequestException as e: 565 return self.error("BadRequestException", e.message) 566 except NotFoundException as e: 567 return self.error("NotFoundException", e.message) 568 569 def individual_deployment(self, request, full_url, headers): 570 self.setup_class(request, full_url, headers) 571 url_path_parts = self.path.split("/") 572 function_id = url_path_parts[2] 573 deployment_id = url_path_parts[4] 574 575 deployment = None 576 if self.method == "GET": 577 deployment = self.backend.get_deployment(function_id, deployment_id) 578 elif self.method == "DELETE": 579 deployment = self.backend.delete_deployment(function_id, deployment_id) 580 return 200, {}, json.dumps(deployment) 581 582 def apikeys(self, request, full_url, headers): 583 self.setup_class(request, full_url, headers) 584 585 if self.method == "POST": 586 try: 587 apikey_response = self.backend.create_api_key(json.loads(self.body)) 588 except ApiKeyAlreadyExists as error: 589 return ( 590 error.code, 591 {}, 592 '{{"message":"{0}","code":"{1}"}}'.format( 593 error.message, error.error_type 594 ), 595 ) 596 597 except ApiKeyValueMinLength as error: 598 return ( 599 error.code, 600 {}, 601 '{{"message":"{0}","code":"{1}"}}'.format( 602 error.message, error.error_type 603 ), 604 ) 605 return 201, {}, json.dumps(apikey_response) 606 607 elif self.method == "GET": 608 include_values = self._get_bool_param("includeValues") 609 apikeys_response = self.backend.get_api_keys(include_values=include_values) 610 return 200, {}, json.dumps({"item": apikeys_response}) 611 612 def apikey_individual(self, request, full_url, headers): 613 self.setup_class(request, full_url, headers) 614 615 url_path_parts = self.path.split("/") 616 apikey = url_path_parts[2] 617 618 status_code = 200 619 if self.method == "GET": 620 include_value = self._get_bool_param("includeValue") 621 try: 622 apikey_response = self.backend.get_api_key( 623 apikey, include_value=include_value 624 ) 625 except ApiKeyNotFoundException as e: 626 return self.error("NotFoundException", e.message) 627 elif self.method == "PATCH": 628 patch_operations = self._get_param("patchOperations") 629 apikey_response = self.backend.update_api_key(apikey, patch_operations) 630 elif self.method == "DELETE": 631 apikey_response = self.backend.delete_api_key(apikey) 632 status_code = 202 633 634 return status_code, {}, json.dumps(apikey_response) 635 636 def usage_plans(self, request, full_url, headers): 637 self.setup_class(request, full_url, headers) 638 if self.method == "POST": 639 usage_plan_response = self.backend.create_usage_plan(json.loads(self.body)) 640 elif self.method == "GET": 641 api_key_id = self.querystring.get("keyId", [None])[0] 642 usage_plans_response = self.backend.get_usage_plans(api_key_id=api_key_id) 643 return 200, {}, json.dumps({"item": usage_plans_response}) 644 return 200, {}, json.dumps(usage_plan_response) 645 646 def usage_plan_individual(self, request, full_url, headers): 647 self.setup_class(request, full_url, headers) 648 649 url_path_parts = self.path.split("/") 650 usage_plan = url_path_parts[2] 651 652 if self.method == "GET": 653 try: 654 usage_plan_response = self.backend.get_usage_plan(usage_plan) 655 except (UsagePlanNotFoundException) as error: 656 return ( 657 error.code, 658 {}, 659 '{{"message":"{0}","code":"{1}"}}'.format( 660 error.message, error.error_type 661 ), 662 ) 663 elif self.method == "DELETE": 664 usage_plan_response = self.backend.delete_usage_plan(usage_plan) 665 elif self.method == "PATCH": 666 patch_operations = self._get_param("patchOperations") 667 usage_plan_response = self.backend.update_usage_plan( 668 usage_plan, patch_operations 669 ) 670 return 200, {}, json.dumps(usage_plan_response) 671 672 def usage_plan_keys(self, request, full_url, headers): 673 self.setup_class(request, full_url, headers) 674 675 url_path_parts = self.path.split("/") 676 usage_plan_id = url_path_parts[2] 677 678 if self.method == "POST": 679 try: 680 usage_plan_response = self.backend.create_usage_plan_key( 681 usage_plan_id, json.loads(self.body) 682 ) 683 except ApiKeyNotFoundException as error: 684 return ( 685 error.code, 686 {}, 687 '{{"message":"{0}","code":"{1}"}}'.format( 688 error.message, error.error_type 689 ), 690 ) 691 return 201, {}, json.dumps(usage_plan_response) 692 elif self.method == "GET": 693 usage_plans_response = self.backend.get_usage_plan_keys(usage_plan_id) 694 return 200, {}, json.dumps({"item": usage_plans_response}) 695 696 def usage_plan_key_individual(self, request, full_url, headers): 697 self.setup_class(request, full_url, headers) 698 699 url_path_parts = self.path.split("/") 700 usage_plan_id = url_path_parts[2] 701 key_id = url_path_parts[4] 702 703 if self.method == "GET": 704 try: 705 usage_plan_response = self.backend.get_usage_plan_key( 706 usage_plan_id, key_id 707 ) 708 except (UsagePlanNotFoundException, ApiKeyNotFoundException) as error: 709 return ( 710 error.code, 711 {}, 712 '{{"message":"{0}","code":"{1}"}}'.format( 713 error.message, error.error_type 714 ), 715 ) 716 elif self.method == "DELETE": 717 usage_plan_response = self.backend.delete_usage_plan_key( 718 usage_plan_id, key_id 719 ) 720 return 200, {}, json.dumps(usage_plan_response) 721 722 def domain_names(self, request, full_url, headers): 723 self.setup_class(request, full_url, headers) 724 725 try: 726 if self.method == "GET": 727 domain_names = self.backend.get_domain_names() 728 return 200, {}, json.dumps({"item": domain_names}) 729 730 elif self.method == "POST": 731 domain_name = self._get_param("domainName") 732 certificate_name = self._get_param("certificateName") 733 tags = self._get_param("tags") 734 certificate_arn = self._get_param("certificateArn") 735 certificate_body = self._get_param("certificateBody") 736 certificate_private_key = self._get_param("certificatePrivateKey") 737 certificate_chain = self._get_param("certificateChain") 738 regional_certificate_name = self._get_param("regionalCertificateName") 739 regional_certificate_arn = self._get_param("regionalCertificateArn") 740 endpoint_configuration = self._get_param("endpointConfiguration") 741 security_policy = self._get_param("securityPolicy") 742 generate_cli_skeleton = self._get_param("generateCliSkeleton") 743 domain_name_resp = self.backend.create_domain_name( 744 domain_name, 745 certificate_name, 746 tags, 747 certificate_arn, 748 certificate_body, 749 certificate_private_key, 750 certificate_chain, 751 regional_certificate_name, 752 regional_certificate_arn, 753 endpoint_configuration, 754 security_policy, 755 generate_cli_skeleton, 756 ) 757 return 200, {}, json.dumps(domain_name_resp) 758 759 except InvalidDomainName as error: 760 return ( 761 error.code, 762 {}, 763 '{{"message":"{0}","code":"{1}"}}'.format( 764 error.message, error.error_type 765 ), 766 ) 767 768 def domain_name_induvidual(self, request, full_url, headers): 769 self.setup_class(request, full_url, headers) 770 771 url_path_parts = self.path.split("/") 772 domain_name = url_path_parts[2] 773 domain_names = {} 774 try: 775 if self.method == "GET": 776 if domain_name is not None: 777 domain_names = self.backend.get_domain_name(domain_name) 778 elif self.method == "DELETE": 779 if domain_name is not None: 780 self.backend.delete_domain_name(domain_name) 781 elif self.method == "PATCH": 782 if domain_name is not None: 783 patch_operations = self._get_param("patchOperations") 784 self.backend.update_domain_name(domain_name, patch_operations) 785 else: 786 msg = ( 787 'Method "%s" for API GW domain names not implemented' % self.method 788 ) 789 return 404, {}, json.dumps({"error": msg}) 790 return 200, {}, json.dumps(domain_names) 791 except DomainNameNotFound as error: 792 return self.error("NotFoundException", error.message) 793 794 def models(self, request, full_url, headers): 795 self.setup_class(request, full_url, headers) 796 rest_api_id = self.path.replace("/restapis/", "", 1).split("/")[0] 797 798 try: 799 if self.method == "GET": 800 models = self.backend.get_models(rest_api_id) 801 return 200, {}, json.dumps({"item": models}) 802 803 elif self.method == "POST": 804 name = self._get_param("name") 805 description = self._get_param("description") 806 schema = self._get_param("schema") 807 content_type = self._get_param("contentType") 808 cli_input_json = self._get_param("cliInputJson") 809 generate_cli_skeleton = self._get_param("generateCliSkeleton") 810 model = self.backend.create_model( 811 rest_api_id, 812 name, 813 content_type, 814 description, 815 schema, 816 cli_input_json, 817 generate_cli_skeleton, 818 ) 819 820 return 200, {}, json.dumps(model) 821 822 except (InvalidRestApiId, InvalidModelName, RestAPINotFound) as error: 823 return ( 824 error.code, 825 {}, 826 '{{"message":"{0}","code":"{1}"}}'.format( 827 error.message, error.error_type 828 ), 829 ) 830 831 def model_induvidual(self, request, full_url, headers): 832 self.setup_class(request, full_url, headers) 833 url_path_parts = self.path.split("/") 834 rest_api_id = url_path_parts[2] 835 model_name = url_path_parts[4] 836 model_info = {} 837 try: 838 if self.method == "GET": 839 model_info = self.backend.get_model(rest_api_id, model_name) 840 return 200, {}, json.dumps(model_info) 841 except ( 842 ModelNotFound, 843 RestAPINotFound, 844 InvalidRestApiId, 845 InvalidModelName, 846 ) as error: 847 return ( 848 error.code, 849 {}, 850 '{{"message":"{0}","code":"{1}"}}'.format( 851 error.message, error.error_type 852 ), 853 ) 854 855 def base_path_mappings(self, request, full_url, headers): 856 self.setup_class(request, full_url, headers) 857 858 url_path_parts = self.path.split("/") 859 domain_name = url_path_parts[2] 860 861 try: 862 if self.method == "GET": 863 base_path_mappings = self.backend.get_base_path_mappings(domain_name) 864 return 200, {}, json.dumps({"item": base_path_mappings}) 865 elif self.method == "POST": 866 base_path = self._get_param("basePath") 867 rest_api_id = self._get_param("restApiId") 868 stage = self._get_param("stage") 869 870 base_path_mapping_resp = self.backend.create_base_path_mapping( 871 domain_name, rest_api_id, base_path, stage, 872 ) 873 return 201, {}, json.dumps(base_path_mapping_resp) 874 except BadRequestException as e: 875 return self.error("BadRequestException", e.message) 876 except NotFoundException as e: 877 return self.error("NotFoundException", e.message, 404) 878 except ConflictException as e: 879 return self.error("ConflictException", e.message, 409) 880 881 def base_path_mapping_individual(self, request, full_url, headers): 882 883 self.setup_class(request, full_url, headers) 884 885 url_path_parts = self.path.split("/") 886 domain_name = url_path_parts[2] 887 base_path = unquote(url_path_parts[4]) 888 889 try: 890 if self.method == "GET": 891 base_path_mapping = self.backend.get_base_path_mapping( 892 domain_name, base_path 893 ) 894 return 200, {}, json.dumps(base_path_mapping) 895 elif self.method == "DELETE": 896 self.backend.delete_base_path_mapping(domain_name, base_path) 897 return 202, {}, "" 898 except NotFoundException as e: 899 return self.error("NotFoundException", e.message, 404) 900