diff --git a/moto/apigatewayv2/exceptions.py b/moto/apigatewayv2/exceptions.py index 795805026..73f8a213a 100644 --- a/moto/apigatewayv2/exceptions.py +++ b/moto/apigatewayv2/exceptions.py @@ -8,7 +8,7 @@ class APIGatewayV2Error(JsonRESTError): class ApiNotFound(APIGatewayV2Error): code = 404 - def __init__(self, api_id): + def __init__(self, api_id: str): super().__init__( "NotFoundException", f"Invalid API identifier specified {api_id}" ) @@ -17,7 +17,7 @@ class ApiNotFound(APIGatewayV2Error): class AuthorizerNotFound(APIGatewayV2Error): code = 404 - def __init__(self, authorizer_id): + def __init__(self, authorizer_id: str): super().__init__( "NotFoundException", f"Invalid Authorizer identifier specified {authorizer_id}", @@ -27,7 +27,7 @@ class AuthorizerNotFound(APIGatewayV2Error): class ModelNotFound(APIGatewayV2Error): code = 404 - def __init__(self, model_id): + def __init__(self, model_id: str): super().__init__( "NotFoundException", f"Invalid Model identifier specified {model_id}" ) @@ -36,7 +36,7 @@ class ModelNotFound(APIGatewayV2Error): class RouteResponseNotFound(APIGatewayV2Error): code = 404 - def __init__(self, rr_id): + def __init__(self, rr_id: str): super().__init__( "NotFoundException", f"Invalid RouteResponse identifier specified {rr_id}" ) @@ -45,14 +45,14 @@ class RouteResponseNotFound(APIGatewayV2Error): class BadRequestException(APIGatewayV2Error): code = 400 - def __init__(self, message): + def __init__(self, message: str): super().__init__("BadRequestException", message) class IntegrationNotFound(APIGatewayV2Error): code = 404 - def __init__(self, integration_id): + def __init__(self, integration_id: str): super().__init__( "NotFoundException", f"Invalid Integration identifier specified {integration_id}", @@ -62,7 +62,7 @@ class IntegrationNotFound(APIGatewayV2Error): class IntegrationResponseNotFound(APIGatewayV2Error): code = 404 - def __init__(self, int_res_id): + def __init__(self, int_res_id: str): super().__init__( "NotFoundException", f"Invalid IntegrationResponse identifier specified {int_res_id}", @@ -72,7 +72,7 @@ class IntegrationResponseNotFound(APIGatewayV2Error): class RouteNotFound(APIGatewayV2Error): code = 404 - def __init__(self, route_id): + def __init__(self, route_id: str): super().__init__( "NotFoundException", f"Invalid Route identifier specified {route_id}" ) @@ -81,14 +81,14 @@ class RouteNotFound(APIGatewayV2Error): class VpcLinkNotFound(APIGatewayV2Error): code = 404 - def __init__(self, vpc_link_id): + def __init__(self, vpc_link_id: str): super().__init__( "NotFoundException", f"Invalid VpcLink identifier specified {vpc_link_id}" ) class UnknownProtocol(APIGatewayV2Error): - def __init__(self): + def __init__(self) -> None: super().__init__( "BadRequestException", "Invalid protocol specified. Must be one of [HTTP, WEBSOCKET]", diff --git a/moto/apigatewayv2/models.py b/moto/apigatewayv2/models.py index 52b338cad..4634e9cbd 100644 --- a/moto/apigatewayv2/models.py +++ b/moto/apigatewayv2/models.py @@ -1,6 +1,7 @@ """ApiGatewayV2Backend class with methods for supported APIs.""" import string import yaml +from typing import Any, Dict, List, Optional from moto.core import BaseBackend, BaseModel from moto.core.utils import BackendDict, unix_time @@ -23,16 +24,16 @@ from .exceptions import ( class Authorizer(BaseModel): def __init__( self, - auth_creds_arn, - auth_payload_format_version, - auth_result_ttl, - authorizer_type, - authorizer_uri, - enable_simple_response, - identity_source, - identity_validation_expr, - jwt_config, - name, + auth_creds_arn: str, + auth_payload_format_version: str, + auth_result_ttl: str, + authorizer_type: str, + authorizer_uri: str, + enable_simple_response: str, + identity_source: str, + identity_validation_expr: str, + jwt_config: str, + name: str, ): self.id = "".join(random.choice(string.ascii_lowercase) for _ in range(8)) self.auth_creds_arn = auth_creds_arn @@ -48,17 +49,17 @@ class Authorizer(BaseModel): def update( self, - auth_creds_arn, - auth_payload_format_version, - auth_result_ttl, - authorizer_type, - authorizer_uri, - enable_simple_response, - identity_source, - identity_validation_expr, - jwt_config, - name, - ): + auth_creds_arn: str, + auth_payload_format_version: str, + auth_result_ttl: str, + authorizer_type: str, + authorizer_uri: str, + enable_simple_response: str, + identity_source: str, + identity_validation_expr: str, + jwt_config: str, + name: str, + ) -> None: if auth_creds_arn is not None: self.auth_creds_arn = auth_creds_arn if auth_payload_format_version is not None: @@ -80,7 +81,7 @@ class Authorizer(BaseModel): if name is not None: self.name = name - def to_json(self): + def to_json(self) -> Dict[str, Any]: return { "authorizerId": self.id, "authorizerCredentialsArn": self.auth_creds_arn, @@ -99,23 +100,23 @@ class Authorizer(BaseModel): class Integration(BaseModel): def __init__( self, - connection_id, - connection_type, - content_handling_strategy, - credentials_arn, - description, - integration_method, - integration_type, - integration_uri, - passthrough_behavior, - payload_format_version, - integration_subtype, - request_parameters, - request_templates, - response_parameters, - template_selection_expression, - timeout_in_millis, - tls_config, + connection_id: Optional[str], + connection_type: str, + content_handling_strategy: Optional[str], + credentials_arn: Optional[str], + description: str, + integration_method: str, + integration_type: str, + integration_uri: str, + passthrough_behavior: Optional[str], + payload_format_version: Optional[str], + integration_subtype: Optional[str], + request_parameters: Optional[Dict[str, str]], + request_templates: Optional[Dict[str, str]], + response_parameters: Optional[Dict[str, Dict[str, str]]], + template_selection_expression: Optional[str], + timeout_in_millis: Optional[str], + tls_config: Optional[Dict[str, str]], ): self.id = "".join(random.choice(string.ascii_lowercase) for _ in range(8)) self.connection_id = connection_id @@ -134,7 +135,7 @@ class Integration(BaseModel): self.request_templates = request_templates self.response_parameters = response_parameters self.template_selection_expression = template_selection_expression - self.timeout_in_millis = timeout_in_millis + self.timeout_in_millis = int(timeout_in_millis) if timeout_in_millis else None self.tls_config = tls_config if self.integration_type in ["MOCK", "HTTP"]: @@ -157,16 +158,16 @@ class Integration(BaseModel): else: self.timeout_in_millis = self.timeout_in_millis or 30000 - self.responses = dict() + self.responses: Dict[str, IntegrationResponse] = dict() def create_response( self, - content_handling_strategy, - integration_response_key, - response_parameters, - response_templates, - template_selection_expression, - ): + content_handling_strategy: str, + integration_response_key: str, + response_parameters: str, + response_templates: str, + template_selection_expression: str, + ) -> "IntegrationResponse": response = IntegrationResponse( content_handling_strategy=content_handling_strategy, integration_response_key=integration_response_key, @@ -177,26 +178,26 @@ class Integration(BaseModel): self.responses[response.id] = response return response - def delete_response(self, integration_response_id): + def delete_response(self, integration_response_id: str) -> None: self.responses.pop(integration_response_id) - def get_response(self, integration_response_id): + def get_response(self, integration_response_id: str) -> "IntegrationResponse": if integration_response_id not in self.responses: raise IntegrationResponseNotFound(integration_response_id) return self.responses[integration_response_id] - def get_responses(self): - return self.responses.values() + def get_responses(self) -> List["IntegrationResponse"]: + return list(self.responses.values()) def update_response( self, - integration_response_id, - content_handling_strategy, - integration_response_key, - response_parameters, - response_templates, - template_selection_expression, - ): + integration_response_id: str, + content_handling_strategy: str, + integration_response_key: str, + response_parameters: str, + response_templates: str, + template_selection_expression: str, + ) -> "IntegrationResponse": int_response = self.responses[integration_response_id] int_response.update( content_handling_strategy=content_handling_strategy, @@ -209,24 +210,24 @@ class Integration(BaseModel): def update( self, - connection_id, - connection_type, - content_handling_strategy, - credentials_arn, - description, - integration_method, - integration_type, - integration_uri, - passthrough_behavior, - payload_format_version, - integration_subtype, - request_parameters, - request_templates, - response_parameters, - template_selection_expression, - timeout_in_millis, - tls_config, - ): + connection_id: str, + connection_type: str, + content_handling_strategy: str, + credentials_arn: str, + description: str, + integration_method: str, + integration_type: str, + integration_uri: str, + passthrough_behavior: str, + payload_format_version: str, + integration_subtype: str, + request_parameters: Dict[str, str], + request_templates: Dict[str, str], + response_parameters: Dict[str, Dict[str, str]], + template_selection_expression: str, + timeout_in_millis: Optional[int], + tls_config: Dict[str, str], + ) -> None: if connection_id is not None: self.connection_id = connection_id if connection_type is not None: @@ -266,7 +267,7 @@ class Integration(BaseModel): if tls_config is not None: self.tls_config = tls_config - def to_json(self): + def to_json(self) -> Dict[str, Any]: return { "connectionId": self.connection_id, "connectionType": self.connection_type, @@ -293,11 +294,11 @@ class Integration(BaseModel): class IntegrationResponse(BaseModel): def __init__( self, - content_handling_strategy, - integration_response_key, - response_parameters, - response_templates, - template_selection_expression, + content_handling_strategy: str, + integration_response_key: str, + response_parameters: str, + response_templates: str, + template_selection_expression: str, ): self.id = "".join(random.choice(string.ascii_lowercase) for _ in range(8)) self.content_handling_strategy = content_handling_strategy @@ -308,12 +309,12 @@ class IntegrationResponse(BaseModel): def update( self, - content_handling_strategy, - integration_response_key, - response_parameters, - response_templates, - template_selection_expression, - ): + content_handling_strategy: str, + integration_response_key: str, + response_parameters: str, + response_templates: str, + template_selection_expression: str, + ) -> None: if content_handling_strategy is not None: self.content_handling_strategy = content_handling_strategy if integration_response_key is not None: @@ -325,7 +326,7 @@ class IntegrationResponse(BaseModel): if template_selection_expression is not None: self.template_selection_expression = template_selection_expression - def to_json(self): + def to_json(self) -> Dict[str, str]: return { "integrationResponseId": self.id, "integrationResponseKey": self.integration_response_key, @@ -337,14 +338,16 @@ class IntegrationResponse(BaseModel): class Model(BaseModel): - def __init__(self, content_type, description, name, schema): + def __init__(self, content_type: str, description: str, name: str, schema: str): self.id = "".join(random.choice(string.ascii_lowercase) for _ in range(8)) self.content_type = content_type self.description = description self.name = name self.schema = schema - def update(self, content_type, description, name, schema): + def update( + self, content_type: str, description: str, name: str, schema: str + ) -> None: if content_type is not None: self.content_type = content_type if description is not None: @@ -354,7 +357,7 @@ class Model(BaseModel): if schema is not None: self.schema = schema - def to_json(self): + def to_json(self) -> Dict[str, str]: return { "modelId": self.id, "contentType": self.content_type, @@ -365,13 +368,18 @@ class Model(BaseModel): class RouteResponse(BaseModel): - def __init__(self, route_response_key, model_selection_expression, response_models): + def __init__( + self, + route_response_key: str, + model_selection_expression: str, + response_models: str, + ): self.id = "".join(random.choice(string.ascii_lowercase) for _ in range(8)) self.route_response_key = route_response_key self.model_selection_expression = model_selection_expression self.response_models = response_models - def to_json(self): + def to_json(self) -> Dict[str, str]: return { "modelSelectionExpression": self.model_selection_expression, "responseModels": self.response_models, @@ -383,17 +391,17 @@ class RouteResponse(BaseModel): class Route(BaseModel): def __init__( self, - api_key_required, - authorization_scopes, - authorization_type, - authorizer_id, - model_selection_expression, - operation_name, - request_models, - request_parameters, - route_key, - route_response_selection_expression, - target, + api_key_required: bool, + authorization_scopes: List[str], + authorization_type: Optional[str], + authorizer_id: Optional[str], + model_selection_expression: Optional[str], + operation_name: Optional[str], + request_models: Optional[Dict[str, str]], + request_parameters: Optional[Dict[str, Dict[str, bool]]], + route_key: str, + route_response_selection_expression: Optional[str], + target: str, ): self.route_id = "".join(random.choice(string.ascii_lowercase) for _ in range(8)) self.api_key_required = api_key_required @@ -408,11 +416,14 @@ class Route(BaseModel): self.route_response_selection_expression = route_response_selection_expression self.target = target - self.route_responses = dict() + self.route_responses: Dict[str, RouteResponse] = dict() def create_route_response( - self, route_response_key, model_selection_expression, response_models - ): + self, + route_response_key: str, + model_selection_expression: str, + response_models: str, + ) -> RouteResponse: route_response = RouteResponse( route_response_key, model_selection_expression=model_selection_expression, @@ -421,31 +432,31 @@ class Route(BaseModel): self.route_responses[route_response.id] = route_response return route_response - def get_route_response(self, route_response_id): + def get_route_response(self, route_response_id: str) -> RouteResponse: if route_response_id not in self.route_responses: raise RouteResponseNotFound(route_response_id) return self.route_responses[route_response_id] - def delete_route_response(self, route_response_id): + def delete_route_response(self, route_response_id: str) -> None: self.route_responses.pop(route_response_id, None) - def delete_route_request_parameter(self, request_param): + def delete_route_request_parameter(self, request_param: str) -> None: del self.request_parameters[request_param] def update( self, - api_key_required, - authorization_scopes, - authorization_type, - authorizer_id, - model_selection_expression, - operation_name, - request_models, - request_parameters, - route_key, - route_response_selection_expression, - target, - ): + api_key_required: Optional[bool], + authorization_scopes: Optional[List[str]], + authorization_type: str, + authorizer_id: str, + model_selection_expression: str, + operation_name: str, + request_models: Dict[str, str], + request_parameters: Dict[str, Dict[str, bool]], + route_key: str, + route_response_selection_expression: str, + target: str, + ) -> None: if api_key_required is not None: self.api_key_required = api_key_required if authorization_scopes: @@ -471,7 +482,7 @@ class Route(BaseModel): if target: self.target = target - def to_json(self): + def to_json(self) -> Dict[str, Any]: return { "apiKeyRequired": self.api_key_required, "authorizationScopes": self.authorization_scopes, @@ -491,18 +502,18 @@ class Route(BaseModel): class Api(BaseModel): def __init__( self, - region, - name, - api_key_selection_expression, - cors_configuration, - description, - disable_execute_api_endpoint, - disable_schema_validation, - protocol_type, - route_selection_expression, - tags, - version, - backend, + region: str, + name: str, + api_key_selection_expression: str, + cors_configuration: Optional[str], + description: str, + disable_execute_api_endpoint: str, + disable_schema_validation: str, + protocol_type: str, + route_selection_expression: str, + tags: Dict[str, str], + version: str, + backend: "ApiGatewayV2Backend", ): self.api_id = "".join(random.choice(string.ascii_lowercase) for _ in range(8)) self.api_endpoint = f"https://{self.api_id}.execute-api.{region}.amazonaws.com" @@ -522,36 +533,36 @@ class Api(BaseModel): ) self.version = version - self.authorizers = dict() - self.integrations = dict() - self.models = dict() - self.routes = dict() + self.authorizers: Dict[str, Authorizer] = dict() + self.integrations: Dict[str, Integration] = dict() + self.models: Dict[str, Model] = dict() + self.routes: Dict[str, Route] = dict() self.arn = f"arn:aws:apigateway:{region}::/apis/{self.api_id}" self.backend.tag_resource(self.arn, tags) - def clear(self): - self.authorizers = dict() - self.integrations = dict() - self.models = dict() - self.routes = dict() + def clear(self) -> None: + self.authorizers.clear() + self.integrations.clear() + self.models.clear() + self.routes.clear() - def delete_cors_configuration(self): + def delete_cors_configuration(self) -> None: self.cors_configuration = None def create_authorizer( self, - auth_creds_arn, - auth_payload_format_version, - auth_result_ttl, - authorizer_type, - authorizer_uri, - enable_simple_response, - identity_source, - identity_validation_expr, - jwt_config, - name, - ): + auth_creds_arn: str, + auth_payload_format_version: str, + auth_result_ttl: str, + authorizer_type: str, + authorizer_uri: str, + enable_simple_response: str, + identity_source: str, + identity_validation_expr: str, + jwt_config: str, + name: str, + ) -> Authorizer: authorizer = Authorizer( auth_creds_arn=auth_creds_arn, auth_payload_format_version=auth_payload_format_version, @@ -567,28 +578,28 @@ class Api(BaseModel): self.authorizers[authorizer.id] = authorizer return authorizer - def delete_authorizer(self, authorizer_id): + def delete_authorizer(self, authorizer_id: str) -> None: self.authorizers.pop(authorizer_id, None) - def get_authorizer(self, authorizer_id): + def get_authorizer(self, authorizer_id: str) -> Authorizer: if authorizer_id not in self.authorizers: raise AuthorizerNotFound(authorizer_id) return self.authorizers[authorizer_id] def update_authorizer( self, - authorizer_id, - auth_creds_arn, - auth_payload_format_version, - auth_result_ttl, - authorizer_type, - authorizer_uri, - enable_simple_response, - identity_source, - identity_validation_expr, - jwt_config, - name, - ): + authorizer_id: str, + auth_creds_arn: str, + auth_payload_format_version: str, + auth_result_ttl: str, + authorizer_type: str, + authorizer_uri: str, + enable_simple_response: str, + identity_source: str, + identity_validation_expr: str, + jwt_config: str, + name: str, + ) -> Authorizer: authorizer = self.authorizers[authorizer_id] authorizer.update( auth_creds_arn=auth_creds_arn, @@ -604,27 +615,31 @@ class Api(BaseModel): ) return authorizer - def create_model(self, content_type, description, name, schema): + def create_model( + self, content_type: str, description: str, name: str, schema: str + ) -> Model: model = Model(content_type, description, name, schema) self.models[model.id] = model return model - def delete_model(self, model_id): + def delete_model(self, model_id: str) -> None: self.models.pop(model_id, None) - def get_model(self, model_id): + def get_model(self, model_id: str) -> Model: if model_id not in self.models: raise ModelNotFound(model_id) return self.models[model_id] - def update_model(self, model_id, content_type, description, name, schema): + def update_model( + self, model_id: str, content_type: str, description: str, name: str, schema: str + ) -> Model: model = self.models[model_id] model.update(content_type, description, name, schema) return model - def import_api(self, body, fail_on_warnings): + def import_api(self, body_str: str, fail_on_warnings: bool) -> None: self.clear() - body = yaml.safe_load(body) + body = yaml.safe_load(body_str) for path, path_details in body.get("paths", {}).items(): for method, method_details in path_details.items(): route_key = f"{method.upper()} {path}" @@ -665,15 +680,15 @@ class Api(BaseModel): def update( self, - api_key_selection_expression, - cors_configuration, - description, - disable_schema_validation, - disable_execute_api_endpoint, - name, - route_selection_expression, - version, - ): + api_key_selection_expression: str, + cors_configuration: str, + description: str, + disable_schema_validation: str, + disable_execute_api_endpoint: str, + name: str, + route_selection_expression: str, + version: str, + ) -> None: if api_key_selection_expression is not None: self.api_key_selection_expression = api_key_selection_expression if cors_configuration is not None: @@ -693,24 +708,24 @@ class Api(BaseModel): def create_integration( self, - connection_type, - description, - integration_method, - integration_type, - integration_uri, - connection_id=None, - content_handling_strategy=None, - credentials_arn=None, - passthrough_behavior=None, - payload_format_version=None, - integration_subtype=None, - request_parameters=None, - request_templates=None, - response_parameters=None, - template_selection_expression=None, - timeout_in_millis=None, - tls_config=None, - ): + connection_type: str, + description: str, + integration_method: str, + integration_type: str, + integration_uri: str, + connection_id: Optional[str] = None, + content_handling_strategy: Optional[str] = None, + credentials_arn: Optional[str] = None, + passthrough_behavior: Optional[str] = None, + payload_format_version: Optional[str] = None, + integration_subtype: Optional[str] = None, + request_parameters: Optional[Dict[str, str]] = None, + request_templates: Optional[Dict[str, str]] = None, + response_parameters: Optional[Dict[str, Dict[str, str]]] = None, + template_selection_expression: Optional[str] = None, + timeout_in_millis: Optional[str] = None, + tls_config: Optional[Dict[str, str]] = None, + ) -> Integration: integration = Integration( connection_id=connection_id, connection_type=connection_type, @@ -733,38 +748,38 @@ class Api(BaseModel): self.integrations[integration.id] = integration return integration - def delete_integration(self, integration_id): + def delete_integration(self, integration_id: str) -> None: self.integrations.pop(integration_id, None) - def get_integration(self, integration_id): + def get_integration(self, integration_id: str) -> Integration: if integration_id not in self.integrations: raise IntegrationNotFound(integration_id) return self.integrations[integration_id] - def get_integrations(self): - return self.integrations.values() + def get_integrations(self) -> List[Integration]: + return list(self.integrations.values()) def update_integration( self, - integration_id, - connection_id, - connection_type, - content_handling_strategy, - credentials_arn, - description, - integration_method, - integration_type, - integration_uri, - passthrough_behavior, - payload_format_version, - integration_subtype, - request_parameters, - request_templates, - response_parameters, - template_selection_expression, - timeout_in_millis, - tls_config, - ): + integration_id: str, + connection_id: str, + connection_type: str, + content_handling_strategy: str, + credentials_arn: str, + description: str, + integration_method: str, + integration_type: str, + integration_uri: str, + passthrough_behavior: str, + payload_format_version: str, + integration_subtype: str, + request_parameters: Dict[str, str], + request_templates: Dict[str, str], + response_parameters: Dict[str, Dict[str, str]], + template_selection_expression: str, + timeout_in_millis: Optional[int], + tls_config: Dict[str, str], + ) -> Integration: integration = self.integrations[integration_id] integration.update( connection_id=connection_id, @@ -789,13 +804,13 @@ class Api(BaseModel): def create_integration_response( self, - integration_id, - content_handling_strategy, - integration_response_key, - response_parameters, - response_templates, - template_selection_expression, - ): + integration_id: str, + content_handling_strategy: str, + integration_response_key: str, + response_parameters: str, + response_templates: str, + template_selection_expression: str, + ) -> IntegrationResponse: integration = self.get_integration(integration_id) return integration.create_response( content_handling_strategy=content_handling_strategy, @@ -805,28 +820,34 @@ class Api(BaseModel): template_selection_expression=template_selection_expression, ) - def delete_integration_response(self, integration_id, integration_response_id): + def delete_integration_response( + self, integration_id: str, integration_response_id: str + ) -> None: integration = self.get_integration(integration_id) integration.delete_response(integration_response_id) - def get_integration_response(self, integration_id, integration_response_id): + def get_integration_response( + self, integration_id: str, integration_response_id: str + ) -> IntegrationResponse: integration = self.get_integration(integration_id) return integration.get_response(integration_response_id) - def get_integration_responses(self, integration_id): + def get_integration_responses( + self, integration_id: str + ) -> List[IntegrationResponse]: integration = self.get_integration(integration_id) return integration.get_responses() def update_integration_response( self, - integration_id, - integration_response_id, - content_handling_strategy, - integration_response_key, - response_parameters, - response_templates, - template_selection_expression, - ): + integration_id: str, + integration_response_id: str, + content_handling_strategy: str, + integration_response_key: str, + response_parameters: str, + response_templates: str, + template_selection_expression: str, + ) -> IntegrationResponse: integration = self.get_integration(integration_id) return integration.update_response( integration_response_id=integration_response_id, @@ -839,18 +860,18 @@ class Api(BaseModel): def create_route( self, - api_key_required, - authorization_scopes, - route_key, - target, - authorization_type=None, - authorizer_id=None, - model_selection_expression=None, - operation_name=None, - request_models=None, - request_parameters=None, - route_response_selection_expression=None, - ): + api_key_required: bool, + authorization_scopes: List[str], + route_key: str, + target: str, + authorization_type: Optional[str] = None, + authorizer_id: Optional[str] = None, + model_selection_expression: Optional[str] = None, + operation_name: Optional[str] = None, + request_models: Optional[Dict[str, str]] = None, + request_parameters: Optional[Dict[str, Dict[str, bool]]] = None, + route_response_selection_expression: Optional[str] = None, + ) -> Route: route = Route( api_key_required=api_key_required, authorization_scopes=authorization_scopes, @@ -867,36 +888,36 @@ class Api(BaseModel): self.routes[route.route_id] = route return route - def delete_route(self, route_id): + def delete_route(self, route_id: str) -> None: self.routes.pop(route_id, None) - def delete_route_request_parameter(self, route_id, request_param): + def delete_route_request_parameter(self, route_id: str, request_param: str) -> None: route = self.get_route(route_id) route.delete_route_request_parameter(request_param) - def get_route(self, route_id): + def get_route(self, route_id: str) -> Route: if route_id not in self.routes: raise RouteNotFound(route_id) return self.routes[route_id] - def get_routes(self): - return self.routes.values() + def get_routes(self) -> List[Route]: + return list(self.routes.values()) def update_route( self, - route_id, - api_key_required, - authorization_scopes, - authorization_type, - authorizer_id, - model_selection_expression, - operation_name, - request_models, - request_parameters, - route_key, - route_response_selection_expression, - target, - ): + route_id: str, + api_key_required: Optional[bool], + authorization_scopes: List[str], + authorization_type: str, + authorizer_id: str, + model_selection_expression: str, + operation_name: str, + request_models: Dict[str, str], + request_parameters: Dict[str, Dict[str, bool]], + route_key: str, + route_response_selection_expression: str, + target: str, + ) -> Route: route = self.get_route(route_id) route.update( api_key_required=api_key_required, @@ -914,8 +935,12 @@ class Api(BaseModel): return route def create_route_response( - self, route_id, route_response_key, model_selection_expression, response_models - ): + self, + route_id: str, + route_response_key: str, + model_selection_expression: str, + response_models: str, + ) -> RouteResponse: route = self.get_route(route_id) return route.create_route_response( route_response_key, @@ -923,15 +948,17 @@ class Api(BaseModel): response_models=response_models, ) - def delete_route_response(self, route_id, route_response_id): + def delete_route_response(self, route_id: str, route_response_id: str) -> None: route = self.get_route(route_id) route.delete_route_response(route_response_id) - def get_route_response(self, route_id, route_response_id): + def get_route_response( + self, route_id: str, route_response_id: str + ) -> RouteResponse: route = self.get_route(route_id) return route.get_route_response(route_response_id) - def to_json(self): + def to_json(self) -> Dict[str, Any]: return { "apiId": self.api_id, "apiEndpoint": self.api_endpoint, @@ -950,7 +977,14 @@ class Api(BaseModel): class VpcLink(BaseModel): - def __init__(self, name, sg_ids, subnet_ids, tags, backend): + def __init__( + self, + name: str, + sg_ids: List[str], + subnet_ids: List[str], + tags: Dict[str, str], + backend: "ApiGatewayV2Backend", + ): self.created = unix_time() self.id = "".join(random.choice(string.ascii_lowercase) for _ in range(8)) self.name = name @@ -961,10 +995,10 @@ class VpcLink(BaseModel): self.backend = backend self.backend.tag_resource(self.arn, tags) - def update(self, name): + def update(self, name: str) -> None: self.name = name - def to_json(self): + def to_json(self) -> Dict[str, Any]: return { "createdDate": self.created, "name": self.name, @@ -980,25 +1014,25 @@ class VpcLink(BaseModel): class ApiGatewayV2Backend(BaseBackend): """Implementation of ApiGatewayV2 APIs.""" - def __init__(self, region_name, account_id): + def __init__(self, region_name: str, account_id: str): super().__init__(region_name, account_id) - self.apis = dict() - self.vpc_links = dict() + self.apis: Dict[str, Api] = dict() + self.vpc_links: Dict[str, VpcLink] = dict() self.tagger = TaggingService() def create_api( self, - api_key_selection_expression, - cors_configuration, - description, - disable_schema_validation, - disable_execute_api_endpoint, - name, - protocol_type, - route_selection_expression, - tags, - version, - ): + api_key_selection_expression: str, + cors_configuration: str, + description: str, + disable_schema_validation: str, + disable_execute_api_endpoint: str, + name: str, + protocol_type: str, + route_selection_expression: str, + tags: Dict[str, str], + version: str, + ) -> Api: """ The following parameters are not yet implemented: CredentialsArn, RouteKey, Tags, Target @@ -1020,32 +1054,32 @@ class ApiGatewayV2Backend(BaseBackend): self.apis[api.api_id] = api return api - def delete_api(self, api_id): + def delete_api(self, api_id: str) -> None: self.apis.pop(api_id, None) - def get_api(self, api_id) -> Api: + def get_api(self, api_id: str) -> Api: if api_id not in self.apis: raise ApiNotFound(api_id) return self.apis[api_id] - def get_apis(self): + def get_apis(self) -> List[Api]: """ Pagination is not yet implemented """ - return self.apis.values() + return list(self.apis.values()) def update_api( self, - api_id, - api_key_selection_expression, - cors_configuration, - description, - disable_schema_validation, - disable_execute_api_endpoint, - name, - route_selection_expression, - version, - ): + api_id: str, + api_key_selection_expression: str, + cors_configuration: str, + description: str, + disable_schema_validation: str, + disable_execute_api_endpoint: str, + name: str, + route_selection_expression: str, + version: str, + ) -> Api: """ The following parameters have not yet been implemented: CredentialsArn, RouteKey, Target """ @@ -1062,7 +1096,7 @@ class ApiGatewayV2Backend(BaseBackend): ) return api - def reimport_api(self, api_id, body, fail_on_warnings): + def reimport_api(self, api_id: str, body: str, fail_on_warnings: bool) -> Api: """ Only YAML is supported at the moment. Full OpenAPI-support is not guaranteed. Only limited validation is implemented """ @@ -1070,24 +1104,24 @@ class ApiGatewayV2Backend(BaseBackend): api.import_api(body, fail_on_warnings) return api - def delete_cors_configuration(self, api_id): + def delete_cors_configuration(self, api_id: str) -> None: api = self.get_api(api_id) api.delete_cors_configuration() def create_authorizer( self, - api_id, - auth_creds_arn, - auth_payload_format_version, - auth_result_ttl, - authorizer_uri, - authorizer_type, - enable_simple_response, - identity_source, - identity_validation_expr, - jwt_config, - name, - ): + api_id: str, + auth_creds_arn: str, + auth_payload_format_version: str, + auth_result_ttl: str, + authorizer_uri: str, + authorizer_type: str, + enable_simple_response: str, + identity_source: str, + identity_validation_expr: str, + jwt_config: str, + name: str, + ) -> Authorizer: api = self.get_api(api_id) if ( @@ -1113,30 +1147,30 @@ class ApiGatewayV2Backend(BaseBackend): ) return authorizer - def delete_authorizer(self, api_id, authorizer_id): + def delete_authorizer(self, api_id: str, authorizer_id: str) -> None: api = self.get_api(api_id) api.delete_authorizer(authorizer_id=authorizer_id) - def get_authorizer(self, api_id, authorizer_id): + def get_authorizer(self, api_id: str, authorizer_id: str) -> Authorizer: api = self.get_api(api_id) authorizer = api.get_authorizer(authorizer_id=authorizer_id) return authorizer def update_authorizer( self, - api_id, - authorizer_id, - auth_creds_arn, - auth_payload_format_version, - auth_result_ttl, - authorizer_uri, - authorizer_type, - enable_simple_response, - identity_source, - identity_validation_expr, - jwt_config, - name, - ): + api_id: str, + authorizer_id: str, + auth_creds_arn: str, + auth_payload_format_version: str, + auth_result_ttl: str, + authorizer_uri: str, + authorizer_type: str, + enable_simple_response: str, + identity_source: str, + identity_validation_expr: str, + jwt_config: str, + name: str, + ) -> Authorizer: api = self.get_api(api_id) authorizer = api.update_authorizer( authorizer_id=authorizer_id, @@ -1153,50 +1187,60 @@ class ApiGatewayV2Backend(BaseBackend): ) return authorizer - def create_model(self, api_id, content_type, description, name, schema): + def create_model( + self, api_id: str, content_type: str, description: str, name: str, schema: str + ) -> Model: api = self.get_api(api_id) model = api.create_model( content_type=content_type, description=description, name=name, schema=schema ) return model - def delete_model(self, api_id, model_id): + def delete_model(self, api_id: str, model_id: str) -> None: api = self.get_api(api_id) api.delete_model(model_id=model_id) - def get_model(self, api_id, model_id): + def get_model(self, api_id: str, model_id: str) -> Model: api = self.get_api(api_id) return api.get_model(model_id) - def update_model(self, api_id, model_id, content_type, description, name, schema): + def update_model( + self, + api_id: str, + model_id: str, + content_type: str, + description: str, + name: str, + schema: str, + ) -> Model: api = self.get_api(api_id) return api.update_model(model_id, content_type, description, name, schema) - def get_tags(self, resource_id): + def get_tags(self, resource_id: str) -> Dict[str, str]: return self.tagger.get_tag_dict_for_resource(resource_id) - def tag_resource(self, resource_arn, tags): - tags = TaggingService.convert_dict_to_tags_input(tags or {}) - self.tagger.tag_resource(resource_arn, tags) + def tag_resource(self, resource_arn: str, tags: Dict[str, str]) -> None: + tags_input = TaggingService.convert_dict_to_tags_input(tags or {}) + self.tagger.tag_resource(resource_arn, tags_input) - 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 create_route( self, - api_id, - api_key_required, - authorization_scopes, - authorization_type, - authorizer_id, - model_selection_expression, - operation_name, - request_models, - request_parameters, - route_key, - route_response_selection_expression, - target, - ): + api_id: str, + api_key_required: bool, + authorization_scopes: List[str], + authorization_type: str, + authorizer_id: str, + model_selection_expression: str, + operation_name: str, + request_models: Optional[Dict[str, str]], + request_parameters: Optional[Dict[str, Dict[str, bool]]], + route_key: str, + route_response_selection_expression: str, + target: str, + ) -> Route: api = self.get_api(api_id) route = api.create_route( api_key_required=api_key_required, @@ -1213,19 +1257,21 @@ class ApiGatewayV2Backend(BaseBackend): ) return route - def delete_route(self, api_id, route_id): + def delete_route(self, api_id: str, route_id: str) -> None: api = self.get_api(api_id) api.delete_route(route_id) - def delete_route_request_parameter(self, api_id, route_id, request_param): + def delete_route_request_parameter( + self, api_id: str, route_id: str, request_param: str + ) -> None: api = self.get_api(api_id) api.delete_route_request_parameter(route_id, request_param) - def get_route(self, api_id, route_id): + def get_route(self, api_id: str, route_id: str) -> Route: api = self.get_api(api_id) return api.get_route(route_id) - def get_routes(self, api_id): + def get_routes(self, api_id: str) -> List[Route]: """ Pagination is not yet implemented """ @@ -1234,20 +1280,20 @@ class ApiGatewayV2Backend(BaseBackend): def update_route( self, - api_id, - api_key_required, - authorization_scopes, - authorization_type, - authorizer_id, - model_selection_expression, - operation_name, - request_models, - request_parameters, - route_id, - route_key, - route_response_selection_expression, - target, - ): + api_id: str, + api_key_required: bool, + authorization_scopes: List[str], + authorization_type: str, + authorizer_id: str, + model_selection_expression: str, + operation_name: str, + request_models: Dict[str, str], + request_parameters: Dict[str, Dict[str, bool]], + route_id: str, + route_key: str, + route_response_selection_expression: str, + target: str, + ) -> Route: api = self.get_api(api_id) route = api.update_route( route_id=route_id, @@ -1267,12 +1313,12 @@ class ApiGatewayV2Backend(BaseBackend): def create_route_response( self, - api_id, - route_id, - route_response_key, - model_selection_expression, - response_models, - ): + api_id: str, + route_id: str, + route_response_key: str, + model_selection_expression: str, + response_models: str, + ) -> RouteResponse: """ The following parameters are not yet implemented: ResponseModels, ResponseParameters """ @@ -1284,35 +1330,39 @@ class ApiGatewayV2Backend(BaseBackend): response_models=response_models, ) - def delete_route_response(self, api_id, route_id, route_response_id): + def delete_route_response( + self, api_id: str, route_id: str, route_response_id: str + ) -> None: api = self.get_api(api_id) api.delete_route_response(route_id, route_response_id) - def get_route_response(self, api_id, route_id, route_response_id): + def get_route_response( + self, api_id: str, route_id: str, route_response_id: str + ) -> RouteResponse: api = self.get_api(api_id) return api.get_route_response(route_id, route_response_id) def create_integration( self, - api_id, - connection_id, - connection_type, - content_handling_strategy, - credentials_arn, - description, - integration_method, - integration_subtype, - integration_type, - integration_uri, - passthrough_behavior, - payload_format_version, - request_parameters, - request_templates, - response_parameters, - template_selection_expression, - timeout_in_millis, - tls_config, - ): + api_id: str, + connection_id: str, + connection_type: str, + content_handling_strategy: str, + credentials_arn: str, + description: str, + integration_method: str, + integration_subtype: str, + integration_type: str, + integration_uri: str, + passthrough_behavior: str, + payload_format_version: str, + request_parameters: Optional[Dict[str, str]], + request_templates: Optional[Dict[str, str]], + response_parameters: Optional[Dict[str, Dict[str, str]]], + template_selection_expression: str, + timeout_in_millis: str, + tls_config: Dict[str, str], + ) -> Integration: api = self.get_api(api_id) integration = api.create_integration( connection_id=connection_id, @@ -1335,44 +1385,44 @@ class ApiGatewayV2Backend(BaseBackend): ) return integration - def get_integration(self, api_id, integration_id): + def get_integration(self, api_id: str, integration_id: str) -> Integration: api = self.get_api(api_id) integration = api.get_integration(integration_id) return integration - def get_integrations(self, api_id): + def get_integrations(self, api_id: str) -> List[Integration]: """ Pagination is not yet implemented """ api = self.get_api(api_id) return api.get_integrations() - def delete_integration(self, api_id, integration_id): + def delete_integration(self, api_id: str, integration_id: str) -> None: api = self.get_api(api_id) api.delete_integration(integration_id) def update_integration( self, - api_id, - connection_id, - connection_type, - content_handling_strategy, - credentials_arn, - description, - integration_id, - integration_method, - integration_subtype, - integration_type, - integration_uri, - passthrough_behavior, - payload_format_version, - request_parameters, - request_templates, - response_parameters, - template_selection_expression, - timeout_in_millis, - tls_config, - ): + api_id: str, + connection_id: str, + connection_type: str, + content_handling_strategy: str, + credentials_arn: str, + description: str, + integration_id: str, + integration_method: str, + integration_subtype: str, + integration_type: str, + integration_uri: str, + passthrough_behavior: str, + payload_format_version: str, + request_parameters: Dict[str, str], + request_templates: Dict[str, str], + response_parameters: Dict[str, Dict[str, str]], + template_selection_expression: str, + timeout_in_millis: Optional[int], + tls_config: Dict[str, str], + ) -> Integration: api = self.get_api(api_id) integration = api.update_integration( integration_id=integration_id, @@ -1398,14 +1448,14 @@ class ApiGatewayV2Backend(BaseBackend): def create_integration_response( self, - api_id, - integration_id, - content_handling_strategy, - integration_response_key, - response_parameters, - response_templates, - template_selection_expression, - ): + api_id: str, + integration_id: str, + content_handling_strategy: str, + integration_response_key: str, + response_parameters: str, + response_templates: str, + template_selection_expression: str, + ) -> IntegrationResponse: api = self.get_api(api_id) integration_response = api.create_integration_response( integration_id=integration_id, @@ -1418,34 +1468,38 @@ class ApiGatewayV2Backend(BaseBackend): return integration_response def delete_integration_response( - self, api_id, integration_id, integration_response_id - ): + self, api_id: str, integration_id: str, integration_response_id: str + ) -> None: api = self.get_api(api_id) api.delete_integration_response( integration_id, integration_response_id=integration_response_id ) - def get_integration_response(self, api_id, integration_id, integration_response_id): + def get_integration_response( + self, api_id: str, integration_id: str, integration_response_id: str + ) -> IntegrationResponse: api = self.get_api(api_id) return api.get_integration_response( integration_id, integration_response_id=integration_response_id ) - def get_integration_responses(self, api_id, integration_id): + def get_integration_responses( + self, api_id: str, integration_id: str + ) -> List[IntegrationResponse]: api = self.get_api(api_id) return api.get_integration_responses(integration_id) def update_integration_response( self, - api_id, - integration_id, - integration_response_id, - content_handling_strategy, - integration_response_key, - response_parameters, - response_templates, - template_selection_expression, - ): + api_id: str, + integration_id: str, + integration_response_id: str, + content_handling_strategy: str, + integration_response_key: str, + response_parameters: str, + response_templates: str, + template_selection_expression: str, + ) -> IntegrationResponse: api = self.get_api(api_id) integration_response = api.update_integration_response( integration_id=integration_id, @@ -1458,25 +1512,27 @@ class ApiGatewayV2Backend(BaseBackend): ) return integration_response - def create_vpc_link(self, name, sg_ids, subnet_ids, tags): + def create_vpc_link( + self, name: str, sg_ids: List[str], subnet_ids: List[str], tags: Dict[str, str] + ) -> VpcLink: vpc_link = VpcLink( name, sg_ids=sg_ids, subnet_ids=subnet_ids, tags=tags, backend=self ) self.vpc_links[vpc_link.id] = vpc_link return vpc_link - def get_vpc_link(self, vpc_link_id): + def get_vpc_link(self, vpc_link_id: str) -> VpcLink: if vpc_link_id not in self.vpc_links: raise VpcLinkNotFound(vpc_link_id) return self.vpc_links[vpc_link_id] - def delete_vpc_link(self, vpc_link_id): + def delete_vpc_link(self, vpc_link_id: str) -> None: self.vpc_links.pop(vpc_link_id, None) - def get_vpc_links(self): - return self.vpc_links.values() + def get_vpc_links(self) -> List[VpcLink]: + return list(self.vpc_links.values()) - def update_vpc_link(self, vpc_link_id, name): + def update_vpc_link(self, vpc_link_id: str, name: str) -> VpcLink: vpc_link = self.get_vpc_link(vpc_link_id) vpc_link.update(name) return vpc_link diff --git a/moto/apigatewayv2/responses.py b/moto/apigatewayv2/responses.py index f17e8c74e..a9659c25e 100644 --- a/moto/apigatewayv2/responses.py +++ b/moto/apigatewayv2/responses.py @@ -2,24 +2,28 @@ import json from moto.core.responses import BaseResponse +from typing import Any, Tuple, Dict from urllib.parse import unquote from .exceptions import UnknownProtocol -from .models import apigatewayv2_backends +from .models import apigatewayv2_backends, ApiGatewayV2Backend + + +RESPONSE_TYPE = Tuple[int, Dict[str, str], str] class ApiGatewayV2Response(BaseResponse): """Handler for ApiGatewayV2 requests and responses.""" - def __init__(self): + def __init__(self) -> None: super().__init__(service_name="apigatewayv2") @property - def apigatewayv2_backend(self): + def apigatewayv2_backend(self) -> ApiGatewayV2Backend: """Return backend instance specific for this region.""" return apigatewayv2_backends[self.current_account][self.region] - def apis(self, request, full_url, headers): + def apis(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "POST": @@ -27,7 +31,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "GET": return self.get_apis() - def api(self, request, full_url, headers): + def api(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "GET": @@ -39,7 +43,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "DELETE": return self.delete_api() - def authorizer(self, request, full_url, headers): + def authorizer(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "DELETE": @@ -49,25 +53,25 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "PATCH": return self.update_authorizer() - def authorizers(self, request, full_url, headers): + def authorizers(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "POST": return self.create_authorizer() - def cors(self, request, full_url, headers): + def cors(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # 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, full_url, headers): + def route_request_parameter(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "DELETE": return self.delete_route_request_parameter() - def model(self, request, full_url, headers): + def model(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "DELETE": @@ -77,13 +81,13 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "PATCH": return self.update_model() - def models(self, request, full_url, headers): + def models(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "POST": return self.create_model() - def integration(self, request, full_url, headers): + def integration(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "DELETE": @@ -93,7 +97,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "PATCH": return self.update_integration() - def integrations(self, request, full_url, headers): + def integrations(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "GET": @@ -101,7 +105,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "POST": return self.create_integration() - def integration_response(self, request, full_url, headers): + def integration_response(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "DELETE": @@ -111,7 +115,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "PATCH": return self.update_integration_response() - def integration_responses(self, request, full_url, headers): + def integration_responses(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "GET": @@ -119,7 +123,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "POST": return self.create_integration_response() - def route(self, request, full_url, headers): + def route(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "DELETE": @@ -129,7 +133,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "PATCH": return self.update_route() - def routes(self, request, full_url, headers): + def routes(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "GET": @@ -137,7 +141,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "POST": return self.create_route() - def route_response(self, request, full_url, headers): + def route_response(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "DELETE": @@ -145,13 +149,13 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "GET": return self.get_route_response() - def route_responses(self, request, full_url, headers): + def route_responses(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "POST": return self.create_route_response() - def tags(self, request, full_url, headers): + def tags(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if self.method == "POST": @@ -161,7 +165,7 @@ class ApiGatewayV2Response(BaseResponse): if self.method == "DELETE": return self.untag_resource() - def vpc_link(self, request, full_url, headers): + def vpc_link(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if request.method == "DELETE": @@ -171,7 +175,7 @@ class ApiGatewayV2Response(BaseResponse): if request.method == "PATCH": return self.update_vpc_link() - def vpc_links(self, request, full_url, headers): + def vpc_links(self, request: Any, full_url: str, headers: Any) -> RESPONSE_TYPE: # type: ignore[return] self.setup_class(request, full_url, headers) if request.method == "GET": @@ -179,7 +183,7 @@ class ApiGatewayV2Response(BaseResponse): if request.method == "POST": return self.create_vpc_link() - def create_api(self): + def create_api(self) -> RESPONSE_TYPE: params = json.loads(self.body) api_key_selection_expression = params.get("apiKeySelectionExpression") @@ -210,21 +214,21 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(api.to_json()) - def delete_api(self): + def delete_api(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-1] self.apigatewayv2_backend.delete_api(api_id=api_id) - return 200, "", "{}" + return 200, {}, "{}" - def get_api(self): + def get_api(self) -> RESPONSE_TYPE: 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): + def get_apis(self) -> RESPONSE_TYPE: apis = self.apigatewayv2_backend.get_apis() return 200, {}, json.dumps({"items": [a.to_json() for a in apis]}) - def update_api(self): + def update_api(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-1] params = json.loads(self.body) api_key_selection_expression = params.get("apiKeySelectionExpression") @@ -248,7 +252,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(api.to_json()) - def reimport_api(self): + def reimport_api(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-1] params = json.loads(self.body) body = params.get("body") @@ -259,7 +263,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): + def create_authorizer(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-2] params = json.loads(self.body) @@ -288,21 +292,21 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(authorizer.to_json()) - def delete_authorizer(self): + def delete_authorizer(self) -> RESPONSE_TYPE: 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): + def get_authorizer(self) -> RESPONSE_TYPE: 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): + def update_authorizer(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-3] authorizer_id = self.path.split("/")[-1] params = json.loads(self.body) @@ -333,12 +337,12 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(authorizer.to_json()) - def delete_cors_configuration(self): + def delete_cors_configuration(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-2] self.apigatewayv2_backend.delete_cors_configuration(api_id) return 200, {}, "{}" - def create_model(self): + def create_model(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-2] params = json.loads(self.body) @@ -351,21 +355,21 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(model.to_json()) - def delete_model(self): + def delete_model(self) -> RESPONSE_TYPE: 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): + def get_model(self) -> RESPONSE_TYPE: 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): + def update_model(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-3] model_id = self.path.split("/")[-1] params = json.loads(self.body) @@ -385,27 +389,27 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(model.to_json()) - def get_tags(self): + def get_tags(self) -> RESPONSE_TYPE: 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): + def tag_resource(self) -> RESPONSE_TYPE: 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): + def untag_resource(self) -> RESPONSE_TYPE: resource_arn = unquote(self.path.split("/tags/")[1]) - tag_keys = self.querystring.get("tagKeys") + tag_keys = self.querystring.get("tagKeys") or [] self.apigatewayv2_backend.untag_resource(resource_arn, tag_keys) return 200, {}, "{}" - def create_route(self): + def create_route(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-2] params = json.loads(self.body) - api_key_required = params.get("apiKeyRequired", False) + api_key_required: bool = params.get("apiKeyRequired", False) authorization_scopes = params.get("authorizationScopes") authorization_type = params.get("authorizationType", "NONE") authorizer_id = params.get("authorizerId") @@ -434,13 +438,13 @@ class ApiGatewayV2Response(BaseResponse): ) return 201, {}, json.dumps(route.to_json()) - def delete_route(self): + def delete_route(self) -> RESPONSE_TYPE: 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): + def delete_route_request_parameter(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-5] route_id = self.path.split("/")[-3] request_param = self.path.split("/")[-1] @@ -449,18 +453,18 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, "{}" - def get_route(self): + def get_route(self) -> RESPONSE_TYPE: 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): + def get_routes(self) -> RESPONSE_TYPE: 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): + def update_route(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-3] route_id = self.path.split("/")[-1] @@ -495,7 +499,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(api.to_json()) - def create_route_response(self): + def create_route_response(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-4] route_id = self.path.split("/")[-2] params = json.loads(self.body) @@ -512,7 +516,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(route_response.to_json()) - def delete_route_response(self): + def delete_route_response(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-5] route_id = self.path.split("/")[-3] route_response_id = self.path.split("/")[-1] @@ -522,7 +526,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, "{}" - def get_route_response(self): + def get_route_response(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-5] route_id = self.path.split("/")[-3] route_response_id = self.path.split("/")[-1] @@ -532,7 +536,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(route_response.to_json()) - def create_integration(self): + def create_integration(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-2] params = json.loads(self.body) @@ -575,7 +579,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(integration.to_json()) - def get_integration(self): + def get_integration(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-3] integration_id = self.path.split("/")[-1] @@ -584,13 +588,13 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(integration.to_json()) - def get_integrations(self): + def get_integrations(self) -> RESPONSE_TYPE: 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): + def delete_integration(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-3] integration_id = self.path.split("/")[-1] @@ -599,7 +603,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, "{}" - def update_integration(self): + def update_integration(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-3] integration_id = self.path.split("/")[-1] @@ -644,7 +648,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(integration.to_json()) - def create_integration_response(self): + def create_integration_response(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-4] int_id = self.path.split("/")[-2] @@ -665,7 +669,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(integration_response.to_json()) - def delete_integration_response(self): + def delete_integration_response(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-5] int_id = self.path.split("/")[-3] int_res_id = self.path.split("/")[-1] @@ -675,7 +679,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, "{}" - def get_integration_response(self): + def get_integration_response(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-5] int_id = self.path.split("/")[-3] int_res_id = self.path.split("/")[-1] @@ -685,7 +689,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(int_response.to_json()) - def get_integration_responses(self): + def get_integration_responses(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-4] int_id = self.path.split("/")[-2] @@ -694,7 +698,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps({"items": [res.to_json() for res in int_response]}) - def update_integration_response(self): + def update_integration_response(self) -> RESPONSE_TYPE: api_id = self.path.split("/")[-5] int_id = self.path.split("/")[-3] int_res_id = self.path.split("/")[-1] @@ -717,7 +721,7 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(integration_response.to_json()) - def create_vpc_link(self): + def create_vpc_link(self) -> RESPONSE_TYPE: params = json.loads(self.body) name = params.get("name") @@ -729,21 +733,21 @@ class ApiGatewayV2Response(BaseResponse): ) return 200, {}, json.dumps(vpc_link.to_json()) - def delete_vpc_link(self): + def delete_vpc_link(self) -> RESPONSE_TYPE: vpc_link_id = self.path.split("/")[-1] self.apigatewayv2_backend.delete_vpc_link(vpc_link_id) return 200, {}, "{}" - def get_vpc_link(self): + def get_vpc_link(self) -> RESPONSE_TYPE: 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): + def get_vpc_links(self) -> RESPONSE_TYPE: 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): + def update_vpc_link(self) -> RESPONSE_TYPE: vpc_link_id = self.path.split("/")[-1] params = json.loads(self.body) name = params.get("name") diff --git a/setup.cfg b/setup.cfg index 2ae771c60..8233837a8 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/applicationautoscaling/ +files= moto/acm,moto/amp,moto/apigateway,moto/apigatewayv2,moto/applicationautoscaling/ show_column_numbers=True show_error_codes = True disable_error_code=abstract