Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

TechDebt: MyPy AppSync (#5570)

Browse Source
This commit is contained in:
Bert Blommers 2022-10-16 18:14:15 +00:00 committed by GitHub
parent 8e3f2331d4
commit 047c1a6b7e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 257 additions and 244 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
moto/amp/responses.py
View File

@ -10,7 +10,7 @@ from urllib.parse import unquote
class PrometheusServiceResponse(BaseResponse): class PrometheusServiceResponse(BaseResponse):
"""Handler for PrometheusService requests and responses.""" """Handler for PrometheusService requests and responses."""
def tags(self, request: Any, full_url: str, headers: Any) -> str: # type: ignore [return] def tags(self, request: Any, full_url: str, headers: Any) -> str: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if request.method == "GET": if request.method == "GET":
return self.list_tags_for_resource() return self.list_tags_for_resource()

88
moto/apigateway/responses.py
View File

@ -1,9 +1,9 @@
import json import json
from typing import Any, Dict, List, Tuple from typing import Any, Dict, List
from urllib.parse import unquote from urllib.parse import unquote
from moto.utilities.utils import merge_multiple_dicts from moto.utilities.utils import merge_multiple_dicts
from moto.core.responses import BaseResponse from moto.core.responses import BaseResponse, TYPE_RESPONSE
from .models import apigateway_backends, APIGatewayBackend from .models import apigateway_backends, APIGatewayBackend
from .utils import deserialize_body from .utils import deserialize_body
from .exceptions import InvalidRequestInput from .exceptions import InvalidRequestInput
@ -12,14 +12,12 @@ API_KEY_SOURCES = ["AUTHORIZER", "HEADER"]
AUTHORIZER_TYPES = ["TOKEN", "REQUEST", "COGNITO_USER_POOLS"] AUTHORIZER_TYPES = ["TOKEN", "REQUEST", "COGNITO_USER_POOLS"]
ENDPOINT_CONFIGURATION_TYPES = ["PRIVATE", "EDGE", "REGIONAL"] ENDPOINT_CONFIGURATION_TYPES = ["PRIVATE", "EDGE", "REGIONAL"]
RESPONSE_TYPE = Tuple[int, Dict[str, str], str]
class APIGatewayResponse(BaseResponse): class APIGatewayResponse(BaseResponse):
def __init__(self) -> None: def __init__(self) -> None:
super().__init__(service_name="apigateway") super().__init__(service_name="apigateway")
def error(self, type_: str, message: str, status: int = 400) -> RESPONSE_TYPE: def error(self, type_: str, message: str, status: int = 400) -> TYPE_RESPONSE:
headers = self.response_headers or {} headers = self.response_headers or {}
headers["X-Amzn-Errortype"] = type_ headers["X-Amzn-Errortype"] = type_
return (status, headers, json.dumps({"__type": type_, "message": message})) return (status, headers, json.dumps({"__type": type_, "message": message}))
@ -28,7 +26,7 @@ class APIGatewayResponse(BaseResponse):
def backend(self) -> APIGatewayBackend: def backend(self) -> APIGatewayBackend:
return apigateway_backends[self.current_account][self.region] return apigateway_backends[self.current_account][self.region]
def __validate_api_key_source(self, api_key_source: str) -> RESPONSE_TYPE: # type: ignore[return] 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: if api_key_source and api_key_source not in API_KEY_SOURCES:
return self.error( return self.error(
"ValidationException", "ValidationException",
@ -40,7 +38,7 @@ class APIGatewayResponse(BaseResponse):
).format(api_key_source=api_key_source), ).format(api_key_source=api_key_source),
) )
def __validate_endpoint_configuration(self, endpoint_configuration: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def __validate_endpoint_configuration(self, endpoint_configuration: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
if endpoint_configuration and "types" in endpoint_configuration: if endpoint_configuration and "types" in endpoint_configuration:
invalid_types = list( invalid_types = list(
set(endpoint_configuration["types"]) - set(ENDPOINT_CONFIGURATION_TYPES) set(endpoint_configuration["types"]) - set(ENDPOINT_CONFIGURATION_TYPES)
@ -56,7 +54,7 @@ class APIGatewayResponse(BaseResponse):
).format(endpoint_type=invalid_types[0]), ).format(endpoint_type=invalid_types[0]),
) )
def restapis(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def restapis(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "GET": if self.method == "GET":
@ -100,7 +98,7 @@ class APIGatewayResponse(BaseResponse):
return 200, {}, json.dumps(rest_api.to_dict()) return 200, {}, json.dumps(rest_api.to_dict())
def __validte_rest_patch_operations(self, patch_operations: List[Dict[str, str]]) -> RESPONSE_TYPE: # type: ignore[return] def __validte_rest_patch_operations(self, patch_operations: List[Dict[str, str]]) -> TYPE_RESPONSE: # type: ignore[return]
for op in patch_operations: for op in patch_operations:
path = op["path"] path = op["path"]
if "apiKeySource" in path: if "apiKeySource" in path:
@ -109,7 +107,7 @@ class APIGatewayResponse(BaseResponse):
def restapis_individual( def restapis_individual(
self, request: Any, full_url: str, headers: Dict[str, str] self, request: Any, full_url: str, headers: Dict[str, str]
) -> RESPONSE_TYPE: ) -> TYPE_RESPONSE:
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0] function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
@ -135,7 +133,7 @@ class APIGatewayResponse(BaseResponse):
return 200, {}, json.dumps(rest_api.to_dict()) return 200, {}, json.dumps(rest_api.to_dict())
def resources(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def resources(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0] function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
@ -147,7 +145,7 @@ class APIGatewayResponse(BaseResponse):
json.dumps({"item": [resource.to_dict() for resource in resources]}), json.dumps({"item": [resource.to_dict() for resource in resources]}),
) )
def gateway_response(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def gateway_response(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if request.method == "PUT": if request.method == "PUT":
return self.put_gateway_response() return self.put_gateway_response()
@ -156,12 +154,12 @@ class APIGatewayResponse(BaseResponse):
elif request.method == "DELETE": elif request.method == "DELETE":
return self.delete_gateway_response() return self.delete_gateway_response()
def gateway_responses(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def gateway_responses(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if request.method == "GET": if request.method == "GET":
return self.get_gateway_responses() return self.get_gateway_responses()
def resource_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def resource_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0] function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
resource_id = self.path.split("/")[-1] resource_id = self.path.split("/")[-1]
@ -179,7 +177,7 @@ class APIGatewayResponse(BaseResponse):
def resource_methods( def resource_methods(
self, request: Any, full_url: str, headers: Dict[str, str] self, request: Any, full_url: str, headers: Dict[str, str]
) -> RESPONSE_TYPE: ) -> TYPE_RESPONSE:
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
function_id = url_path_parts[2] function_id = url_path_parts[2]
@ -221,7 +219,7 @@ class APIGatewayResponse(BaseResponse):
def resource_method_responses( def resource_method_responses(
self, request: Any, full_url: str, headers: Dict[str, str] self, request: Any, full_url: str, headers: Dict[str, str]
) -> RESPONSE_TYPE: ) -> TYPE_RESPONSE:
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
function_id = url_path_parts[2] function_id = url_path_parts[2]
@ -253,7 +251,7 @@ class APIGatewayResponse(BaseResponse):
return 204, {}, json.dumps(method_response) return 204, {}, json.dumps(method_response)
raise Exception('Unexpected HTTP method "%s"' % self.method) raise Exception('Unexpected HTTP method "%s"' % self.method)
def restapis_authorizers(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def restapis_authorizers(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
restapi_id = url_path_parts[2] restapi_id = url_path_parts[2]
@ -303,7 +301,7 @@ class APIGatewayResponse(BaseResponse):
authorizers = self.backend.get_authorizers(restapi_id) authorizers = self.backend.get_authorizers(restapi_id)
return 200, {}, json.dumps({"item": authorizers}) return 200, {}, json.dumps({"item": authorizers})
def request_validators(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def request_validators(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
restapi_id = url_path_parts[2] restapi_id = url_path_parts[2]
@ -323,7 +321,7 @@ class APIGatewayResponse(BaseResponse):
) )
return 201, {}, json.dumps(validator) return 201, {}, json.dumps(validator)
def request_validator_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def request_validator_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
restapi_id = url_path_parts[2] restapi_id = url_path_parts[2]
@ -342,7 +340,7 @@ class APIGatewayResponse(BaseResponse):
) )
return 200, {}, json.dumps(validator) return 200, {}, json.dumps(validator)
def authorizers(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def authorizers(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
restapi_id = url_path_parts[2] restapi_id = url_path_parts[2]
@ -361,7 +359,7 @@ class APIGatewayResponse(BaseResponse):
self.backend.delete_authorizer(restapi_id, authorizer_id) self.backend.delete_authorizer(restapi_id, authorizer_id)
return 202, {}, "{}" return 202, {}, "{}"
def restapis_stages(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def restapis_stages(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
function_id = url_path_parts[2] function_id = url_path_parts[2]
@ -392,7 +390,7 @@ class APIGatewayResponse(BaseResponse):
stages = self.backend.get_stages(function_id) stages = self.backend.get_stages(function_id)
return 200, {}, json.dumps({"item": stages}) return 200, {}, json.dumps({"item": stages})
def restapis_stages_tags(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def restapis_stages_tags(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
function_id = url_path_parts[4] function_id = url_path_parts[4]
@ -410,7 +408,7 @@ class APIGatewayResponse(BaseResponse):
stage["tags"].pop(tag, None) stage["tags"].pop(tag, None)
return 200, {}, json.dumps({"item": ""}) return 200, {}, json.dumps({"item": ""})
def stages(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def stages(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
function_id = url_path_parts[2] function_id = url_path_parts[2]
@ -429,7 +427,7 @@ class APIGatewayResponse(BaseResponse):
self.backend.delete_stage(function_id, stage_name) self.backend.delete_stage(function_id, stage_name)
return 202, {}, "{}" return 202, {}, "{}"
def integrations(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def integrations(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
function_id = url_path_parts[2] function_id = url_path_parts[2]
@ -479,7 +477,7 @@ class APIGatewayResponse(BaseResponse):
) )
return 204, {}, json.dumps(integration_response) return 204, {}, json.dumps(integration_response)
def integration_responses(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def integration_responses(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
function_id = url_path_parts[2] function_id = url_path_parts[2]
@ -515,7 +513,7 @@ class APIGatewayResponse(BaseResponse):
) )
return 204, {}, json.dumps(integration_response) return 204, {}, json.dumps(integration_response)
def deployments(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def deployments(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
function_id = self.path.replace("/restapis/", "", 1).split("/")[0] function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
@ -531,7 +529,7 @@ class APIGatewayResponse(BaseResponse):
) )
return 201, {}, json.dumps(deployment) return 201, {}, json.dumps(deployment)
def individual_deployment(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def individual_deployment(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
function_id = url_path_parts[2] function_id = url_path_parts[2]
@ -544,7 +542,7 @@ class APIGatewayResponse(BaseResponse):
deployment = self.backend.delete_deployment(function_id, deployment_id) deployment = self.backend.delete_deployment(function_id, deployment_id)
return 202, {}, json.dumps(deployment) return 202, {}, json.dumps(deployment)
def apikeys(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def apikeys(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "POST": if self.method == "POST":
@ -558,7 +556,7 @@ class APIGatewayResponse(BaseResponse):
def apikey_individual( def apikey_individual(
self, request: Any, full_url: str, headers: Dict[str, str] self, request: Any, full_url: str, headers: Dict[str, str]
) -> RESPONSE_TYPE: ) -> TYPE_RESPONSE:
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
@ -578,7 +576,7 @@ class APIGatewayResponse(BaseResponse):
return 200, {}, json.dumps(apikey_response) return 200, {}, json.dumps(apikey_response)
def usage_plans(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def usage_plans(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "POST": if self.method == "POST":
usage_plan_response = self.backend.create_usage_plan(json.loads(self.body)) usage_plan_response = self.backend.create_usage_plan(json.loads(self.body))
@ -588,7 +586,7 @@ class APIGatewayResponse(BaseResponse):
usage_plans_response = self.backend.get_usage_plans(api_key_id=api_key_id) usage_plans_response = self.backend.get_usage_plans(api_key_id=api_key_id)
return 200, {}, json.dumps({"item": usage_plans_response}) return 200, {}, json.dumps({"item": usage_plans_response})
def usage_plan_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def usage_plan_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
@ -607,7 +605,7 @@ class APIGatewayResponse(BaseResponse):
) )
return 200, {}, json.dumps(usage_plan_response) return 200, {}, json.dumps(usage_plan_response)
def usage_plan_keys(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def usage_plan_keys(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
@ -622,7 +620,7 @@ class APIGatewayResponse(BaseResponse):
usage_plans_response = self.backend.get_usage_plan_keys(usage_plan_id) usage_plans_response = self.backend.get_usage_plan_keys(usage_plan_id)
return 200, {}, json.dumps({"item": usage_plans_response}) return 200, {}, json.dumps({"item": usage_plans_response})
def usage_plan_key_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def usage_plan_key_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
@ -636,7 +634,7 @@ class APIGatewayResponse(BaseResponse):
self.backend.delete_usage_plan_key(usage_plan_id, key_id) self.backend.delete_usage_plan_key(usage_plan_id, key_id)
return 202, {}, "{}" return 202, {}, "{}"
def domain_names(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def domain_names(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "GET": if self.method == "GET":
@ -672,7 +670,7 @@ class APIGatewayResponse(BaseResponse):
def domain_name_induvidual( def domain_name_induvidual(
self, request: Any, full_url: str, headers: Dict[str, str] self, request: Any, full_url: str, headers: Dict[str, str]
) -> RESPONSE_TYPE: ) -> TYPE_RESPONSE:
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
@ -691,7 +689,7 @@ class APIGatewayResponse(BaseResponse):
msg = 'Method "%s" for API GW domain names not implemented' % self.method msg = 'Method "%s" for API GW domain names not implemented' % self.method
return 404, {}, json.dumps({"error": msg}) return 404, {}, json.dumps({"error": msg})
def models(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def models(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
rest_api_id = self.path.replace("/restapis/", "", 1).split("/")[0] rest_api_id = self.path.replace("/restapis/", "", 1).split("/")[0]
@ -715,7 +713,7 @@ class APIGatewayResponse(BaseResponse):
def model_induvidual( def model_induvidual(
self, request: Any, full_url: str, headers: Dict[str, str] self, request: Any, full_url: str, headers: Dict[str, str]
) -> RESPONSE_TYPE: ) -> TYPE_RESPONSE:
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
rest_api_id = url_path_parts[2] rest_api_id = url_path_parts[2]
@ -726,7 +724,7 @@ class APIGatewayResponse(BaseResponse):
return 200, {}, json.dumps(model_info) return 200, {}, json.dumps(model_info)
return 200, {}, "{}" return 200, {}, "{}"
def base_path_mappings(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def base_path_mappings(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
@ -745,7 +743,7 @@ class APIGatewayResponse(BaseResponse):
) )
return 201, {}, json.dumps(base_path_mapping_resp) return 201, {}, json.dumps(base_path_mapping_resp)
def base_path_mapping_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def base_path_mapping_individual(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
@ -768,7 +766,7 @@ class APIGatewayResponse(BaseResponse):
) )
return 200, {}, json.dumps(base_path_mapping) return 200, {}, json.dumps(base_path_mapping)
def vpc_link(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def vpc_link(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
url_path_parts = self.path.split("/") url_path_parts = self.path.split("/")
vpc_link_id = url_path_parts[-1] vpc_link_id = url_path_parts[-1]
@ -780,7 +778,7 @@ class APIGatewayResponse(BaseResponse):
vpc_link = self.backend.get_vpc_link(vpc_link_id=vpc_link_id) vpc_link = self.backend.get_vpc_link(vpc_link_id=vpc_link_id)
return 200, {}, json.dumps(vpc_link) return 200, {}, json.dumps(vpc_link)
def vpc_links(self, request: Any, full_url: str, headers: Dict[str, str]) -> RESPONSE_TYPE: # type: ignore[return] def vpc_links(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "GET": if self.method == "GET":
@ -796,7 +794,7 @@ class APIGatewayResponse(BaseResponse):
) )
return 202, {}, json.dumps(vpc_link) return 202, {}, json.dumps(vpc_link)
def put_gateway_response(self) -> RESPONSE_TYPE: def put_gateway_response(self) -> TYPE_RESPONSE:
rest_api_id = self.path.split("/")[-3] rest_api_id = self.path.split("/")[-3]
response_type = self.path.split("/")[-1] response_type = self.path.split("/")[-1]
params = json.loads(self.body) params = json.loads(self.body)
@ -812,7 +810,7 @@ class APIGatewayResponse(BaseResponse):
) )
return 201, {}, json.dumps(response) return 201, {}, json.dumps(response)
def get_gateway_response(self) -> RESPONSE_TYPE: def get_gateway_response(self) -> TYPE_RESPONSE:
rest_api_id = self.path.split("/")[-3] rest_api_id = self.path.split("/")[-3]
response_type = self.path.split("/")[-1] response_type = self.path.split("/")[-1]
response = self.backend.get_gateway_response( response = self.backend.get_gateway_response(
@ -820,12 +818,12 @@ class APIGatewayResponse(BaseResponse):
) )
return 200, {}, json.dumps(response) return 200, {}, json.dumps(response)
def get_gateway_responses(self) -> RESPONSE_TYPE: def get_gateway_responses(self) -> TYPE_RESPONSE:
rest_api_id = self.path.split("/")[-2] rest_api_id = self.path.split("/")[-2]
responses = self.backend.get_gateway_responses(rest_api_id=rest_api_id) responses = self.backend.get_gateway_responses(rest_api_id=rest_api_id)
return 200, {}, json.dumps(dict(item=responses)) return 200, {}, json.dumps(dict(item=responses))
def delete_gateway_response(self) -> RESPONSE_TYPE: def delete_gateway_response(self) -> TYPE_RESPONSE:
rest_api_id = self.path.split("/")[-3] rest_api_id = self.path.split("/")[-3]
response_type = self.path.split("/")[-1] response_type = self.path.split("/")[-1]
self.backend.delete_gateway_response( self.backend.delete_gateway_response(

129
moto/apigatewayv2/responses.py
View File

@ -1,17 +1,14 @@
"""Handles incoming apigatewayv2 requests, invokes methods, returns responses.""" """Handles incoming apigatewayv2 requests, invokes methods, returns responses."""
import json import json
from moto.core.responses import BaseResponse from moto.core.responses import BaseResponse, TYPE_RESPONSE
from typing import Any, Tuple, Dict from typing import Any
from urllib.parse import unquote from urllib.parse import unquote
from .exceptions import UnknownProtocol from .exceptions import UnknownProtocol
from .models import apigatewayv2_backends, ApiGatewayV2Backend from .models import apigatewayv2_backends, ApiGatewayV2Backend
RESPONSE_TYPE = Tuple[int, Dict[str, str], str]
class ApiGatewayV2Response(BaseResponse): class ApiGatewayV2Response(BaseResponse):
"""Handler for ApiGatewayV2 requests and responses.""" """Handler for ApiGatewayV2 requests and responses."""
@ -23,7 +20,7 @@ class ApiGatewayV2Response(BaseResponse):
"""Return backend instance specific for this region.""" """Return backend instance specific for this region."""
return apigatewayv2_backends[self.current_account][self.region] return apigatewayv2_backends[self.current_account][self.region]
def apis(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def apis(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "POST": if self.method == "POST":
@ -31,7 +28,7 @@ class ApiGatewayV2Response(BaseResponse):
if self.method == "GET": if self.method == "GET":
return self.get_apis() return self.get_apis()
def api(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def api(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "GET": if self.method == "GET":
@ -43,7 +40,7 @@ class ApiGatewayV2Response(BaseResponse):
if self.method == "DELETE": if self.method == "DELETE":
return self.delete_api() return self.delete_api()
def authorizer(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def authorizer(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "DELETE": if self.method == "DELETE":
@ -53,25 +50,25 @@ class ApiGatewayV2Response(BaseResponse):
if self.method == "PATCH": if self.method == "PATCH":
return self.update_authorizer() return self.update_authorizer()
def authorizers(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def authorizers(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "POST": if self.method == "POST":
return self.create_authorizer() return self.create_authorizer()
def cors(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def cors(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "DELETE": if self.method == "DELETE":
return self.delete_cors_configuration() return self.delete_cors_configuration()
def route_request_parameter(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def route_request_parameter(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "DELETE": if self.method == "DELETE":
return self.delete_route_request_parameter() return self.delete_route_request_parameter()
def model(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def model(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "DELETE": if self.method == "DELETE":
@ -81,13 +78,13 @@ class ApiGatewayV2Response(BaseResponse):
if self.method == "PATCH": if self.method == "PATCH":
return self.update_model() return self.update_model()
def models(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def models(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "POST": if self.method == "POST":
return self.create_model() return self.create_model()
def integration(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def integration(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "DELETE": if self.method == "DELETE":
@ -97,7 +94,7 @@ class ApiGatewayV2Response(BaseResponse):
if self.method == "PATCH": if self.method == "PATCH":
return self.update_integration() return self.update_integration()
def integrations(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def integrations(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "GET": if self.method == "GET":
@ -105,7 +102,7 @@ class ApiGatewayV2Response(BaseResponse):
if self.method == "POST": if self.method == "POST":
return self.create_integration() return self.create_integration()
def integration_response(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def integration_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "DELETE": if self.method == "DELETE":
@ -115,7 +112,7 @@ class ApiGatewayV2Response(BaseResponse):
if self.method == "PATCH": if self.method == "PATCH":
return self.update_integration_response() return self.update_integration_response()
def integration_responses(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def integration_responses(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "GET": if self.method == "GET":
@ -123,7 +120,7 @@ class ApiGatewayV2Response(BaseResponse):
if self.method == "POST": if self.method == "POST":
return self.create_integration_response() return self.create_integration_response()
def route(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def route(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "DELETE": if self.method == "DELETE":
@ -133,7 +130,7 @@ class ApiGatewayV2Response(BaseResponse):
if self.method == "PATCH": if self.method == "PATCH":
return self.update_route() return self.update_route()
def routes(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def routes(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "GET": if self.method == "GET":
@ -141,7 +138,7 @@ class ApiGatewayV2Response(BaseResponse):
if self.method == "POST": if self.method == "POST":
return self.create_route() return self.create_route()
def route_response(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def route_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "DELETE": if self.method == "DELETE":
@ -149,13 +146,13 @@ class ApiGatewayV2Response(BaseResponse):
if self.method == "GET": if self.method == "GET":
return self.get_route_response() return self.get_route_response()
def route_responses(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def route_responses(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "POST": if self.method == "POST":
return self.create_route_response() return self.create_route_response()
def tags(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def tags(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if self.method == "POST": if self.method == "POST":
@ -165,7 +162,7 @@ class ApiGatewayV2Response(BaseResponse):
if self.method == "DELETE": if self.method == "DELETE":
return self.untag_resource() return self.untag_resource()
def vpc_link(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def vpc_link(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if request.method == "DELETE": if request.method == "DELETE":
@ -175,7 +172,7 @@ class ApiGatewayV2Response(BaseResponse):
if request.method == "PATCH": if request.method == "PATCH":
return self.update_vpc_link() return self.update_vpc_link()
def vpc_links(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] def vpc_links(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if request.method == "GET": if request.method == "GET":
@ -183,7 +180,7 @@ class ApiGatewayV2Response(BaseResponse):
if request.method == "POST": if request.method == "POST":
return self.create_vpc_link() return self.create_vpc_link()
def create_api(self) -> RESPONSE_TYPE: def create_api(self) -> TYPE_RESPONSE:
params = json.loads(self.body) params = json.loads(self.body)
api_key_selection_expression = params.get("apiKeySelectionExpression") api_key_selection_expression = params.get("apiKeySelectionExpression")
@ -214,21 +211,21 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(api.to_json()) return 200, {}, json.dumps(api.to_json())
def delete_api(self) -> RESPONSE_TYPE: def delete_api(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-1] api_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_api(api_id=api_id) self.apigatewayv2_backend.delete_api(api_id=api_id)
return 200, {}, "{}" return 200, {}, "{}"
def get_api(self) -> RESPONSE_TYPE: def get_api(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-1] api_id = self.path.split("/")[-1]
api = self.apigatewayv2_backend.get_api(api_id=api_id) api = self.apigatewayv2_backend.get_api(api_id=api_id)
return 200, {}, json.dumps(api.to_json()) return 200, {}, json.dumps(api.to_json())
def get_apis(self) -> RESPONSE_TYPE: def get_apis(self) -> TYPE_RESPONSE:
apis = self.apigatewayv2_backend.get_apis() apis = self.apigatewayv2_backend.get_apis()
return 200, {}, json.dumps({"items": [a.to_json() for a in apis]}) return 200, {}, json.dumps({"items": [a.to_json() for a in apis]})
def update_api(self) -> RESPONSE_TYPE: def update_api(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-1] api_id = self.path.split("/")[-1]
params = json.loads(self.body) params = json.loads(self.body)
api_key_selection_expression = params.get("apiKeySelectionExpression") api_key_selection_expression = params.get("apiKeySelectionExpression")
@ -252,7 +249,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(api.to_json()) return 200, {}, json.dumps(api.to_json())
def reimport_api(self) -> RESPONSE_TYPE: def reimport_api(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-1] api_id = self.path.split("/")[-1]
params = json.loads(self.body) params = json.loads(self.body)
body = params.get("body") body = params.get("body")
@ -263,7 +260,7 @@ class ApiGatewayV2Response(BaseResponse):
api = self.apigatewayv2_backend.reimport_api(api_id, body, fail_on_warnings) api = self.apigatewayv2_backend.reimport_api(api_id, body, fail_on_warnings)
return 201, {}, json.dumps(api.to_json()) return 201, {}, json.dumps(api.to_json())
def create_authorizer(self) -> RESPONSE_TYPE: def create_authorizer(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-2] api_id = self.path.split("/")[-2]
params = json.loads(self.body) params = json.loads(self.body)
@ -292,21 +289,21 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(authorizer.to_json()) return 200, {}, json.dumps(authorizer.to_json())
def delete_authorizer(self) -> RESPONSE_TYPE: def delete_authorizer(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
authorizer_id = self.path.split("/")[-1] authorizer_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_authorizer(api_id, authorizer_id) self.apigatewayv2_backend.delete_authorizer(api_id, authorizer_id)
return 200, {}, "{}" return 200, {}, "{}"
def get_authorizer(self) -> RESPONSE_TYPE: def get_authorizer(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
authorizer_id = self.path.split("/")[-1] authorizer_id = self.path.split("/")[-1]
authorizer = self.apigatewayv2_backend.get_authorizer(api_id, authorizer_id) authorizer = self.apigatewayv2_backend.get_authorizer(api_id, authorizer_id)
return 200, {}, json.dumps(authorizer.to_json()) return 200, {}, json.dumps(authorizer.to_json())
def update_authorizer(self) -> RESPONSE_TYPE: def update_authorizer(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
authorizer_id = self.path.split("/")[-1] authorizer_id = self.path.split("/")[-1]
params = json.loads(self.body) params = json.loads(self.body)
@ -337,12 +334,12 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(authorizer.to_json()) return 200, {}, json.dumps(authorizer.to_json())
def delete_cors_configuration(self) -> RESPONSE_TYPE: def delete_cors_configuration(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-2] api_id = self.path.split("/")[-2]
self.apigatewayv2_backend.delete_cors_configuration(api_id) self.apigatewayv2_backend.delete_cors_configuration(api_id)
return 200, {}, "{}" return 200, {}, "{}"
def create_model(self) -> RESPONSE_TYPE: def create_model(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-2] api_id = self.path.split("/")[-2]
params = json.loads(self.body) params = json.loads(self.body)
@ -355,21 +352,21 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(model.to_json()) return 200, {}, json.dumps(model.to_json())
def delete_model(self) -> RESPONSE_TYPE: def delete_model(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
model_id = self.path.split("/")[-1] model_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_model(api_id, model_id) self.apigatewayv2_backend.delete_model(api_id, model_id)
return 200, {}, "{}" return 200, {}, "{}"
def get_model(self) -> RESPONSE_TYPE: def get_model(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
model_id = self.path.split("/")[-1] model_id = self.path.split("/")[-1]
model = self.apigatewayv2_backend.get_model(api_id, model_id) model = self.apigatewayv2_backend.get_model(api_id, model_id)
return 200, {}, json.dumps(model.to_json()) return 200, {}, json.dumps(model.to_json())
def update_model(self) -> RESPONSE_TYPE: def update_model(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
model_id = self.path.split("/")[-1] model_id = self.path.split("/")[-1]
params = json.loads(self.body) params = json.loads(self.body)
@ -389,24 +386,24 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(model.to_json()) return 200, {}, json.dumps(model.to_json())
def get_tags(self) -> RESPONSE_TYPE: def get_tags(self) -> TYPE_RESPONSE:
resource_arn = unquote(self.path.split("/tags/")[1]) resource_arn = unquote(self.path.split("/tags/")[1])
tags = self.apigatewayv2_backend.get_tags(resource_arn) tags = self.apigatewayv2_backend.get_tags(resource_arn)
return 200, {}, json.dumps({"tags": tags}) return 200, {}, json.dumps({"tags": tags})
def tag_resource(self) -> RESPONSE_TYPE: def tag_resource(self) -> TYPE_RESPONSE:
resource_arn = unquote(self.path.split("/tags/")[1]) resource_arn = unquote(self.path.split("/tags/")[1])
tags = json.loads(self.body).get("tags", {}) tags = json.loads(self.body).get("tags", {})
self.apigatewayv2_backend.tag_resource(resource_arn, tags) self.apigatewayv2_backend.tag_resource(resource_arn, tags)
return 201, {}, "{}" return 201, {}, "{}"
def untag_resource(self) -> RESPONSE_TYPE: def untag_resource(self) -> TYPE_RESPONSE:
resource_arn = unquote(self.path.split("/tags/")[1]) resource_arn = unquote(self.path.split("/tags/")[1])
tag_keys = self.querystring.get("tagKeys") or [] tag_keys = self.querystring.get("tagKeys") or []
self.apigatewayv2_backend.untag_resource(resource_arn, tag_keys) self.apigatewayv2_backend.untag_resource(resource_arn, tag_keys)
return 200, {}, "{}" return 200, {}, "{}"
def create_route(self) -> RESPONSE_TYPE: def create_route(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-2] api_id = self.path.split("/")[-2]
params = json.loads(self.body) params = json.loads(self.body)
api_key_required: bool = params.get("apiKeyRequired", False) api_key_required: bool = params.get("apiKeyRequired", False)
@ -438,13 +435,13 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 201, {}, json.dumps(route.to_json()) return 201, {}, json.dumps(route.to_json())
def delete_route(self) -> RESPONSE_TYPE: def delete_route(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
route_id = self.path.split("/")[-1] route_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_route(api_id=api_id, route_id=route_id) self.apigatewayv2_backend.delete_route(api_id=api_id, route_id=route_id)
return 200, {}, "{}" return 200, {}, "{}"
def delete_route_request_parameter(self) -> RESPONSE_TYPE: def delete_route_request_parameter(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-5] api_id = self.path.split("/")[-5]
route_id = self.path.split("/")[-3] route_id = self.path.split("/")[-3]
request_param = self.path.split("/")[-1] request_param = self.path.split("/")[-1]
@ -453,18 +450,18 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, "{}" return 200, {}, "{}"
def get_route(self) -> RESPONSE_TYPE: def get_route(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
route_id = self.path.split("/")[-1] route_id = self.path.split("/")[-1]
api = self.apigatewayv2_backend.get_route(api_id=api_id, route_id=route_id) api = self.apigatewayv2_backend.get_route(api_id=api_id, route_id=route_id)
return 200, {}, json.dumps(api.to_json()) return 200, {}, json.dumps(api.to_json())
def get_routes(self) -> RESPONSE_TYPE: def get_routes(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-2] api_id = self.path.split("/")[-2]
apis = self.apigatewayv2_backend.get_routes(api_id=api_id) apis = self.apigatewayv2_backend.get_routes(api_id=api_id)
return 200, {}, json.dumps({"items": [api.to_json() for api in apis]}) return 200, {}, json.dumps({"items": [api.to_json() for api in apis]})
def update_route(self) -> RESPONSE_TYPE: def update_route(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
route_id = self.path.split("/")[-1] route_id = self.path.split("/")[-1]
@ -499,7 +496,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(api.to_json()) return 200, {}, json.dumps(api.to_json())
def create_route_response(self) -> RESPONSE_TYPE: def create_route_response(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-4] api_id = self.path.split("/")[-4]
route_id = self.path.split("/")[-2] route_id = self.path.split("/")[-2]
params = json.loads(self.body) params = json.loads(self.body)
@ -516,7 +513,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(route_response.to_json()) return 200, {}, json.dumps(route_response.to_json())
def delete_route_response(self) -> RESPONSE_TYPE: def delete_route_response(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-5] api_id = self.path.split("/")[-5]
route_id = self.path.split("/")[-3] route_id = self.path.split("/")[-3]
route_response_id = self.path.split("/")[-1] route_response_id = self.path.split("/")[-1]
@ -526,7 +523,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, "{}" return 200, {}, "{}"
def get_route_response(self) -> RESPONSE_TYPE: def get_route_response(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-5] api_id = self.path.split("/")[-5]
route_id = self.path.split("/")[-3] route_id = self.path.split("/")[-3]
route_response_id = self.path.split("/")[-1] route_response_id = self.path.split("/")[-1]
@ -536,7 +533,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(route_response.to_json()) return 200, {}, json.dumps(route_response.to_json())
def create_integration(self) -> RESPONSE_TYPE: def create_integration(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-2] api_id = self.path.split("/")[-2]
params = json.loads(self.body) params = json.loads(self.body)
@ -579,7 +576,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(integration.to_json()) return 200, {}, json.dumps(integration.to_json())
def get_integration(self) -> RESPONSE_TYPE: def get_integration(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
integration_id = self.path.split("/")[-1] integration_id = self.path.split("/")[-1]
@ -588,13 +585,13 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(integration.to_json()) return 200, {}, json.dumps(integration.to_json())
def get_integrations(self) -> RESPONSE_TYPE: def get_integrations(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-2] api_id = self.path.split("/")[-2]
integrations = self.apigatewayv2_backend.get_integrations(api_id=api_id) integrations = self.apigatewayv2_backend.get_integrations(api_id=api_id)
return 200, {}, json.dumps({"items": [i.to_json() for i in integrations]}) return 200, {}, json.dumps({"items": [i.to_json() for i in integrations]})
def delete_integration(self) -> RESPONSE_TYPE: def delete_integration(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
integration_id = self.path.split("/")[-1] integration_id = self.path.split("/")[-1]
@ -603,7 +600,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, "{}" return 200, {}, "{}"
def update_integration(self) -> RESPONSE_TYPE: def update_integration(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
integration_id = self.path.split("/")[-1] integration_id = self.path.split("/")[-1]
@ -648,7 +645,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(integration.to_json()) return 200, {}, json.dumps(integration.to_json())
def create_integration_response(self) -> RESPONSE_TYPE: def create_integration_response(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-4] api_id = self.path.split("/")[-4]
int_id = self.path.split("/")[-2] int_id = self.path.split("/")[-2]
@ -669,7 +666,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(integration_response.to_json()) return 200, {}, json.dumps(integration_response.to_json())
def delete_integration_response(self) -> RESPONSE_TYPE: def delete_integration_response(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-5] api_id = self.path.split("/")[-5]
int_id = self.path.split("/")[-3] int_id = self.path.split("/")[-3]
int_res_id = self.path.split("/")[-1] int_res_id = self.path.split("/")[-1]
@ -679,7 +676,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, "{}" return 200, {}, "{}"
def get_integration_response(self) -> RESPONSE_TYPE: def get_integration_response(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-5] api_id = self.path.split("/")[-5]
int_id = self.path.split("/")[-3] int_id = self.path.split("/")[-3]
int_res_id = self.path.split("/")[-1] int_res_id = self.path.split("/")[-1]
@ -689,7 +686,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(int_response.to_json()) return 200, {}, json.dumps(int_response.to_json())
def get_integration_responses(self) -> RESPONSE_TYPE: def get_integration_responses(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-4] api_id = self.path.split("/")[-4]
int_id = self.path.split("/")[-2] int_id = self.path.split("/")[-2]
@ -698,7 +695,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps({"items": [res.to_json() for res in int_response]}) return 200, {}, json.dumps({"items": [res.to_json() for res in int_response]})
def update_integration_response(self) -> RESPONSE_TYPE: def update_integration_response(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-5] api_id = self.path.split("/")[-5]
int_id = self.path.split("/")[-3] int_id = self.path.split("/")[-3]
int_res_id = self.path.split("/")[-1] int_res_id = self.path.split("/")[-1]
@ -721,7 +718,7 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(integration_response.to_json()) return 200, {}, json.dumps(integration_response.to_json())
def create_vpc_link(self) -> RESPONSE_TYPE: def create_vpc_link(self) -> TYPE_RESPONSE:
params = json.loads(self.body) params = json.loads(self.body)
name = params.get("name") name = params.get("name")
@ -733,21 +730,21 @@ class ApiGatewayV2Response(BaseResponse):
) )
return 200, {}, json.dumps(vpc_link.to_json()) return 200, {}, json.dumps(vpc_link.to_json())
def delete_vpc_link(self) -> RESPONSE_TYPE: def delete_vpc_link(self) -> TYPE_RESPONSE:
vpc_link_id = self.path.split("/")[-1] vpc_link_id = self.path.split("/")[-1]
self.apigatewayv2_backend.delete_vpc_link(vpc_link_id) self.apigatewayv2_backend.delete_vpc_link(vpc_link_id)
return 200, {}, "{}" return 200, {}, "{}"
def get_vpc_link(self) -> RESPONSE_TYPE: def get_vpc_link(self) -> TYPE_RESPONSE:
vpc_link_id = self.path.split("/")[-1] vpc_link_id = self.path.split("/")[-1]
vpc_link = self.apigatewayv2_backend.get_vpc_link(vpc_link_id) vpc_link = self.apigatewayv2_backend.get_vpc_link(vpc_link_id)
return 200, {}, json.dumps(vpc_link.to_json()) return 200, {}, json.dumps(vpc_link.to_json())
def get_vpc_links(self) -> RESPONSE_TYPE: def get_vpc_links(self) -> TYPE_RESPONSE:
vpc_links = self.apigatewayv2_backend.get_vpc_links() vpc_links = self.apigatewayv2_backend.get_vpc_links()
return 200, {}, json.dumps({"items": [link.to_json() for link in vpc_links]}) return 200, {}, json.dumps({"items": [link.to_json() for link in vpc_links]})
def update_vpc_link(self) -> RESPONSE_TYPE: def update_vpc_link(self) -> TYPE_RESPONSE:
vpc_link_id = self.path.split("/")[-1] vpc_link_id = self.path.split("/")[-1]
params = json.loads(self.body) params = json.loads(self.body)
name = params.get("name") name = params.get("name")

2
moto/appsync/exceptions.py
View File

@ -9,6 +9,6 @@ class AppSyncExceptions(JsonRESTError):
class GraphqlAPINotFound(AppSyncExceptions): class GraphqlAPINotFound(AppSyncExceptions):
code = 404 code = 404
def __init__(self, api_id): def __init__(self, api_id: str):
super().__init__("NotFoundException", f"GraphQL API {api_id} not found.") super().__init__("NotFoundException", f"GraphQL API {api_id} not found.")
self.description = json.dumps({"message": self.message}) self.description = json.dumps({"message": self.message})

215
moto/appsync/models.py
View File

@ -1,5 +1,6 @@
import base64 import base64
from datetime import timedelta, datetime, timezone from datetime import timedelta, datetime, timezone
from typing import Any, Dict, Iterable, List, Optional, Tuple
from moto.core import BaseBackend, BaseModel from moto.core import BaseBackend, BaseModel
from moto.core.utils import BackendDict, unix_time from moto.core.utils import BackendDict, unix_time
from moto.moto_api._internal import mock_random from moto.moto_api._internal import mock_random
@ -9,16 +10,16 @@ from .exceptions import GraphqlAPINotFound
class GraphqlSchema(BaseModel): class GraphqlSchema(BaseModel):
def __init__(self, definition): def __init__(self, definition: Any):
self.definition = definition self.definition = definition
# [graphql.language.ast.ObjectTypeDefinitionNode, ..] # [graphql.language.ast.ObjectTypeDefinitionNode, ..]
self.types = [] self.types: List[Any] = []
self.status = "PROCESSING" self.status = "PROCESSING"
self.parse_error = None self.parse_error: Optional[str] = None
self._parse_graphql_definition() self._parse_graphql_definition()
def get_type(self, name): def get_type(self, name: str) -> Optional[Dict[str, Any]]: # type: ignore[return]
for graphql_type in self.types: for graphql_type in self.types:
if graphql_type.name.value == name: if graphql_type.name.value == name:
return { return {
@ -30,10 +31,10 @@ class GraphqlSchema(BaseModel):
"definition": "NotYetImplemented", "definition": "NotYetImplemented",
} }
def get_status(self): def get_status(self) -> Tuple[str, Optional[str]]:
return self.status, self.parse_error return self.status, self.parse_error
def _parse_graphql_definition(self): def _parse_graphql_definition(self) -> None:
try: try:
from graphql import parse from graphql import parse
from graphql.language.ast import ObjectTypeDefinitionNode from graphql.language.ast import ObjectTypeDefinitionNode
@ -49,19 +50,47 @@ class GraphqlSchema(BaseModel):
self.parse_error = str(e) self.parse_error = str(e)
class GraphqlAPIKey(BaseModel):
def __init__(self, description: str, expires: Optional[datetime]):
self.key_id = str(mock_random.uuid4())[0:6]
self.description = description
self.expires = expires
if not self.expires:
default_expiry = datetime.now(timezone.utc)
default_expiry = default_expiry.replace(
minute=0, second=0, microsecond=0, tzinfo=None
)
default_expiry = default_expiry + timedelta(days=7)
self.expires = unix_time(default_expiry)
def update(self, description: Optional[str], expires: Optional[datetime]) -> None:
if description:
self.description = description
if expires:
self.expires = expires
def to_json(self) -> Dict[str, Any]:
return {
"id": self.key_id,
"description": self.description,
"expires": self.expires,
"deletes": self.expires,
}
class GraphqlAPI(BaseModel): class GraphqlAPI(BaseModel):
def __init__( def __init__(
self, self,
account_id, account_id: str,
region, region: str,
name, name: str,
authentication_type, authentication_type: str,
additional_authentication_providers, additional_authentication_providers: Optional[List[str]],
log_config, log_config: str,
xray_enabled, xray_enabled: str,
user_pool_config, user_pool_config: str,
open_id_connect_config, open_id_connect_config: str,
lambda_authorizer_config, lambda_authorizer_config: str,
): ):
self.region = region self.region = region
self.name = name self.name = name
@ -75,21 +104,21 @@ class GraphqlAPI(BaseModel):
self.xray_enabled = xray_enabled self.xray_enabled = xray_enabled
self.arn = f"arn:aws:appsync:{self.region}:{account_id}:apis/{self.api_id}" self.arn = f"arn:aws:appsync:{self.region}:{account_id}:apis/{self.api_id}"
self.graphql_schema = None self.graphql_schema: Optional[GraphqlSchema] = None
self.api_keys = dict() self.api_keys: Dict[str, GraphqlAPIKey] = dict()
def update( def update(
self, self,
name, name: str,
additional_authentication_providers, additional_authentication_providers: Optional[List[str]],
authentication_type, authentication_type: str,
lambda_authorizer_config, lambda_authorizer_config: str,
log_config, log_config: str,
open_id_connect_config, open_id_connect_config: str,
user_pool_config, user_pool_config: str,
xray_enabled, xray_enabled: str,
): ) -> None:
if name: if name:
self.name = name self.name = name
if additional_authentication_providers: if additional_authentication_providers:
@ -109,36 +138,40 @@ class GraphqlAPI(BaseModel):
if xray_enabled is not None: if xray_enabled is not None:
self.xray_enabled = xray_enabled self.xray_enabled = xray_enabled
def create_api_key(self, description, expires): def create_api_key(
self, description: str, expires: Optional[datetime]
) -> GraphqlAPIKey:
api_key = GraphqlAPIKey(description, expires) api_key = GraphqlAPIKey(description, expires)
self.api_keys[api_key.key_id] = api_key self.api_keys[api_key.key_id] = api_key
return api_key return api_key
def list_api_keys(self): def list_api_keys(self) -> Iterable[GraphqlAPIKey]:
return self.api_keys.values() return self.api_keys.values()
def delete_api_key(self, api_key_id): def delete_api_key(self, api_key_id: str) -> None:
self.api_keys.pop(api_key_id) self.api_keys.pop(api_key_id)
def update_api_key(self, api_key_id, description, expires): def update_api_key(
self, api_key_id: str, description: str, expires: Optional[datetime]
) -> GraphqlAPIKey:
api_key = self.api_keys[api_key_id] api_key = self.api_keys[api_key_id]
api_key.update(description, expires) api_key.update(description, expires)
return api_key return api_key
def start_schema_creation(self, definition): def start_schema_creation(self, definition: str) -> None:
graphql_definition = base64.b64decode(definition).decode("utf-8") graphql_definition = base64.b64decode(definition).decode("utf-8")
self.graphql_schema = GraphqlSchema(graphql_definition) self.graphql_schema = GraphqlSchema(graphql_definition)
def get_schema_status(self): def get_schema_status(self) -> Any:
return self.graphql_schema.get_status() return self.graphql_schema.get_status() # type: ignore[union-attr]
def get_type(self, type_name, type_format): def get_type(self, type_name: str, type_format: str) -> Any:
graphql_type = self.graphql_schema.get_type(type_name) graphql_type = self.graphql_schema.get_type(type_name) # type: ignore[union-attr]
graphql_type["format"] = type_format graphql_type["format"] = type_format # type: ignore[index]
return graphql_type return graphql_type
def to_json(self): def to_json(self) -> Dict[str, Any]:
return { return {
"name": self.name, "name": self.name,
"apiId": self.api_id, "apiId": self.api_id,
@ -154,54 +187,26 @@ class GraphqlAPI(BaseModel):
} }
class GraphqlAPIKey(BaseModel):
def __init__(self, description, expires):
self.key_id = str(mock_random.uuid4())[0:6]
self.description = description
self.expires = expires
if not self.expires:
default_expiry = datetime.now(timezone.utc)
default_expiry = default_expiry.replace(
minute=0, second=0, microsecond=0, tzinfo=None
)
default_expiry = default_expiry + timedelta(days=7)
self.expires = unix_time(default_expiry)
def update(self, description, expires):
if description:
self.description = description
if expires:
self.expires = expires
def to_json(self):
return {
"id": self.key_id,
"description": self.description,
"expires": self.expires,
"deletes": self.expires,
}
class AppSyncBackend(BaseBackend): class AppSyncBackend(BaseBackend):
"""Implementation of AppSync APIs.""" """Implementation of AppSync APIs."""
def __init__(self, region_name, account_id): def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id) super().__init__(region_name, account_id)
self.graphql_apis = dict() self.graphql_apis: Dict[str, GraphqlAPI] = dict()
self.tagger = TaggingService() self.tagger = TaggingService()
def create_graphql_api( def create_graphql_api(
self, self,
name, name: str,
log_config, log_config: str,
authentication_type, authentication_type: str,
user_pool_config, user_pool_config: str,
open_id_connect_config, open_id_connect_config: str,
additional_authentication_providers, additional_authentication_providers: Optional[List[str]],
xray_enabled, xray_enabled: str,
lambda_authorizer_config, lambda_authorizer_config: str,
tags, tags: Dict[str, str],
): ) -> GraphqlAPI:
graphql_api = GraphqlAPI( graphql_api = GraphqlAPI(
account_id=self.account_id, account_id=self.account_id,
region=self.region_name, region=self.region_name,
@ -222,16 +227,16 @@ class AppSyncBackend(BaseBackend):
def update_graphql_api( def update_graphql_api(
self, self,
api_id, api_id: str,
name, name: str,
log_config, log_config: str,
authentication_type, authentication_type: str,
user_pool_config, user_pool_config: str,
open_id_connect_config, open_id_connect_config: str,
additional_authentication_providers, additional_authentication_providers: Optional[List[str]],
xray_enabled, xray_enabled: str,
lambda_authorizer_config, lambda_authorizer_config: str,
): ) -> GraphqlAPI:
graphql_api = self.graphql_apis[api_id] graphql_api = self.graphql_apis[api_id]
graphql_api.update( graphql_api.update(
name, name,
@ -245,27 +250,29 @@ class AppSyncBackend(BaseBackend):
) )
return graphql_api return graphql_api
def get_graphql_api(self, api_id): def get_graphql_api(self, api_id: str) -> GraphqlAPI:
if api_id not in self.graphql_apis: if api_id not in self.graphql_apis:
raise GraphqlAPINotFound(api_id) raise GraphqlAPINotFound(api_id)
return self.graphql_apis[api_id] return self.graphql_apis[api_id]
def delete_graphql_api(self, api_id): def delete_graphql_api(self, api_id: str) -> None:
self.graphql_apis.pop(api_id) self.graphql_apis.pop(api_id)
def list_graphql_apis(self): def list_graphql_apis(self) -> Iterable[GraphqlAPI]:
""" """
Pagination or the maxResults-parameter have not yet been implemented. Pagination or the maxResults-parameter have not yet been implemented.
""" """
return self.graphql_apis.values() return self.graphql_apis.values()
def create_api_key(self, api_id, description, expires): def create_api_key(
self, api_id: str, description: str, expires: Optional[datetime]
) -> GraphqlAPIKey:
return self.graphql_apis[api_id].create_api_key(description, expires) return self.graphql_apis[api_id].create_api_key(description, expires)
def delete_api_key(self, api_id, api_key_id): def delete_api_key(self, api_id: str, api_key_id: str) -> None:
self.graphql_apis[api_id].delete_api_key(api_key_id) self.graphql_apis[api_id].delete_api_key(api_key_id)
def list_api_keys(self, api_id): def list_api_keys(self, api_id: str) -> Iterable[GraphqlAPIKey]:
""" """
Pagination or the maxResults-parameter have not yet been implemented. Pagination or the maxResults-parameter have not yet been implemented.
""" """
@ -274,30 +281,36 @@ class AppSyncBackend(BaseBackend):
else: else:
return [] return []
def update_api_key(self, api_id, api_key_id, description, expires): def update_api_key(
self,
api_id: str,
api_key_id: str,
description: str,
expires: Optional[datetime],
) -> GraphqlAPIKey:
return self.graphql_apis[api_id].update_api_key( return self.graphql_apis[api_id].update_api_key(
api_key_id, description, expires api_key_id, description, expires
) )
def start_schema_creation(self, api_id, definition): def start_schema_creation(self, api_id: str, definition: str) -> str:
self.graphql_apis[api_id].start_schema_creation(definition) self.graphql_apis[api_id].start_schema_creation(definition)
return "PROCESSING" return "PROCESSING"
def get_schema_creation_status(self, api_id): def get_schema_creation_status(self, api_id: str) -> Any:
return self.graphql_apis[api_id].get_schema_status() return self.graphql_apis[api_id].get_schema_status()
def tag_resource(self, resource_arn, tags): def tag_resource(self, resource_arn: str, tags: Dict[str, str]) -> None:
self.tagger.tag_resource( self.tagger.tag_resource(
resource_arn, TaggingService.convert_dict_to_tags_input(tags) resource_arn, TaggingService.convert_dict_to_tags_input(tags)
) )
def untag_resource(self, resource_arn, tag_keys): def untag_resource(self, resource_arn: str, tag_keys: List[str]) -> None:
self.tagger.untag_resource_using_names(resource_arn, tag_keys) self.tagger.untag_resource_using_names(resource_arn, tag_keys)
def list_tags_for_resource(self, resource_arn): def list_tags_for_resource(self, resource_arn: str) -> Dict[str, str]:
return self.tagger.get_tag_dict_for_resource(resource_arn) return self.tagger.get_tag_dict_for_resource(resource_arn)
def get_type(self, api_id, type_name, type_format): def get_type(self, api_id: str, type_name: str, type_format: str) -> Any:
return self.graphql_apis[api_id].get_type(type_name, type_format) return self.graphql_apis[api_id].get_type(type_name, type_format)

59
moto/appsync/responses.py
View File

@ -1,30 +1,31 @@
"""Handles incoming appsync requests, invokes methods, returns responses.""" """Handles incoming appsync requests, invokes methods, returns responses."""
import json import json
from moto.core.responses import BaseResponse from moto.core.responses import BaseResponse, TYPE_RESPONSE
from typing import Any
from urllib.parse import unquote from urllib.parse import unquote
from .models import appsync_backends from .models import appsync_backends, AppSyncBackend
class AppSyncResponse(BaseResponse): class AppSyncResponse(BaseResponse):
"""Handler for AppSync requests and responses.""" """Handler for AppSync requests and responses."""
def __init__(self): def __init__(self) -> None:
super().__init__(service_name="appsync") super().__init__(service_name="appsync")
@property @property
def appsync_backend(self): def appsync_backend(self) -> AppSyncBackend:
"""Return backend instance specific for this region.""" """Return backend instance specific for this region."""
return appsync_backends[self.current_account][self.region] return appsync_backends[self.current_account][self.region]
def graph_ql(self, request, full_url, headers): def graph_ql(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if request.method == "POST": if request.method == "POST":
return self.create_graphql_api() return self.create_graphql_api()
if request.method == "GET": if request.method == "GET":
return self.list_graphql_apis() return self.list_graphql_apis()
def graph_ql_individual(self, request, full_url, headers): def graph_ql_individual(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if request.method == "GET": if request.method == "GET":
return self.get_graphql_api() return self.get_graphql_api()
@ -33,29 +34,31 @@ class AppSyncResponse(BaseResponse):
if request.method == "POST": if request.method == "POST":
return self.update_graphql_api() return self.update_graphql_api()
def api_key(self, request, full_url, headers): def api_key(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if request.method == "POST": if request.method == "POST":
return self.create_api_key() return self.create_api_key()
if request.method == "GET": if request.method == "GET":
return self.list_api_keys() return self.list_api_keys()
def schemacreation(self, request, full_url, headers): def schemacreation(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if request.method == "POST": if request.method == "POST":
return self.start_schema_creation() return self.start_schema_creation()
if request.method == "GET": if request.method == "GET":
return self.get_schema_creation_status() return self.get_schema_creation_status()
def api_key_individual(self, request, full_url, headers): def api_key_individual(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if request.method == "DELETE": if request.method == "DELETE":
return self.delete_api_key() return self.delete_api_key()
if request.method == "POST": if request.method == "POST":
return self.update_api_key() return self.update_api_key()
def tags(self, request, full_url, headers): def tags(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
print("tags")
print(request.method)
if request.method == "POST": if request.method == "POST":
return self.tag_resource() return self.tag_resource()
if request.method == "DELETE": if request.method == "DELETE":
@ -63,12 +66,12 @@ class AppSyncResponse(BaseResponse):
if request.method == "GET": if request.method == "GET":
return self.list_tags_for_resource() return self.list_tags_for_resource()
def types(self, request, full_url, headers): def types(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if request.method == "GET": if request.method == "GET":
return self.get_type() return self.get_type()
def create_graphql_api(self): def create_graphql_api(self) -> TYPE_RESPONSE:
params = json.loads(self.body) params = json.loads(self.body)
name = params.get("name") name = params.get("name")
log_config = params.get("logConfig") log_config = params.get("logConfig")
@ -96,7 +99,7 @@ class AppSyncResponse(BaseResponse):
response["tags"] = self.appsync_backend.list_tags_for_resource(graphql_api.arn) response["tags"] = self.appsync_backend.list_tags_for_resource(graphql_api.arn)
return 200, {}, json.dumps(dict(graphqlApi=response)) return 200, {}, json.dumps(dict(graphqlApi=response))
def get_graphql_api(self): def get_graphql_api(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-1] api_id = self.path.split("/")[-1]
graphql_api = self.appsync_backend.get_graphql_api(api_id=api_id) graphql_api = self.appsync_backend.get_graphql_api(api_id=api_id)
@ -104,12 +107,12 @@ class AppSyncResponse(BaseResponse):
response["tags"] = self.appsync_backend.list_tags_for_resource(graphql_api.arn) response["tags"] = self.appsync_backend.list_tags_for_resource(graphql_api.arn)
return 200, {}, json.dumps(dict(graphqlApi=response)) return 200, {}, json.dumps(dict(graphqlApi=response))
def delete_graphql_api(self): def delete_graphql_api(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-1] api_id = self.path.split("/")[-1]
self.appsync_backend.delete_graphql_api(api_id=api_id) self.appsync_backend.delete_graphql_api(api_id=api_id)
return 200, {}, json.dumps(dict()) return 200, {}, json.dumps(dict())
def update_graphql_api(self): def update_graphql_api(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-1] api_id = self.path.split("/")[-1]
params = json.loads(self.body) params = json.loads(self.body)
@ -137,7 +140,7 @@ class AppSyncResponse(BaseResponse):
) )
return 200, {}, json.dumps(dict(graphqlApi=api.to_json())) return 200, {}, json.dumps(dict(graphqlApi=api.to_json()))
def list_graphql_apis(self): def list_graphql_apis(self) -> TYPE_RESPONSE:
graphql_apis = self.appsync_backend.list_graphql_apis() graphql_apis = self.appsync_backend.list_graphql_apis()
return ( return (
200, 200,
@ -145,7 +148,7 @@ class AppSyncResponse(BaseResponse):
json.dumps(dict(graphqlApis=[api.to_json() for api in graphql_apis])), json.dumps(dict(graphqlApis=[api.to_json() for api in graphql_apis])),
) )
def create_api_key(self): def create_api_key(self) -> TYPE_RESPONSE:
params = json.loads(self.body) params = json.loads(self.body)
# /v1/apis/[api_id]/apikeys # /v1/apis/[api_id]/apikeys
api_id = self.path.split("/")[-2] api_id = self.path.split("/")[-2]
@ -156,19 +159,19 @@ class AppSyncResponse(BaseResponse):
) )
return 200, {}, json.dumps(dict(apiKey=api_key.to_json())) return 200, {}, json.dumps(dict(apiKey=api_key.to_json()))
def delete_api_key(self): def delete_api_key(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
api_key_id = self.path.split("/")[-1] api_key_id = self.path.split("/")[-1]
self.appsync_backend.delete_api_key(api_id=api_id, api_key_id=api_key_id) self.appsync_backend.delete_api_key(api_id=api_id, api_key_id=api_key_id)
return 200, {}, json.dumps(dict()) return 200, {}, json.dumps(dict())
def list_api_keys(self): def list_api_keys(self) -> TYPE_RESPONSE:
# /v1/apis/[api_id]/apikeys # /v1/apis/[api_id]/apikeys
api_id = self.path.split("/")[-2] api_id = self.path.split("/")[-2]
api_keys = self.appsync_backend.list_api_keys(api_id=api_id) api_keys = self.appsync_backend.list_api_keys(api_id=api_id)
return 200, {}, json.dumps(dict(apiKeys=[key.to_json() for key in api_keys])) return 200, {}, json.dumps(dict(apiKeys=[key.to_json() for key in api_keys]))
def update_api_key(self): def update_api_key(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-3] api_id = self.path.split("/")[-3]
api_key_id = self.path.split("/")[-1] api_key_id = self.path.split("/")[-1]
params = json.loads(self.body) params = json.loads(self.body)
@ -182,7 +185,7 @@ class AppSyncResponse(BaseResponse):
) )
return 200, {}, json.dumps(dict(apiKey=api_key.to_json())) return 200, {}, json.dumps(dict(apiKey=api_key.to_json()))
def start_schema_creation(self): def start_schema_creation(self) -> TYPE_RESPONSE:
params = json.loads(self.body) params = json.loads(self.body)
api_id = self.path.split("/")[-2] api_id = self.path.split("/")[-2]
definition = params.get("definition") definition = params.get("definition")
@ -191,19 +194,19 @@ class AppSyncResponse(BaseResponse):
) )
return 200, {}, json.dumps({"status": status}) return 200, {}, json.dumps({"status": status})
def get_schema_creation_status(self): def get_schema_creation_status(self) -> TYPE_RESPONSE:
api_id = self.path.split("/")[-2] api_id = self.path.split("/")[-2]
status, details = self.appsync_backend.get_schema_creation_status(api_id=api_id) status, details = self.appsync_backend.get_schema_creation_status(api_id=api_id)
return 200, {}, json.dumps(dict(status=status, details=details)) return 200, {}, json.dumps(dict(status=status, details=details))
def tag_resource(self): def tag_resource(self) -> TYPE_RESPONSE:
resource_arn = self._extract_arn_from_path() resource_arn = self._extract_arn_from_path()
params = json.loads(self.body) params = json.loads(self.body)
tags = params.get("tags") tags = params.get("tags")
self.appsync_backend.tag_resource(resource_arn=resource_arn, tags=tags) self.appsync_backend.tag_resource(resource_arn=resource_arn, tags=tags)
return 200, {}, json.dumps(dict()) return 200, {}, json.dumps(dict())
def untag_resource(self): def untag_resource(self) -> TYPE_RESPONSE:
resource_arn = self._extract_arn_from_path() resource_arn = self._extract_arn_from_path()
tag_keys = self.querystring.get("tagKeys", []) tag_keys = self.querystring.get("tagKeys", [])
self.appsync_backend.untag_resource( self.appsync_backend.untag_resource(
@ -211,20 +214,20 @@ class AppSyncResponse(BaseResponse):
) )
return 200, {}, json.dumps(dict()) return 200, {}, json.dumps(dict())
def list_tags_for_resource(self): def list_tags_for_resource(self) -> TYPE_RESPONSE:
resource_arn = self._extract_arn_from_path() resource_arn = self._extract_arn_from_path()
tags = self.appsync_backend.list_tags_for_resource(resource_arn=resource_arn) tags = self.appsync_backend.list_tags_for_resource(resource_arn=resource_arn)
return 200, {}, json.dumps(dict(tags=tags)) return 200, {}, json.dumps(dict(tags=tags))
def _extract_arn_from_path(self): def _extract_arn_from_path(self) -> str:
# /v1/tags/arn_that_may_contain_a_slash # /v1/tags/arn_that_may_contain_a_slash
path = unquote(self.path) path = unquote(self.path)
return "/".join(path.split("/")[3:]) return "/".join(path.split("/")[3:])
def get_type(self): def get_type(self) -> TYPE_RESPONSE:
api_id = unquote(self.path.split("/")[-3]) api_id = unquote(self.path.split("/")[-3])
type_name = self.path.split("/")[-1] type_name = self.path.split("/")[-1]
type_format = self.querystring.get("format")[0] type_format = self.querystring.get("format")[0] # type: ignore[index]
graphql_type = self.appsync_backend.get_type( graphql_type = self.appsync_backend.get_type(
api_id=api_id, type_name=type_name, type_format=type_format api_id=api_id, type_name=type_name, type_format=type_format
) )

4
moto/core/responses.py
View File

@ -15,7 +15,7 @@ from moto.core.exceptions import DryRunClientError
from moto.core.utils import camelcase_to_underscores, method_names_from_class from moto.core.utils import camelcase_to_underscores, method_names_from_class
from moto.utilities.utils import load_resource from moto.utilities.utils import load_resource
from jinja2 import Environment, DictLoader from jinja2 import Environment, DictLoader
from typing import Dict, List, Union, Any, Optional from typing import Dict, List, Union, Any, Optional, Tuple
from urllib.parse import parse_qs, parse_qsl, urlparse from urllib.parse import parse_qs, parse_qsl, urlparse
from werkzeug.exceptions import HTTPException from werkzeug.exceptions import HTTPException
from xml.dom.minidom import parseString as parseXML from xml.dom.minidom import parseString as parseXML
@ -25,6 +25,8 @@ log = logging.getLogger(__name__)
JINJA_ENVS = {} JINJA_ENVS = {}
TYPE_RESPONSE = Tuple[int, Dict[str, str], str]
def _decode_dict(d): def _decode_dict(d):
decoded = OrderedDict() decoded = OrderedDict()

2
setup.cfg
View File

@ -18,7 +18,7 @@ disable = W,C,R,E
enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import
[mypy] [mypy]
files= moto/acm,moto/amp,moto/apigateway,moto/apigatewayv2,moto/applicationautoscaling/ files= moto/acm,moto/amp,moto/apigateway,moto/apigatewayv2,moto/applicationautoscaling/,moto/appsync
show_column_numbers=True show_column_numbers=True
show_error_codes = True show_error_codes = True
disable_error_code=abstract disable_error_code=abstract