APIGatewayV2 mappings (#5711)

IMPLEMENTATION_COVERAGE.md

@@ -175,13 +175,13 @@
 
 ## apigatewayv2
 <details>
-<summary>58% implemented</summary>
+<summary>69% implemented</summary>
 
 - [X] create_api
-- [ ] create_api_mapping
+- [X] create_api_mapping
 - [X] create_authorizer
 - [ ] create_deployment
-- [ ] create_domain_name
+- [X] create_domain_name
 - [X] create_integration
 - [X] create_integration_response
 - [X] create_model
@@ -191,11 +191,11 @@
 - [X] create_vpc_link
 - [ ] delete_access_log_settings
 - [X] delete_api
-- [ ] delete_api_mapping
+- [X] delete_api_mapping
 - [X] delete_authorizer
 - [X] delete_cors_configuration
 - [ ] delete_deployment
-- [ ] delete_domain_name
+- [X] delete_domain_name
 - [X] delete_integration
 - [X] delete_integration_response
 - [X] delete_model
@@ -207,15 +207,15 @@
 - [X] delete_vpc_link
 - [ ] export_api
 - [X] get_api
-- [ ] get_api_mapping
-- [ ] get_api_mappings
+- [X] get_api_mapping
+- [X] get_api_mappings
 - [X] get_apis
 - [X] get_authorizer
 - [ ] get_authorizers
 - [ ] get_deployment
 - [ ] get_deployments
-- [ ] get_domain_name
-- [ ] get_domain_names
+- [X] get_domain_name
+- [X] get_domain_names
 - [X] get_integration
 - [X] get_integration_response
 - [X] get_integration_responses

docs/docs/services/apigatewayv2.rst

@@ -33,10 +33,10 @@ apigatewayv2
         CredentialsArn, RouteKey, Tags, Target
 
 
-- [ ] create_api_mapping
+- [X] create_api_mapping
 - [X] create_authorizer
 - [ ] create_deployment
-- [ ] create_domain_name
+- [X] create_domain_name
 - [X] create_integration
 - [X] create_integration_response
 - [X] create_model
@@ -50,11 +50,11 @@ apigatewayv2
 - [X] create_vpc_link
 - [ ] delete_access_log_settings
 - [X] delete_api
-- [ ] delete_api_mapping
+- [X] delete_api_mapping
 - [X] delete_authorizer
 - [X] delete_cors_configuration
 - [ ] delete_deployment
-- [ ] delete_domain_name
+- [X] delete_domain_name
 - [X] delete_integration
 - [X] delete_integration_response
 - [X] delete_model
@@ -66,8 +66,8 @@ apigatewayv2
 - [X] delete_vpc_link
 - [ ] export_api
 - [X] get_api
-- [ ] get_api_mapping
-- [ ] get_api_mappings
+- [X] get_api_mapping
+- [X] get_api_mappings
 - [X] get_apis
 
         Pagination is not yet implemented
@@ -77,8 +77,12 @@ apigatewayv2
 - [ ] get_authorizers
 - [ ] get_deployment
 - [ ] get_deployments
-- [ ] get_domain_name
-- [ ] get_domain_names
+- [X] get_domain_name
+- [X] get_domain_names
+
+        Pagination is not yet implemented
+
+
 - [X] get_integration
 - [X] get_integration_response
 - [X] get_integration_responses

moto/apigatewayv2/exceptions.py

@@ -93,3 +93,33 @@ def __init__(self) -> None:
             "BadRequestException",
             "Invalid protocol specified. Must be one of [HTTP, WEBSOCKET]",
         )
+
+
+class DomainNameNotFound(APIGatewayV2Error):
+    code = 404
+
+    def __init__(self) -> None:
+        super().__init__(
+            "NotFoundException",
+            "The domain name resource specified in the request was not found.",
+        )
+
+
+class DomainNameAlreadyExists(APIGatewayV2Error):
+    code = 409
+
+    def __init__(self) -> None:
+        super().__init__(
+            "ConflictException",
+            "The domain name resource already exists.",
+        )
+
+
+class ApiMappingNotFound(APIGatewayV2Error):
+    code = 404
+
+    def __init__(self) -> None:
+        super().__init__(
+            "NotFoundException",
+            "The api mapping resource specified in the request was not found.",
+        )

moto/apigatewayv2/models.py

@@ -1,14 +1,16 @@
 """ApiGatewayV2Backend class with methods for supported APIs."""
+import hashlib
 import string
 import yaml
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Union
 
 from moto.core import BaseBackend, BackendDict, BaseModel
 from moto.core.utils import unix_time
 from moto.moto_api._internal import mock_random as random
 from moto.utilities.tagging_service import TaggingService
 
 from .exceptions import (
+    ApiMappingNotFound,
     ApiNotFound,
     AuthorizerNotFound,
     BadRequestException,
@@ -18,6 +20,8 @@
     IntegrationResponseNotFound,
     RouteNotFound,
     VpcLinkNotFound,
+    DomainNameNotFound,
+    DomainNameAlreadyExists,
 )
 
 
@@ -1011,13 +1015,64 @@ def to_json(self) -> Dict[str, Any]:
         }
 
 
+class DomainName(BaseModel):
+    def __init__(
+        self,
+        domain_name: str,
+        domain_name_configurations: List[Dict[str, str]],
+        mutual_tls_authentication: Dict[str, str],
+        tags: Dict[str, str],
+    ):
+        self.api_mapping_selection_expression = "$request.basepath"
+        self.domain_name = domain_name
+        self.domain_name_configurations = domain_name_configurations
+        self.mutual_tls_authentication = mutual_tls_authentication
+        self.tags = tags
+
+    def to_json(self) -> Dict[str, Any]:
+        return {
+            "apiMappingSelectionExpression": self.api_mapping_selection_expression,
+            "domainName": self.domain_name,
+            "domainNameConfigurations": self.domain_name_configurations,
+            "mutualTlsAuthentication": self.mutual_tls_authentication,
+            "tags": self.tags,
+        }
+
+
+class ApiMapping(BaseModel):
+    def __init__(
+        self,
+        api_id: str,
+        api_mapping_key: str,
+        api_mapping_id: str,
+        domain_name: str,
+        stage: str,
+    ) -> None:
+        self.api_id = api_id
+        self.api_mapping_key = api_mapping_key
+        self.api_mapping_id = api_mapping_id
+        self.domain_name = domain_name
+        self.stage = stage
+
+    def to_json(self) -> Dict[str, Any]:
+        return {
+            "apiId": self.api_id,
+            "apiMappingId": self.api_mapping_id,
+            "apiMappingKey": self.api_mapping_key,
+            "domainName": self.domain_name,
+            "stage": self.stage,
+        }
+
+
 class ApiGatewayV2Backend(BaseBackend):
     """Implementation of ApiGatewayV2 APIs."""
 
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
         self.apis: Dict[str, Api] = dict()
         self.vpc_links: Dict[str, VpcLink] = dict()
+        self.domain_names: Dict[str, DomainName] = dict()
+        self.api_mappings: Dict[str, ApiMapping] = dict()
         self.tagger = TaggingService()
 
     def create_api(
@@ -1537,5 +1592,114 @@ def update_vpc_link(self, vpc_link_id: str, name: str) -> VpcLink:
         vpc_link.update(name)
         return vpc_link
 
+    def create_domain_name(
+        self,
+        domain_name: str,
+        domain_name_configurations: List[Dict[str, str]],
+        mutual_tls_authentication: Dict[str, str],
+        tags: Dict[str, str],
+    ) -> DomainName:
+
+        if domain_name in self.domain_names.keys():
+            raise DomainNameAlreadyExists
+
+        domain = DomainName(
+            domain_name=domain_name,
+            domain_name_configurations=domain_name_configurations,
+            mutual_tls_authentication=mutual_tls_authentication,
+            tags=tags,
+        )
+        self.domain_names[domain.domain_name] = domain
+        return domain
+
+    def get_domain_name(self, domain_name: Union[str, None]) -> DomainName:
+        if domain_name is None or domain_name not in self.domain_names:
+            raise DomainNameNotFound
+        return self.domain_names[domain_name]
+
+    def get_domain_names(self) -> List[DomainName]:
+        """
+        Pagination is not yet implemented
+        """
+        return list(self.domain_names.values())
+
+    def delete_domain_name(self, domain_name: str) -> None:
+        if domain_name not in self.domain_names.keys():
+            raise DomainNameNotFound
+
+        for mapping_id, mapping in self.api_mappings.items():
+            if mapping.domain_name == domain_name:
+                del self.api_mappings[mapping_id]
+
+        del self.domain_names[domain_name]
+
+    def _generate_api_maping_id(
+        self, api_mapping_key: str, stage: str, domain_name: str
+    ) -> str:
+        return str(
+            hashlib.sha256(
+                f"{stage} {domain_name}/{api_mapping_key}".encode("utf-8")
+            ).hexdigest()
+        )[:5]
+
+    def create_api_mapping(
+        self, api_id: str, api_mapping_key: str, domain_name: str, stage: str
+    ) -> ApiMapping:
+        if domain_name not in self.domain_names.keys():
+            raise DomainNameNotFound
+
+        if api_id not in self.apis.keys():
+            raise ApiNotFound("The resource specified in the request was not found.")
+
+        if api_mapping_key.startswith("/") or "//" in api_mapping_key:
+            raise BadRequestException(
+                "API mapping key should not start with a '/' or have consecutive '/'s."
+            )
+
+        if api_mapping_key.endswith("/"):
+            raise BadRequestException("API mapping key should not end with a '/'.")
+
+        api_mapping_id = self._generate_api_maping_id(
+            api_mapping_key=api_mapping_key, stage=stage, domain_name=domain_name
+        )
+
+        mapping = ApiMapping(
+            domain_name=domain_name,
+            api_id=api_id,
+            api_mapping_key=api_mapping_key,
+            api_mapping_id=api_mapping_id,
+            stage=stage,
+        )
+
+        self.api_mappings[api_mapping_id] = mapping
+        return mapping
+
+    def get_api_mapping(self, api_mapping_id: str, domain_name: str) -> ApiMapping:
+        if domain_name not in self.domain_names.keys():
+            raise DomainNameNotFound
+
+        if api_mapping_id not in self.api_mappings.keys():
+            raise ApiMappingNotFound
+
+        return self.api_mappings[api_mapping_id]
+
+    def get_api_mappings(self, domain_name: str) -> List[ApiMapping]:
+        domain_mappings = []
+        for mapping in self.api_mappings.values():
+            if mapping.domain_name == domain_name:
+                domain_mappings.append(mapping)
+        return domain_mappings
+
+    def delete_api_mapping(self, api_mapping_id: str, domain_name: str) -> None:
+        if api_mapping_id not in self.api_mappings.keys():
+            raise ApiMappingNotFound
+
+        if self.api_mappings[api_mapping_id].domain_name != domain_name:
+            raise BadRequestException(
+                f"given domain name {domain_name} does not match with mapping definition of mapping {api_mapping_id}"
+            )
+
+        del self.api_mappings[api_mapping_id]
+
 
 apigatewayv2_backends = BackendDict(ApiGatewayV2Backend, "apigatewayv2")