implementation of ResourceIdentifier

moto/apigateway/models.py

@@ -68,7 +68,19 @@
     ValidationException,
     VpcLinkNotFound,
 )
-from .utils import create_apigw_id, create_id, to_path
+from .utils import (
+    ApigwApiKeyIdentifier,
+    ApigwAuthorizerIdentifier,
+    ApigwDeploymentIdentifier,
+    ApigwModelIdentifier,
+    ApigwRequestValidatorIdentifier,
+    ApigwResourceIdentifier,
+    ApigwRestApiValidatorIdentifier,
+    ApigwUsagePlanIdentifier,
+    ApigwVpcLinkIdentifier,
+    create_id,
+    to_path,
+)
 
 STAGE_URL = "https://{api_id}.execute-api.{region_name}.amazonaws.com/{stage_name}"
 PATCH_OPERATIONS = ["add", "remove", "replace", "move", "copy", "test"]
@@ -1165,12 +1177,9 @@ def create_from_cloudformation_json(  # type: ignore[misc]
         )
 
     def add_child(self, path: str, parent_id: Optional[str] = None) -> Resource:
-        child_id = create_apigw_id(
-            self.account_id,
-            self.region_name,
-            "resource",
-            (parent_id or "") + "." + path,
-        )
+        child_id = ApigwResourceIdentifier(
+            self.account_id, self.region_name, parent_id or "", path
+        ).generate()
         child = Resource(
             resource_id=child_id,
             account_id=self.account_id,
@@ -1189,7 +1198,9 @@ def add_model(
         schema: str,
         content_type: str,
     ) -> "Model":
-        model_id = create_apigw_id(self.account_id, self.region_name, "model", name)
+        model_id = ApigwModelIdentifier(
+            self.account_id, self.region_name, name
+        ).generate()
         new_model = Model(
             model_id=model_id,
             name=name,
@@ -1301,9 +1312,9 @@ def create_deployment(
     ) -> Deployment:
         if stage_variables is None:
             stage_variables = {}
-        deployment_id = create_apigw_id(
-            self.account_id, self.region_name, "deployment", name
-        )
+        deployment_id = ApigwDeploymentIdentifier(
+            self.account_id, self.region_name, name
+        ).generate()
         deployment = Deployment(deployment_id, name, description)
         self.deployments[deployment_id] = deployment
         if name:
@@ -1342,9 +1353,9 @@ def create_request_validator(
         validateRequestBody: Optional[bool],
         validateRequestParameters: Any,
     ) -> RequestValidator:
-        validator_id = create_apigw_id(
-            self.account_id, self.region_name, "request_validator", name
-        )
+        validator_id = ApigwRequestValidatorIdentifier(
+            self.account_id, self.region_name, name
+        ).generate()
         request_validator = RequestValidator(
             _id=validator_id,
             name=name,
@@ -1643,7 +1654,9 @@ def create_rest_api(
         minimum_compression_size: Optional[int] = None,
         disable_execute_api_endpoint: Optional[bool] = None,
     ) -> RestAPI:
-        api_id = create_apigw_id(self.account_id, self.region_name, "rest_api", name)
+        api_id = ApigwRestApiValidatorIdentifier(
+            self.account_id, self.region_name, name
+        ).generate()
         rest_api = RestAPI(
             api_id,
             self.account_id,
@@ -1894,9 +1907,9 @@ def create_authorizer(
         self, restapi_id: str, name: str, authorizer_type: str, **kwargs: Any
     ) -> Authorizer:
         api = self.get_rest_api(restapi_id)
-        authorizer_id = create_apigw_id(
-            self.account_id, self.region_name, "authorizer", name
-        )
+        authorizer_id = ApigwAuthorizerIdentifier(
+            self.account_id, self.region_name, name
+        ).generate()
         return api.create_authorizer(
             authorizer_id,
             name,
@@ -2160,9 +2173,12 @@ def create_api_key(self, payload: Dict[str, Any]) -> ApiKey:
             for api_key in self.get_api_keys():
                 if api_key.value == payload["value"]:
                     raise ApiKeyAlreadyExists()
-        api_key_id = create_apigw_id(
-            self.account_id, self.region_name, "api_key", payload["name"]
-        )
+        api_key_id = ApigwApiKeyIdentifier(
+            self.account_id,
+            self.region_name,
+            # The value of an api key must be unique on aws
+            payload["value"],
+        ).generate()
         key = ApiKey(api_key_id=api_key_id, **payload)
         self.keys[key.id] = key
         return key
@@ -2183,9 +2199,9 @@ def delete_api_key(self, api_key_id: str) -> None:
         self.keys.pop(api_key_id)
 
     def create_usage_plan(self, payload: Any) -> UsagePlan:
-        usage_plan_id = create_apigw_id(
-            self.account_id, self.region_name, "usage_plan", payload["name"]
-        )
+        usage_plan_id = ApigwUsagePlanIdentifier(
+            self.account_id, self.region_name, payload["name"]
+        ).generate()
         plan = UsagePlan(usage_plan_id=usage_plan_id, **payload)
         self.usage_plans[plan.id] = plan
         return plan
@@ -2509,9 +2525,9 @@ def create_vpc_link(
         target_arns: List[str],
         tags: List[Dict[str, str]],
     ) -> VpcLink:
-        vpc_link_id = create_apigw_id(
-            self.account_id, self.region_name, "vpc_link", name
-        )
+        vpc_link_id = ApigwVpcLinkIdentifier(
+            self.account_id, self.region_name, name
+        ).generate()
         vpc_link = VpcLink(
             vpc_link_id,
             name,

moto/apigateway/utils.py

@@ -5,20 +5,70 @@
 import yaml
 
 from moto.moto_api._internal import mock_random as random
-from moto.utilities.id_generator import generate_str_id
-
-
-def create_apigw_id(account_id: str, region: str, resource: str, name: str) -> str:
-    return generate_str_id(
-        account_id,
-        region,
-        "apigateway",
-        resource,
-        name,
-        length=10,
-        include_digits=True,
-        lower_case=True,
-    )
+from moto.utilities.id_generator import ResourceIdentifier, generate_str_id
+
+
+class ApigwIdentifier(ResourceIdentifier):
+    service = "apigateway"
+
+    def __init__(self, account_id: str, region: str, name: str):
+        super().__init__(account_id, region, name)
+
+    def generate(self) -> str:
+        return generate_str_id(
+            self,
+            length=10,
+            include_digits=True,
+            lower_case=True,
+        )
+
+
+class ApigwApiKeyIdentifier(ApigwIdentifier):
+    resource = "api_key"
+
+    def __init__(self, account_id: str, region: str, value: str):
+        super().__init__(account_id, region, value)
+
+
+class ApigwAuthorizerIdentifier(ApigwIdentifier):
+    resource = "authorizer"
+
+
+class ApigwDeploymentIdentifier(ApigwIdentifier):
+    resource = "deployment"
+
+
+class ApigwModelIdentifier(ApigwIdentifier):
+    resource = "model"
+
+
+class ApigwRequestValidatorIdentifier(ApigwIdentifier):
+    resource = "request_validator"
+
+
+class ApigwResourceIdentifier(ApigwIdentifier):
+    resource = "resource"
+
+    def __init__(
+        self, account_id: str, region: str, parent_id: str = "", path_name: str = "/"
+    ):
+        super().__init__(
+            account_id,
+            region,
+            ".".join((parent_id, path_name)),
+        )
+
+
+class ApigwRestApiValidatorIdentifier(ApigwIdentifier):
+    resource = "rest_api"
+
+
+class ApigwUsagePlanIdentifier(ApigwIdentifier):
+    resource = "usage_plan"
+
+
+class ApigwVpcLinkIdentifier(ApigwIdentifier):
+    resource = "vpc_link"
 
 
 def create_id() -> str:

moto/secretsmanager/models.py

@@ -26,7 +26,11 @@
     tag_key,
     tag_value,
 )
-from .utils import get_secret_name_from_partial_arn, random_password, secret_arn
+from .utils import (
+    SecretsManagerSecretIdentifier,
+    get_secret_name_from_partial_arn,
+    random_password,
+)
 
 MAX_RESULTS_DEFAULT = 100
 
@@ -94,7 +98,9 @@ def __init__(
     ):
         self.secret_id = secret_id
         self.name = secret_id
-        self.arn = secret_arn(account_id, region_name, secret_id)
+        self.arn = SecretsManagerSecretIdentifier(
+            account_id, region_name, secret_id
+        ).generate()
         self.account_id = account_id
         self.region = region_name
         self.secret_string = secret_string
@@ -935,7 +941,9 @@ def delete_secret(
             if not force_delete_without_recovery:
                 raise SecretNotFoundException()
             else:
-                arn = secret_arn(self.account_id, self.region_name, secret_id=secret_id)
+                arn = SecretsManagerSecretIdentifier(
+                    self.account_id, self.region_name, secret_id=secret_id
+                ).generate()
                 name = secret_id
                 deletion_date = utcnow()
                 return arn, name, self._unix_time_secs(deletion_date)

moto/secretsmanager/utils.py

@@ -2,7 +2,7 @@
 import string
 
 from moto.moto_api._internal import mock_random as random
-from moto.utilities.id_generator import generate_str_id
+from moto.utilities.id_generator import ResourceIdentifier, generate_str_id
 from moto.utilities.utils import ARN_PARTITION_REGEX, get_partition
 
 
@@ -63,19 +63,6 @@ def random_password(
     return password
 
 
-def secret_arn(account_id: str, region: str, secret_id: str) -> str:
-    id_string = generate_str_id(
-        account_id,
-        region,
-        "secretsmanager",
-        "secret",
-        secret_id,
-        length=6,
-        include_digits=False,
-    )
-    return f"arn:{get_partition(region)}:secretsmanager:{region}:{account_id}:secret:{secret_id}-{id_string}"
-
-
 def get_secret_name_from_partial_arn(partial_arn: str) -> str:
     # We can retrieve a secret either using a full ARN, or using a partial ARN
     # name:        testsecret
@@ -108,3 +95,20 @@ def _add_password_require_each_included_type(
     password_with_required_char += required_characters
 
     return password_with_required_char
+
+
+class SecretsManagerSecretIdentifier(ResourceIdentifier):
+    service = "secretsmanager"
+    resource = "secret"
+
+    def __init__(self, account_id: str, region: str, secret_id: str):
+        super().__init__(account_id, region, name=secret_id)
+
+    def generate(self) -> str:
+        id_string = generate_str_id(
+            resource_identifier=self, length=6, include_digits=False
+        )
+        return (
+            f"arn:{get_partition(self.region)}:secretsmanager:{self.region}:"
+            f"{self.account_id}:secret:{self.name}-{id_string}"
+        )