Appsync API Cache Implementation (#8115)

moto/appsync/exceptions.py

@@ -23,6 +23,17 @@ def __init__(self, message: str):
         self.description = json.dumps({"message": self.message})
 
 
+class GraphqlAPICacheNotFound(AppSyncExceptions):
+    code = 404
+
+    def __init__(self, op: str):
+        super().__init__(
+            "NotFoundException",
+            f"Unable to {op} the cache as it doesn't exist, please create the cache first.",
+        )
+        self.description = json.dumps({"message": self.message})
+
+
 class BadRequestException(AppSyncExceptions):
     def __init__(self, message: str):
         super().__init__("BadRequestException", message)

moto/appsync/models.py

@@ -10,7 +10,12 @@
 from moto.utilities.tagging_service import TaggingService
 from moto.utilities.utils import get_partition
 
-from .exceptions import BadRequestException, GraphqlAPINotFound, GraphQLSchemaException
+from .exceptions import (
+    BadRequestException,
+    GraphqlAPICacheNotFound,
+    GraphqlAPINotFound,
+    GraphQLSchemaException,
+)
 
 # AWS custom scalars and directives
 # https://github.com/dotansimha/graphql-code-generator/discussions/4311#discussioncomment-2921796
@@ -132,6 +137,49 @@ def to_json(self) -> Dict[str, Any]:
         }
 
 
+class APICache(BaseModel):
+    def __init__(
+        self,
+        ttl: int,
+        api_caching_behavior: str,
+        type_: str,
+        transit_encryption_enabled: Optional[bool] = None,
+        at_rest_encryption_enabled: Optional[bool] = None,
+        health_metrics_config: Optional[str] = None,
+    ):
+        self.ttl = ttl
+        self.api_caching_behavior = api_caching_behavior
+        self.type = type_
+        self.transit_encryption_enabled = transit_encryption_enabled or False
+        self.at_rest_encryption_enabled = at_rest_encryption_enabled or False
+        self.health_metrics_config = health_metrics_config or "DISABLED"
+        self.status = "AVAILABLE"
+
+    def update(
+        self,
+        ttl: int,
+        api_caching_behavior: str,
+        type: str,
+        health_metrics_config: Optional[str] = None,
+    ) -> None:
+        self.ttl = ttl
+        self.api_caching_behavior = api_caching_behavior
+        self.type = type
+        if health_metrics_config is not None:
+            self.health_metrics_config = health_metrics_config
+
+    def to_json(self) -> Dict[str, Any]:
+        return {
+            "ttl": self.ttl,
+            "transitEncryptionEnabled": self.transit_encryption_enabled,
+            "atRestEncryptionEnabled": self.at_rest_encryption_enabled,
+            "apiCachingBehavior": self.api_caching_behavior,
+            "type": self.type,
+            "healthMetricsConfig": self.health_metrics_config,
+            "status": self.status,
+        }
+
+
 class GraphqlAPI(BaseModel):
     def __init__(
         self,
@@ -162,6 +210,8 @@ def __init__(
 
         self.api_keys: Dict[str, GraphqlAPIKey] = dict()
 
+        self.api_cache: Optional[APICache] = None
+
     def update(
         self,
         name: str,
@@ -223,6 +273,38 @@ def get_type(self, type_name: str, type_format: str) -> Any:
         graphql_type["format"] = type_format  # type: ignore[index]
         return graphql_type
 
+    def create_api_cache(
+        self,
+        ttl: int,
+        api_caching_behavior: str,
+        type: str,
+        transit_encryption_enabled: Optional[bool] = None,
+        at_rest_encryption_enabled: Optional[bool] = None,
+        health_metrics_config: Optional[str] = None,
+    ) -> APICache:
+        self.api_cache = APICache(
+            ttl,
+            api_caching_behavior,
+            type,
+            transit_encryption_enabled,
+            at_rest_encryption_enabled,
+            health_metrics_config,
+        )
+        return self.api_cache
+
+    def update_api_cache(
+        self,
+        ttl: int,
+        api_caching_behavior: str,
+        type: str,
+        health_metrics_config: Optional[str] = None,
+    ) -> APICache:
+        self.api_cache.update(ttl, api_caching_behavior, type, health_metrics_config)  # type: ignore[union-attr]
+        return self.api_cache  # type: ignore[return-value]
+
+    def delete_api_cache(self) -> None:
+        self.api_cache = None
+
     def to_json(self) -> Dict[str, Any]:
         return {
             "name": self.name,
@@ -374,5 +456,71 @@ def list_tags_for_resource(self, resource_arn: str) -> Dict[str, str]:
     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)
 
+    def get_api_cache(self, api_id: str) -> APICache:
+        if api_id not in self.graphql_apis:
+            raise GraphqlAPINotFound(api_id)
+        api_cache = self.graphql_apis[api_id].api_cache
+        if api_cache is None:
+            raise GraphqlAPICacheNotFound("get")
+        return api_cache
+
+    def delete_api_cache(self, api_id: str) -> None:
+        if api_id not in self.graphql_apis:
+            raise GraphqlAPINotFound(api_id)
+        if self.graphql_apis[api_id].api_cache is None:
+            raise GraphqlAPICacheNotFound("delete")
+        self.graphql_apis[api_id].delete_api_cache()
+        return
+
+    def create_api_cache(
+        self,
+        api_id: str,
+        ttl: int,
+        api_caching_behavior: str,
+        type: str,
+        transit_encryption_enabled: Optional[bool] = None,
+        at_rest_encryption_enabled: Optional[bool] = None,
+        health_metrics_config: Optional[str] = None,
+    ) -> APICache:
+        if api_id not in self.graphql_apis:
+            raise GraphqlAPINotFound(api_id)
+        graphql_api = self.graphql_apis[api_id]
+        if graphql_api.api_cache is not None:
+            raise BadRequestException(message="The API has already enabled caching.")
+        api_cache = graphql_api.create_api_cache(
+            ttl,
+            api_caching_behavior,
+            type,
+            transit_encryption_enabled,
+            at_rest_encryption_enabled,
+            health_metrics_config,
+        )
+        return api_cache
+
+    def update_api_cache(
+        self,
+        api_id: str,
+        ttl: int,
+        api_caching_behavior: str,
+        type: str,
+        health_metrics_config: Optional[str] = None,
+    ) -> APICache:
+        if api_id not in self.graphql_apis:
+            raise GraphqlAPINotFound(api_id)
+        graphql_api = self.graphql_apis[api_id]
+        if graphql_api.api_cache is None:
+            raise GraphqlAPICacheNotFound("update")
+        api_cache = graphql_api.update_api_cache(
+            ttl, api_caching_behavior, type, health_metrics_config
+        )
+        return api_cache
+
+    def flush_api_cache(self, api_id: str) -> None:
+        if api_id not in self.graphql_apis:
+            raise GraphqlAPINotFound(api_id)
+        if self.graphql_apis[api_id].api_cache is None:
+            raise GraphqlAPICacheNotFound("flush")
+        return
+
 
 appsync_backends = BackendDict(AppSyncBackend, "appsync")

moto/appsync/responses.py

@@ -192,3 +192,60 @@ def get_introspection_schema(self) -> str:
             format_=format_, include_directives=include_directives
         )
         return schema
+
+    def get_api_cache(self) -> str:
+        api_id = self.path.split("/")[-2]
+        api_cache = self.appsync_backend.get_api_cache(
+            api_id=api_id,
+        )
+        return json.dumps(dict(apiCache=api_cache.to_json()))
+
+    def delete_api_cache(self) -> str:
+        api_id = self.path.split("/")[-2]
+        self.appsync_backend.delete_api_cache(
+            api_id=api_id,
+        )
+        return "{}"
+
+    def create_api_cache(self) -> str:
+        params = json.loads(self.body)
+        api_id = self.path.split("/")[-2]
+        ttl = params.get("ttl")
+        transit_encryption_enabled = params.get("transitEncryptionEnabled")
+        at_rest_encryption_enabled = params.get("atRestEncryptionEnabled")
+        api_caching_behavior = params.get("apiCachingBehavior")
+        type = params.get("type")
+        health_metrics_config = params.get("healthMetricsConfig")
+        api_cache = self.appsync_backend.create_api_cache(
+            api_id=api_id,
+            ttl=ttl,
+            transit_encryption_enabled=transit_encryption_enabled,
+            at_rest_encryption_enabled=at_rest_encryption_enabled,
+            api_caching_behavior=api_caching_behavior,
+            type=type,
+            health_metrics_config=health_metrics_config,
+        )
+        return json.dumps(dict(apiCache=api_cache.to_json()))
+
+    def update_api_cache(self) -> str:
+        api_id = self.path.split("/")[-3]
+        params = json.loads(self.body)
+        ttl = params.get("ttl")
+        api_caching_behavior = params.get("apiCachingBehavior")
+        type = params.get("type")
+        health_metrics_config = params.get("healthMetricsConfig")
+        api_cache = self.appsync_backend.update_api_cache(
+            api_id=api_id,
+            ttl=ttl,
+            api_caching_behavior=api_caching_behavior,
+            type=type,
+            health_metrics_config=health_metrics_config,
+        )
+        return json.dumps(dict(apiCache=api_cache.to_json()))
+
+    def flush_api_cache(self) -> str:
+        api_id = self.path.split("/")[-2]
+        self.appsync_backend.flush_api_cache(
+            api_id=api_id,
+        )
+        return "{}"

moto/appsync/urls.py

@@ -17,4 +17,7 @@
     "{0}/v1/tags/(?P<resource_arn>.+)$": AppSyncResponse.dispatch,
     "{0}/v1/tags/(?P<resource_arn_pt1>.+)/(?P<resource_arn_pt2>.+)$": AppSyncResponse.dispatch,
     "{0}/v1/apis/(?P<api_id>[^/]+)/types/(?P<type_name>.+)$": AppSyncResponse.dispatch,
+    "{0}/v1/apis/(?P<apiId>.*)/ApiCaches$": AppSyncResponse.dispatch,
+    "{0}/v1/apis/(?P<apiId>.*)/ApiCaches/update$": AppSyncResponse.dispatch,
+    "{0}/v1/apis/(?P<apiId>.*)/FlushCache$": AppSyncResponse.dispatch,
 }