From 047c1a6b7e4b40e27c323894017a26d035c32e9d Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Sun, 16 Oct 2022 18:14:15 +0000 Subject: [PATCH] TechDebt: MyPy AppSync (#5570) --- moto/amp/responses.py | 2 +- moto/apigateway/responses.py | 88 +++++++------- moto/apigatewayv2/responses.py | 129 ++++++++++---------- moto/appsync/exceptions.py | 2 +- moto/appsync/models.py | 215 +++++++++++++++++---------------- moto/appsync/responses.py | 59 ++++----- moto/core/responses.py | 4 +- setup.cfg | 2 +- 8 files changed, 257 insertions(+), 244 deletions(-) diff --git a/moto/amp/responses.py b/moto/amp/responses.py index 08f43a8bd..cab5cab32 100644 --- a/moto/amp/responses.py +++ b/moto/amp/responses.py @@ -10,7 +10,7 @@ from urllib.parse import unquote class PrometheusServiceResponse(BaseResponse): """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) if request.method == "GET": return self.list_tags_for_resource() diff --git a/moto/apigateway/responses.py b/moto/apigateway/responses.py index 4595ef1f7..48e019cd9 100644 --- a/moto/apigateway/responses.py +++ b/moto/apigateway/responses.py @@ -1,9 +1,9 @@ import json -from typing import Any, Dict, List, Tuple +from typing import Any, Dict, List from urllib.parse import unquote 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 .utils import deserialize_body from .exceptions import InvalidRequestInput @@ -12,14 +12,12 @@ API_KEY_SOURCES = ["AUTHORIZER", "HEADER"] AUTHORIZER_TYPES = ["TOKEN", "REQUEST", "COGNITO_USER_POOLS"] ENDPOINT_CONFIGURATION_TYPES = ["PRIVATE", "EDGE", "REGIONAL"] -RESPONSE_TYPE = Tuple[int, Dict[str, str], str] - class APIGatewayResponse(BaseResponse): def __init__(self) -> None: 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["X-Amzn-Errortype"] = type_ return (status, headers, json.dumps({"__type": type_, "message": message})) @@ -28,7 +26,7 @@ class APIGatewayResponse(BaseResponse): def backend(self) -> APIGatewayBackend: 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: return self.error( "ValidationException", @@ -40,7 +38,7 @@ class APIGatewayResponse(BaseResponse): ).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: invalid_types = list( set(endpoint_configuration["types"]) - set(ENDPOINT_CONFIGURATION_TYPES) @@ -56,7 +54,7 @@ class APIGatewayResponse(BaseResponse): ).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) if self.method == "GET": @@ -100,7 +98,7 @@ class APIGatewayResponse(BaseResponse): 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: path = op["path"] if "apiKeySource" in path: @@ -109,7 +107,7 @@ class APIGatewayResponse(BaseResponse): def restapis_individual( self, request: Any, full_url: str, headers: Dict[str, str] - ) -> RESPONSE_TYPE: + ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) function_id = self.path.replace("/restapis/", "", 1).split("/")[0] @@ -135,7 +133,7 @@ class APIGatewayResponse(BaseResponse): 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) 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]}), ) - 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) if request.method == "PUT": return self.put_gateway_response() @@ -156,12 +154,12 @@ class APIGatewayResponse(BaseResponse): elif request.method == "DELETE": 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) if request.method == "GET": 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) function_id = self.path.replace("/restapis/", "", 1).split("/")[0] resource_id = self.path.split("/")[-1] @@ -179,7 +177,7 @@ class APIGatewayResponse(BaseResponse): def resource_methods( self, request: Any, full_url: str, headers: Dict[str, str] - ) -> RESPONSE_TYPE: + ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") function_id = url_path_parts[2] @@ -221,7 +219,7 @@ class APIGatewayResponse(BaseResponse): def resource_method_responses( self, request: Any, full_url: str, headers: Dict[str, str] - ) -> RESPONSE_TYPE: + ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") function_id = url_path_parts[2] @@ -253,7 +251,7 @@ class APIGatewayResponse(BaseResponse): return 204, {}, json.dumps(method_response) 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) url_path_parts = self.path.split("/") restapi_id = url_path_parts[2] @@ -303,7 +301,7 @@ class APIGatewayResponse(BaseResponse): authorizers = self.backend.get_authorizers(restapi_id) 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) url_path_parts = self.path.split("/") restapi_id = url_path_parts[2] @@ -323,7 +321,7 @@ class APIGatewayResponse(BaseResponse): ) 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) url_path_parts = self.path.split("/") restapi_id = url_path_parts[2] @@ -342,7 +340,7 @@ class APIGatewayResponse(BaseResponse): ) 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) url_path_parts = self.path.split("/") restapi_id = url_path_parts[2] @@ -361,7 +359,7 @@ class APIGatewayResponse(BaseResponse): self.backend.delete_authorizer(restapi_id, authorizer_id) 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) url_path_parts = self.path.split("/") function_id = url_path_parts[2] @@ -392,7 +390,7 @@ class APIGatewayResponse(BaseResponse): stages = self.backend.get_stages(function_id) 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) url_path_parts = self.path.split("/") function_id = url_path_parts[4] @@ -410,7 +408,7 @@ class APIGatewayResponse(BaseResponse): stage["tags"].pop(tag, None) 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) url_path_parts = self.path.split("/") function_id = url_path_parts[2] @@ -429,7 +427,7 @@ class APIGatewayResponse(BaseResponse): self.backend.delete_stage(function_id, stage_name) 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) url_path_parts = self.path.split("/") function_id = url_path_parts[2] @@ -479,7 +477,7 @@ class APIGatewayResponse(BaseResponse): ) 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) url_path_parts = self.path.split("/") function_id = url_path_parts[2] @@ -515,7 +513,7 @@ class APIGatewayResponse(BaseResponse): ) 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) function_id = self.path.replace("/restapis/", "", 1).split("/")[0] @@ -531,7 +529,7 @@ class APIGatewayResponse(BaseResponse): ) 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) url_path_parts = self.path.split("/") function_id = url_path_parts[2] @@ -544,7 +542,7 @@ class APIGatewayResponse(BaseResponse): deployment = self.backend.delete_deployment(function_id, deployment_id) 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) if self.method == "POST": @@ -558,7 +556,7 @@ class APIGatewayResponse(BaseResponse): def apikey_individual( self, request: Any, full_url: str, headers: Dict[str, str] - ) -> RESPONSE_TYPE: + ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") @@ -578,7 +576,7 @@ class APIGatewayResponse(BaseResponse): 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) if self.method == "POST": 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) 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) url_path_parts = self.path.split("/") @@ -607,7 +605,7 @@ class APIGatewayResponse(BaseResponse): ) 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) 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) 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) url_path_parts = self.path.split("/") @@ -636,7 +634,7 @@ class APIGatewayResponse(BaseResponse): self.backend.delete_usage_plan_key(usage_plan_id, key_id) 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) if self.method == "GET": @@ -672,7 +670,7 @@ class APIGatewayResponse(BaseResponse): def domain_name_induvidual( self, request: Any, full_url: str, headers: Dict[str, str] - ) -> RESPONSE_TYPE: + ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) 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 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) rest_api_id = self.path.replace("/restapis/", "", 1).split("/")[0] @@ -715,7 +713,7 @@ class APIGatewayResponse(BaseResponse): def model_induvidual( self, request: Any, full_url: str, headers: Dict[str, str] - ) -> RESPONSE_TYPE: + ) -> TYPE_RESPONSE: self.setup_class(request, full_url, headers) url_path_parts = self.path.split("/") rest_api_id = url_path_parts[2] @@ -726,7 +724,7 @@ class APIGatewayResponse(BaseResponse): return 200, {}, json.dumps(model_info) 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) url_path_parts = self.path.split("/") @@ -745,7 +743,7 @@ class APIGatewayResponse(BaseResponse): ) 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) @@ -768,7 +766,7 @@ class APIGatewayResponse(BaseResponse): ) 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) url_path_parts = self.path.split("/") 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) 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) if self.method == "GET": @@ -796,7 +794,7 @@ class APIGatewayResponse(BaseResponse): ) 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] response_type = self.path.split("/")[-1] params = json.loads(self.body) @@ -812,7 +810,7 @@ class APIGatewayResponse(BaseResponse): ) 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] response_type = self.path.split("/")[-1] response = self.backend.get_gateway_response( @@ -820,12 +818,12 @@ class APIGatewayResponse(BaseResponse): ) 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] responses = self.backend.get_gateway_responses(rest_api_id=rest_api_id) 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] response_type = self.path.split("/")[-1] self.backend.delete_gateway_response( diff --git a/moto/apigatewayv2/responses.py b/moto/apigatewayv2/responses.py index a9659c25e..403eaad18 100644 --- a/moto/apigatewayv2/responses.py +++ b/moto/apigatewayv2/responses.py @@ -1,17 +1,14 @@ """Handles incoming apigatewayv2 requests, invokes methods, returns responses.""" import json -from moto.core.responses import BaseResponse -from typing import Any, Tuple, Dict +from moto.core.responses import BaseResponse, TYPE_RESPONSE +from typing import Any from urllib.parse import unquote from .exceptions import UnknownProtocol from .models import apigatewayv2_backends, ApiGatewayV2Backend -RESPONSE_TYPE = Tuple[int, Dict[str, str], str] - - class ApiGatewayV2Response(BaseResponse): """Handler for ApiGatewayV2 requests and responses.""" @@ -23,7 +20,7 @@ class ApiGatewayV2Response(BaseResponse): """Return backend instance specific for this 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) if self.method == "POST": @@ -31,7 +28,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "GET": 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) if self.method == "GET": @@ -43,7 +40,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "DELETE": 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) if self.method == "DELETE": @@ -53,25 +50,25 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "PATCH": 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) if self.method == "POST": 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) if self.method == "DELETE": 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) if self.method == "DELETE": 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) if self.method == "DELETE": @@ -81,13 +78,13 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "PATCH": 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) if self.method == "POST": 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) if self.method == "DELETE": @@ -97,7 +94,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "PATCH": 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) if self.method == "GET": @@ -105,7 +102,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "POST": 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) if self.method == "DELETE": @@ -115,7 +112,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "PATCH": 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) if self.method == "GET": @@ -123,7 +120,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "POST": 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) if self.method == "DELETE": @@ -133,7 +130,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "PATCH": 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) if self.method == "GET": @@ -141,7 +138,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "POST": 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) if self.method == "DELETE": @@ -149,13 +146,13 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "GET": 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) if self.method == "POST": 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) if self.method == "POST": @@ -165,7 +162,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "DELETE": 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) if request.method == "DELETE": @@ -175,7 +172,7 @@ class ApiGatewayV2Response(BaseResponse): if request.method == "PATCH": 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) if request.method == "GET": @@ -183,7 +180,7 @@ class ApiGatewayV2Response(BaseResponse): if request.method == "POST": return self.create_vpc_link() - def create_api(self) -> RESPONSE_TYPE: + def create_api(self) -> TYPE_RESPONSE: params = json.loads(self.body) api_key_selection_expression = params.get("apiKeySelectionExpression") @@ -214,21 +211,21 @@ class ApiGatewayV2Response(BaseResponse): ) 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] self.apigatewayv2_backend.delete_api(api_id=api_id) return 200, {}, "{}" - def get_api(self) -> RESPONSE_TYPE: + def get_api(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-1] api = self.apigatewayv2_backend.get_api(api_id=api_id) 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() 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] params = json.loads(self.body) api_key_selection_expression = params.get("apiKeySelectionExpression") @@ -252,7 +249,7 @@ class ApiGatewayV2Response(BaseResponse): ) 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] params = json.loads(self.body) body = params.get("body") @@ -263,7 +260,7 @@ class ApiGatewayV2Response(BaseResponse): api = self.apigatewayv2_backend.reimport_api(api_id, body, fail_on_warnings) 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] params = json.loads(self.body) @@ -292,21 +289,21 @@ class ApiGatewayV2Response(BaseResponse): ) 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] authorizer_id = self.path.split("/")[-1] self.apigatewayv2_backend.delete_authorizer(api_id, authorizer_id) return 200, {}, "{}" - def get_authorizer(self) -> RESPONSE_TYPE: + def get_authorizer(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-3] authorizer_id = self.path.split("/")[-1] authorizer = self.apigatewayv2_backend.get_authorizer(api_id, authorizer_id) 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] authorizer_id = self.path.split("/")[-1] params = json.loads(self.body) @@ -337,12 +334,12 @@ class ApiGatewayV2Response(BaseResponse): ) 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] self.apigatewayv2_backend.delete_cors_configuration(api_id) return 200, {}, "{}" - def create_model(self) -> RESPONSE_TYPE: + def create_model(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-2] params = json.loads(self.body) @@ -355,21 +352,21 @@ class ApiGatewayV2Response(BaseResponse): ) 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] model_id = self.path.split("/")[-1] self.apigatewayv2_backend.delete_model(api_id, model_id) return 200, {}, "{}" - def get_model(self) -> RESPONSE_TYPE: + def get_model(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-3] model_id = self.path.split("/")[-1] model = self.apigatewayv2_backend.get_model(api_id, model_id) 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] model_id = self.path.split("/")[-1] params = json.loads(self.body) @@ -389,24 +386,24 @@ class ApiGatewayV2Response(BaseResponse): ) 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]) tags = self.apigatewayv2_backend.get_tags(resource_arn) 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]) tags = json.loads(self.body).get("tags", {}) self.apigatewayv2_backend.tag_resource(resource_arn, tags) return 201, {}, "{}" - def untag_resource(self) -> RESPONSE_TYPE: + def untag_resource(self) -> TYPE_RESPONSE: resource_arn = unquote(self.path.split("/tags/")[1]) tag_keys = self.querystring.get("tagKeys") or [] self.apigatewayv2_backend.untag_resource(resource_arn, tag_keys) return 200, {}, "{}" - def create_route(self) -> RESPONSE_TYPE: + def create_route(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-2] params = json.loads(self.body) api_key_required: bool = params.get("apiKeyRequired", False) @@ -438,13 +435,13 @@ class ApiGatewayV2Response(BaseResponse): ) 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] route_id = self.path.split("/")[-1] self.apigatewayv2_backend.delete_route(api_id=api_id, route_id=route_id) return 200, {}, "{}" - def delete_route_request_parameter(self) -> RESPONSE_TYPE: + def delete_route_request_parameter(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-5] route_id = self.path.split("/")[-3] request_param = self.path.split("/")[-1] @@ -453,18 +450,18 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, "{}" - def get_route(self) -> RESPONSE_TYPE: + def get_route(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-3] route_id = self.path.split("/")[-1] api = self.apigatewayv2_backend.get_route(api_id=api_id, route_id=route_id) 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] apis = self.apigatewayv2_backend.get_routes(api_id=api_id) 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] route_id = self.path.split("/")[-1] @@ -499,7 +496,7 @@ class ApiGatewayV2Response(BaseResponse): ) 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] route_id = self.path.split("/")[-2] params = json.loads(self.body) @@ -516,7 +513,7 @@ class ApiGatewayV2Response(BaseResponse): ) 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] route_id = self.path.split("/")[-3] route_response_id = self.path.split("/")[-1] @@ -526,7 +523,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, "{}" - def get_route_response(self) -> RESPONSE_TYPE: + def get_route_response(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-5] route_id = self.path.split("/")[-3] route_response_id = self.path.split("/")[-1] @@ -536,7 +533,7 @@ class ApiGatewayV2Response(BaseResponse): ) 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] params = json.loads(self.body) @@ -579,7 +576,7 @@ class ApiGatewayV2Response(BaseResponse): ) 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] integration_id = self.path.split("/")[-1] @@ -588,13 +585,13 @@ class ApiGatewayV2Response(BaseResponse): ) 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] integrations = self.apigatewayv2_backend.get_integrations(api_id=api_id) 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] integration_id = self.path.split("/")[-1] @@ -603,7 +600,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, "{}" - def update_integration(self) -> RESPONSE_TYPE: + def update_integration(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-3] integration_id = self.path.split("/")[-1] @@ -648,7 +645,7 @@ class ApiGatewayV2Response(BaseResponse): ) 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] int_id = self.path.split("/")[-2] @@ -669,7 +666,7 @@ class ApiGatewayV2Response(BaseResponse): ) 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] int_id = self.path.split("/")[-3] int_res_id = self.path.split("/")[-1] @@ -679,7 +676,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, "{}" - def get_integration_response(self) -> RESPONSE_TYPE: + def get_integration_response(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-5] int_id = self.path.split("/")[-3] int_res_id = self.path.split("/")[-1] @@ -689,7 +686,7 @@ class ApiGatewayV2Response(BaseResponse): ) 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] 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]}) - def update_integration_response(self) -> RESPONSE_TYPE: + def update_integration_response(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-5] int_id = self.path.split("/")[-3] int_res_id = self.path.split("/")[-1] @@ -721,7 +718,7 @@ class ApiGatewayV2Response(BaseResponse): ) 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) name = params.get("name") @@ -733,21 +730,21 @@ class ApiGatewayV2Response(BaseResponse): ) 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] self.apigatewayv2_backend.delete_vpc_link(vpc_link_id) 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 = self.apigatewayv2_backend.get_vpc_link(vpc_link_id) 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() 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] params = json.loads(self.body) name = params.get("name") diff --git a/moto/appsync/exceptions.py b/moto/appsync/exceptions.py index 70ebe810e..289d5eb77 100644 --- a/moto/appsync/exceptions.py +++ b/moto/appsync/exceptions.py @@ -9,6 +9,6 @@ class AppSyncExceptions(JsonRESTError): class GraphqlAPINotFound(AppSyncExceptions): code = 404 - def __init__(self, api_id): + def __init__(self, api_id: str): super().__init__("NotFoundException", f"GraphQL API {api_id} not found.") self.description = json.dumps({"message": self.message}) diff --git a/moto/appsync/models.py b/moto/appsync/models.py index ecce39a6b..f2da8dd33 100644 --- a/moto/appsync/models.py +++ b/moto/appsync/models.py @@ -1,5 +1,6 @@ import base64 from datetime import timedelta, datetime, timezone +from typing import Any, Dict, Iterable, List, Optional, Tuple from moto.core import BaseBackend, BaseModel from moto.core.utils import BackendDict, unix_time from moto.moto_api._internal import mock_random @@ -9,16 +10,16 @@ from .exceptions import GraphqlAPINotFound class GraphqlSchema(BaseModel): - def __init__(self, definition): + def __init__(self, definition: Any): self.definition = definition # [graphql.language.ast.ObjectTypeDefinitionNode, ..] - self.types = [] + self.types: List[Any] = [] self.status = "PROCESSING" - self.parse_error = None + self.parse_error: Optional[str] = None 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: if graphql_type.name.value == name: return { @@ -30,10 +31,10 @@ class GraphqlSchema(BaseModel): "definition": "NotYetImplemented", } - def get_status(self): + def get_status(self) -> Tuple[str, Optional[str]]: return self.status, self.parse_error - def _parse_graphql_definition(self): + def _parse_graphql_definition(self) -> None: try: from graphql import parse from graphql.language.ast import ObjectTypeDefinitionNode @@ -49,19 +50,47 @@ class GraphqlSchema(BaseModel): 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): def __init__( self, - account_id, - region, - name, - authentication_type, - additional_authentication_providers, - log_config, - xray_enabled, - user_pool_config, - open_id_connect_config, - lambda_authorizer_config, + account_id: str, + region: str, + name: str, + authentication_type: str, + additional_authentication_providers: Optional[List[str]], + log_config: str, + xray_enabled: str, + user_pool_config: str, + open_id_connect_config: str, + lambda_authorizer_config: str, ): self.region = region self.name = name @@ -75,21 +104,21 @@ class GraphqlAPI(BaseModel): self.xray_enabled = xray_enabled 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( self, - name, - additional_authentication_providers, - authentication_type, - lambda_authorizer_config, - log_config, - open_id_connect_config, - user_pool_config, - xray_enabled, - ): + name: str, + additional_authentication_providers: Optional[List[str]], + authentication_type: str, + lambda_authorizer_config: str, + log_config: str, + open_id_connect_config: str, + user_pool_config: str, + xray_enabled: str, + ) -> None: if name: self.name = name if additional_authentication_providers: @@ -109,36 +138,40 @@ class GraphqlAPI(BaseModel): if xray_enabled is not None: 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) self.api_keys[api_key.key_id] = api_key return api_key - def list_api_keys(self): + def list_api_keys(self) -> Iterable[GraphqlAPIKey]: 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) - 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.update(description, expires) 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") self.graphql_schema = GraphqlSchema(graphql_definition) - def get_schema_status(self): - return self.graphql_schema.get_status() + def get_schema_status(self) -> Any: + return self.graphql_schema.get_status() # type: ignore[union-attr] - def get_type(self, type_name, type_format): - graphql_type = self.graphql_schema.get_type(type_name) - graphql_type["format"] = type_format + def get_type(self, type_name: str, type_format: str) -> Any: + graphql_type = self.graphql_schema.get_type(type_name) # type: ignore[union-attr] + graphql_type["format"] = type_format # type: ignore[index] return graphql_type - def to_json(self): + def to_json(self) -> Dict[str, Any]: return { "name": self.name, "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): """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) - self.graphql_apis = dict() + self.graphql_apis: Dict[str, GraphqlAPI] = dict() self.tagger = TaggingService() def create_graphql_api( self, - name, - log_config, - authentication_type, - user_pool_config, - open_id_connect_config, - additional_authentication_providers, - xray_enabled, - lambda_authorizer_config, - tags, - ): + name: str, + log_config: str, + authentication_type: str, + user_pool_config: str, + open_id_connect_config: str, + additional_authentication_providers: Optional[List[str]], + xray_enabled: str, + lambda_authorizer_config: str, + tags: Dict[str, str], + ) -> GraphqlAPI: graphql_api = GraphqlAPI( account_id=self.account_id, region=self.region_name, @@ -222,16 +227,16 @@ class AppSyncBackend(BaseBackend): def update_graphql_api( self, - api_id, - name, - log_config, - authentication_type, - user_pool_config, - open_id_connect_config, - additional_authentication_providers, - xray_enabled, - lambda_authorizer_config, - ): + api_id: str, + name: str, + log_config: str, + authentication_type: str, + user_pool_config: str, + open_id_connect_config: str, + additional_authentication_providers: Optional[List[str]], + xray_enabled: str, + lambda_authorizer_config: str, + ) -> GraphqlAPI: graphql_api = self.graphql_apis[api_id] graphql_api.update( name, @@ -245,27 +250,29 @@ class AppSyncBackend(BaseBackend): ) 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: raise GraphqlAPINotFound(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) - def list_graphql_apis(self): + def list_graphql_apis(self) -> Iterable[GraphqlAPI]: """ Pagination or the maxResults-parameter have not yet been implemented. """ 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) - 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) - 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. """ @@ -274,30 +281,36 @@ class AppSyncBackend(BaseBackend): else: 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( 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) 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() - def tag_resource(self, resource_arn, tags): + def tag_resource(self, resource_arn: str, tags: Dict[str, str]) -> None: self.tagger.tag_resource( 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) - 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) - 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) diff --git a/moto/appsync/responses.py b/moto/appsync/responses.py index eef8da514..cc712da97 100644 --- a/moto/appsync/responses.py +++ b/moto/appsync/responses.py @@ -1,30 +1,31 @@ """Handles incoming appsync requests, invokes methods, returns responses.""" 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 .models import appsync_backends +from .models import appsync_backends, AppSyncBackend class AppSyncResponse(BaseResponse): """Handler for AppSync requests and responses.""" - def __init__(self): + def __init__(self) -> None: super().__init__(service_name="appsync") @property - def appsync_backend(self): + def appsync_backend(self) -> AppSyncBackend: """Return backend instance specific for this 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) if request.method == "POST": return self.create_graphql_api() if request.method == "GET": 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) if request.method == "GET": return self.get_graphql_api() @@ -33,29 +34,31 @@ class AppSyncResponse(BaseResponse): if request.method == "POST": 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) if request.method == "POST": return self.create_api_key() if request.method == "GET": 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) if request.method == "POST": return self.start_schema_creation() if request.method == "GET": 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) if request.method == "DELETE": return self.delete_api_key() if request.method == "POST": 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) + print("tags") + print(request.method) if request.method == "POST": return self.tag_resource() if request.method == "DELETE": @@ -63,12 +66,12 @@ class AppSyncResponse(BaseResponse): if request.method == "GET": 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) if request.method == "GET": return self.get_type() - def create_graphql_api(self): + def create_graphql_api(self) -> TYPE_RESPONSE: params = json.loads(self.body) name = params.get("name") log_config = params.get("logConfig") @@ -96,7 +99,7 @@ class AppSyncResponse(BaseResponse): response["tags"] = self.appsync_backend.list_tags_for_resource(graphql_api.arn) 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] 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) 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] self.appsync_backend.delete_graphql_api(api_id=api_id) return 200, {}, json.dumps(dict()) - def update_graphql_api(self): + def update_graphql_api(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-1] params = json.loads(self.body) @@ -137,7 +140,7 @@ class AppSyncResponse(BaseResponse): ) 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() return ( 200, @@ -145,7 +148,7 @@ class AppSyncResponse(BaseResponse): 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) # /v1/apis/[api_id]/apikeys api_id = self.path.split("/")[-2] @@ -156,19 +159,19 @@ class AppSyncResponse(BaseResponse): ) 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_key_id = self.path.split("/")[-1] self.appsync_backend.delete_api_key(api_id=api_id, api_key_id=api_key_id) return 200, {}, json.dumps(dict()) - def list_api_keys(self): + def list_api_keys(self) -> TYPE_RESPONSE: # /v1/apis/[api_id]/apikeys api_id = self.path.split("/")[-2] 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])) - def update_api_key(self): + def update_api_key(self) -> TYPE_RESPONSE: api_id = self.path.split("/")[-3] api_key_id = self.path.split("/")[-1] params = json.loads(self.body) @@ -182,7 +185,7 @@ class AppSyncResponse(BaseResponse): ) 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) api_id = self.path.split("/")[-2] definition = params.get("definition") @@ -191,19 +194,19 @@ class AppSyncResponse(BaseResponse): ) 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] status, details = self.appsync_backend.get_schema_creation_status(api_id=api_id) 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() params = json.loads(self.body) tags = params.get("tags") self.appsync_backend.tag_resource(resource_arn=resource_arn, tags=tags) return 200, {}, json.dumps(dict()) - def untag_resource(self): + def untag_resource(self) -> TYPE_RESPONSE: resource_arn = self._extract_arn_from_path() tag_keys = self.querystring.get("tagKeys", []) self.appsync_backend.untag_resource( @@ -211,20 +214,20 @@ class AppSyncResponse(BaseResponse): ) 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() tags = self.appsync_backend.list_tags_for_resource(resource_arn=resource_arn) 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 path = unquote(self.path) return "/".join(path.split("/")[3:]) - def get_type(self): + def get_type(self) -> TYPE_RESPONSE: api_id = unquote(self.path.split("/")[-3]) 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( api_id=api_id, type_name=type_name, type_format=type_format ) diff --git a/moto/core/responses.py b/moto/core/responses.py index 52d148098..cce796499 100644 --- a/moto/core/responses.py +++ b/moto/core/responses.py @@ -15,7 +15,7 @@ from moto.core.exceptions import DryRunClientError from moto.core.utils import camelcase_to_underscores, method_names_from_class from moto.utilities.utils import load_resource 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 werkzeug.exceptions import HTTPException from xml.dom.minidom import parseString as parseXML @@ -25,6 +25,8 @@ log = logging.getLogger(__name__) JINJA_ENVS = {} +TYPE_RESPONSE = Tuple[int, Dict[str, str], str] + def _decode_dict(d): decoded = OrderedDict() diff --git a/setup.cfg b/setup.cfg index 8233837a8..9911d789d 100644 --- a/setup.cfg +++ b/setup.cfg @@ -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 [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_error_codes = True disable_error_code=abstract