DSQL: implement create_cluster()

moto/backend_index.py

@@ -45,6 +45,7 @@
     ("directconnect", re.compile("https?://directconnect\\.(.+)\\.amazonaws\\.com")),
     ("dms", re.compile("https?://dms\\.(.+)\\.amazonaws\\.com")),
     ("ds", re.compile("https?://ds\\.(.+)\\.amazonaws\\.com")),
+    ("dsql", re.compile("https?://dsql\\.(.+)\\.api\\.aws")),
     ("dynamodb", re.compile("https?://dynamodb\\.(.+)\\.amazonaws\\.com")),
     (
         "dynamodbstreams",

moto/backends.py

@@ -44,6 +44,7 @@
     from moto.directconnect.models import DirectConnectBackend
     from moto.dms.models import DatabaseMigrationServiceBackend
     from moto.ds.models import DirectoryServiceBackend
+    from moto.dsql.models import AuroraDSQLBackend
     from moto.dynamodb.models import DynamoDBBackend
     from moto.dynamodb_v20111205.models import (
         DynamoDBBackend as DynamoDBBackend_v20111205,
@@ -225,6 +226,7 @@ def get_service_from_url(url: str) -> Optional[str]:
     "Literal['directconnect']",
     "Literal['dms']",
     "Literal['ds']",
+    "Literal['dsql']",
     "Literal['dynamodb']",
     "Literal['dynamodb_v20111205']",
     "Literal['dynamodbstreams']",
@@ -440,6 +442,8 @@ def get_backend(
 @overload
 def get_backend(name: "Literal['ds']") -> "BackendDict[DirectoryServiceBackend]": ...
 @overload
+def get_backend(name: "Literal['dsql']") -> "BackendDict[AuroraDSQLBackend]": ...
+@overload
 def get_backend(name: "Literal['dynamodb']") -> "BackendDict[DynamoDBBackend]": ...
 @overload
 def get_backend(

moto/dsql/__init__.py

@@ -0,0 +1 @@
+from .models import dsql_backends  # noqa: F401

moto/dsql/exceptions.py

@@ -0,0 +1 @@
+"""Exceptions raised by the dsql service."""

moto/dsql/models.py

@@ -0,0 +1,81 @@
+"""AuroraDSQLBackend class with methods for supported APIs."""
+
+from collections import OrderedDict
+from typing import Any, Dict, Optional
+
+from moto.core.base_backend import BackendDict, BaseBackend
+from moto.core.common_models import BaseModel
+from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow
+from moto.moto_api._internal import mock_random
+from moto.moto_api._internal.managed_state_model import ManagedState
+from moto.utilities.utils import get_partition
+
+
+class Cluster(BaseModel, ManagedState):
+    """Model for an AuroraDSQL cluster."""
+
+    def to_dict(self) -> Dict[str, Any]:
+        dct = {
+            "identifier": self.identifier,
+            "arn": self.arn,
+            "status": self.status,
+            "creationTime": iso_8601_datetime_with_milliseconds(self.creation_time),
+            "deletionProtectionEnabled": self.deletion_protection_enabled,
+        }
+        return {k: v for k, v in dct.items() if v is not None}
+
+    def __init__(
+        self,
+        region_name: str,
+        account_id: str,
+        deletion_protection_enabled: Optional[bool],
+        tags: Optional[Dict[str, str]],
+        client_token: Optional[str],
+    ):
+        ManagedState.__init__(
+            self, "dsql::cluster", transitions=[("CREATING", "ACTIVE")]
+        )
+        self.region_name = region_name
+        self.account_id = account_id
+        self.identifier = mock_random.get_random_hex(26)
+        self.arn = f"arn:{get_partition(self.region_name)}:dsql:{self.region_name}:{self.account_id}:cluster/{self.identifier}"
+        self.creation_time = utcnow()
+        self.deletion_protection_enabled = deletion_protection_enabled
+        self.tags = tags
+        self.client_token = client_token
+
+
+class AuroraDSQLBackend(BaseBackend):
+    """Implementation of AuroraDSQL APIs."""
+
+    def __init__(self, region_name: str, account_id: str):
+        super().__init__(region_name, account_id)
+        self.region_name = region_name
+        self.account_id = account_id
+        self.partition = get_partition(region_name)
+        self.clusters: Dict[str, Cluster] = OrderedDict()
+
+    def create_cluster(
+        self,
+        deletion_protection_enabled: bool,
+        tags: Optional[Dict[str, str]],
+        client_token: Optional[str],
+    ) -> Cluster:
+        cluster = Cluster(
+            self.region_name,
+            self.account_id,
+            deletion_protection_enabled,
+            tags,
+            client_token,
+        )
+        return cluster
+
+
+dsql_backends = BackendDict(
+    AuroraDSQLBackend,
+    "dsql",
+    # currently botocore does not provide a dsql endpoint
+    # https://github.com/boto/botocore/blob/e07cddc333fe4fb90efcd5d04324dd83f9cc3a57/botocore/data/endpoints.json
+    use_boto3_regions=False,
+    additional_regions=["us-east-1", "us-east-2"],
+)

moto/dsql/responses.py

@@ -0,0 +1,32 @@
+"""Handles incoming dsql requests, invokes methods, returns responses."""
+
+import json
+
+from moto.core.responses import BaseResponse
+
+from .models import AuroraDSQLBackend, dsql_backends
+
+
+class AuroraDSQLResponse(BaseResponse):
+    """Handler for AuroraDSQL requests and responses."""
+
+    def __init__(self) -> None:
+        super().__init__(service_name="dsql")
+
+    @property
+    def dsql_backend(self) -> AuroraDSQLBackend:
+        """Return backend instance specific for this region."""
+        return dsql_backends[self.current_account][self.region]
+
+    def create_cluster(self) -> str:
+        params = self._get_params()
+        deletion_protection_enabled = params.get("deletionProtectionEnabled", True)
+        tags = params.get("tags")
+        client_token = params.get("clientToken")
+        cluster = self.dsql_backend.create_cluster(
+            deletion_protection_enabled=deletion_protection_enabled,
+            tags=tags,
+            client_token=client_token,
+        )
+
+        return json.dumps(dict(cluster.to_dict()))

moto/dsql/urls.py

@@ -0,0 +1,11 @@
+"""dsql base URL and path."""
+
+from .responses import AuroraDSQLResponse
+
+url_bases = [
+    r"https?://dsql\.(.+)\.api\.aws",
+]
+
+url_paths = {
+    "{0}/cluster$": AuroraDSQLResponse.dispatch,
+}

moto/moto_api/__init__.py

@@ -18,6 +18,9 @@
 state_manager.register_default_transition(
     model_name="dax::cluster", transition={"progression": "manual", "times": 4}
 )
+state_manager.register_default_transition(
+    "dsql::cluster", transition={"progression": "manual", "times": 1}
+)
 state_manager.register_default_transition(
     model_name="ecs::task", transition={"progression": "manual", "times": 1}
 )

moto/moto_server/werkzeug_app.py

@@ -136,6 +136,8 @@
             # All MediaStore API calls have a target header
             # If no target is set, assume we're trying to reach the mediastore-data service
             host = f"data.{service}.{region}.amazonaws.com"
+        elif service == "dsql":
+            host = f"{service}.{region}.api.aws"
         elif service == "dynamodb":
             if environ["HTTP_X_AMZ_TARGET"].startswith("DynamoDBStreams"):
                 host = "dynamodbstreams"

tests/test_dsql/test_dsql.py

@@ -0,0 +1,24 @@
+"""Unit tests for dsql-supported APIs."""
+
+from datetime import datetime
+
+import boto3
+from dateutil.tz import tzutc
+from freezegun import freeze_time
+
+from moto import mock_aws, settings
+
+
+@mock_aws
+def test_create_cluster():
+    client = boto3.client("dsql", region_name="us-east-1")
+    with freeze_time("2024-12-22 12:34:00"):
+        resp = client.create_cluster()
+
+    identifier = resp["identifier"]
+    assert identifier is not None
+    assert resp["arn"] == f"arn:aws:dsql:us-east-1:123456789012:cluster/{identifier}"
+    assert resp["deletionProtectionEnabled"] is True
+    assert resp["status"] == "CREATING"
+    if not settings.TEST_SERVER_MODE:
+        assert resp["creationTime"] == datetime(2024, 12, 22, 12, 34, tzinfo=tzutc())