Techdebt: Improve Glacier URL routing (#7688)

IMPLEMENTATION_COVERAGE.md

@@ -3528,20 +3528,20 @@
 
 ## glacier
 <details>
-<summary>24% implemented</summary>
+<summary>30% implemented</summary>
 
 - [ ] abort_multipart_upload
 - [ ] abort_vault_lock
 - [ ] add_tags_to_vault
 - [ ] complete_multipart_upload
 - [ ] complete_vault_lock
 - [X] create_vault
-- [ ] delete_archive
+- [X] delete_archive
 - [X] delete_vault
 - [ ] delete_vault_access_policy
 - [ ] delete_vault_notifications
 - [X] describe_job
-- [ ] describe_vault
+- [X] describe_vault
 - [ ] get_data_retrieval_policy
 - [X] get_job_output
 - [ ] get_vault_access_policy

docs/docs/services/glacier.rst

@@ -20,12 +20,12 @@ glacier
 - [ ] complete_multipart_upload
 - [ ] complete_vault_lock
 - [X] create_vault
-- [ ] delete_archive
+- [X] delete_archive
 - [X] delete_vault
 - [ ] delete_vault_access_policy
 - [ ] delete_vault_notifications
 - [X] describe_job
-- [ ] describe_vault
+- [X] describe_vault
 - [ ] get_data_retrieval_policy
 - [X] get_job_output
 - [ ] get_vault_access_policy

moto/glacier/models.py

@@ -190,7 +190,7 @@ def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
         self.vaults: Dict[str, Vault] = {}
 
-    def get_vault(self, vault_name: str) -> Vault:
+    def describe_vault(self, vault_name: str) -> Vault:
         if vault_name not in self.vaults:
             raise JsonRESTError(error_type="VaultNotFound", message="Vault not found")
         return self.vaults[vault_name]
@@ -207,31 +207,35 @@ def delete_vault(self, vault_name: str) -> None:
     def initiate_job(
         self, vault_name: str, job_type: str, tier: str, archive_id: Optional[str]
     ) -> str:
-        vault = self.get_vault(vault_name)
+        vault = self.describe_vault(vault_name)
         job_id = vault.initiate_job(job_type, tier, archive_id)
         return job_id
 
     def describe_job(self, vault_name: str, archive_id: str) -> Optional[Job]:
-        vault = self.get_vault(vault_name)
+        vault = self.describe_vault(vault_name)
         return vault.describe_job(archive_id)
 
     def list_jobs(self, vault_name: str) -> List[Job]:
-        vault = self.get_vault(vault_name)
+        vault = self.describe_vault(vault_name)
         return vault.list_jobs()
 
     def get_job_output(
         self, vault_name: str, job_id: str
     ) -> Union[str, Dict[str, Any], None]:
-        vault = self.get_vault(vault_name)
+        vault = self.describe_vault(vault_name)
         if vault.job_ready(job_id):
             return vault.get_job_output(job_id)
         else:
             return None
 
+    def delete_archive(self, vault_name: str, archive_id: str) -> None:
+        vault = self.describe_vault(vault_name)
+        vault.delete_archive(archive_id)
+
     def upload_archive(
         self, vault_name: str, body: bytes, description: str
     ) -> Dict[str, Any]:
-        vault = self.get_vault(vault_name)
+        vault = self.describe_vault(vault_name)
         return vault.create_archive(body, description)
 
 

moto/glacier/responses.py

@@ -1,5 +1,5 @@
 import json
-from typing import Any, Dict
+from typing import Any
 
 from moto.core.common_types import TYPE_RESPONSE
 from moto.core.responses import BaseResponse
@@ -12,183 +12,100 @@ class GlacierResponse(BaseResponse):
     def __init__(self) -> None:
         super().__init__(service_name="glacier")
 
+    def setup_class(
+        self, request: Any, full_url: str, headers: Any, use_raw_body: bool = False
+    ) -> None:
+        super().setup_class(request, full_url, headers, use_raw_body=True)
+
     @property
     def glacier_backend(self) -> GlacierBackend:
         return glacier_backends[self.current_account][self.region]
 
-    def all_vault_response(
-        self, request: Any, full_url: str, headers: Any
-    ) -> TYPE_RESPONSE:
-        self.setup_class(request, full_url, headers)
-        return self._all_vault_response(headers)
-
-    def _all_vault_response(self, headers: Any) -> TYPE_RESPONSE:
+    def list_vaults(self) -> TYPE_RESPONSE:
         vaults = self.glacier_backend.list_vaults()
         response = json.dumps(
             {"Marker": None, "VaultList": [vault.to_dict() for vault in vaults]}
         )
 
-        headers["content-type"] = "application/json"
+        headers = {"content-type": "application/json"}
         return 200, headers, response
 
-    def vault_response(
-        self, request: Any, full_url: str, headers: Any
-    ) -> TYPE_RESPONSE:
-        self.setup_class(request, full_url, headers)
-        return self._vault_response(request, full_url, headers)
-
-    def _vault_response(  # type: ignore[return]
-        self, request: Any, full_url: str, headers: Any
-    ) -> TYPE_RESPONSE:
-        method = request.method
-        vault_name = vault_from_glacier_url(full_url)
-
-        if method == "GET":
-            return self._vault_response_get(vault_name, headers)
-        elif method == "PUT":
-            return self._vault_response_put(vault_name, headers)
-        elif method == "DELETE":
-            return self._vault_response_delete(vault_name, headers)
-
-    def _vault_response_get(self, vault_name: str, headers: Any) -> TYPE_RESPONSE:
-        vault = self.glacier_backend.get_vault(vault_name)
-        headers["content-type"] = "application/json"
+    def describe_vault(self) -> TYPE_RESPONSE:
+        vault_name = vault_from_glacier_url(self.uri)
+        vault = self.glacier_backend.describe_vault(vault_name)
+        headers = {"content-type": "application/json"}
         return 200, headers, json.dumps(vault.to_dict())
 
-    def _vault_response_put(self, vault_name: str, headers: Any) -> TYPE_RESPONSE:
+    def create_vault(self) -> TYPE_RESPONSE:
+        vault_name = vault_from_glacier_url(self.uri)
         self.glacier_backend.create_vault(vault_name)
-        return 201, headers, ""
+        return 201, {"status": 201}, ""
 
-    def _vault_response_delete(self, vault_name: str, headers: Any) -> TYPE_RESPONSE:
+    def delete_vault(self) -> TYPE_RESPONSE:
+        vault_name = vault_from_glacier_url(self.uri)
         self.glacier_backend.delete_vault(vault_name)
-        return 204, headers, ""
-
-    def vault_archive_response(
-        self, request: Any, full_url: str, headers: Any
-    ) -> TYPE_RESPONSE:
-        self.setup_class(request, full_url, headers, use_raw_body=True)
-        return self._vault_archive_response(request, full_url, headers)
-
-    def _vault_archive_response(
-        self, request: Any, full_url: str, headers: Any
-    ) -> TYPE_RESPONSE:
-        method = request.method
-        if hasattr(request, "body"):
-            body = request.body
-        else:
-            body = request.data
-        description = ""
-        if "x-amz-archive-description" in request.headers:
-            description = request.headers["x-amz-archive-description"]
-        vault_name = full_url.split("/")[-2]
-
-        if method == "POST":
-            return self._vault_archive_response_post(
-                vault_name, body, description, headers
-            )
-        else:
-            return 400, headers, "400 Bad Request"
-
-    def _vault_archive_response_post(
-        self, vault_name: str, body: bytes, description: str, headers: Dict[str, Any]
-    ) -> TYPE_RESPONSE:
-        vault = self.glacier_backend.upload_archive(vault_name, body, description)
-        headers["x-amz-archive-id"] = vault["archive_id"]
-        headers["x-amz-sha256-tree-hash"] = vault["sha256"]
+        return 204, {"status": 204}, ""
+
+    def upload_archive(self) -> TYPE_RESPONSE:
+        description = self.headers.get("x-amz-archive-description") or ""
+        vault_name = self.parsed_url.path.split("/")[-2]
+        vault = self.glacier_backend.upload_archive(vault_name, self.body, description)
+        headers = {
+            "x-amz-archive-id": vault["archive_id"],
+            "x-amz-sha256-tree-hash": vault["sha256"],
+            "status": 201,
+        }
         return 201, headers, ""
 
-    def vault_archive_individual_response(
-        self, request: Any, full_url: str, headers: Any
-    ) -> TYPE_RESPONSE:
-        self.setup_class(request, full_url, headers)
-        return self._vault_archive_individual_response(request, full_url, headers)
-
-    def _vault_archive_individual_response(  # type: ignore[return]
-        self, request: Any, full_url: str, headers: Any
-    ) -> TYPE_RESPONSE:
-        method = request.method
-        vault_name = full_url.split("/")[-3]
-        archive_id = full_url.split("/")[-1]
-
-        if method == "DELETE":
-            vault = self.glacier_backend.get_vault(vault_name)
-            vault.delete_archive(archive_id)
-            return 204, headers, ""
-
-    def vault_jobs_response(
-        self, request: Any, full_url: str, headers: Any
-    ) -> TYPE_RESPONSE:
-        self.setup_class(request, full_url, headers)
-        return self._vault_jobs_response(request, full_url, headers)
-
-    def _vault_jobs_response(  # type: ignore[return]
-        self, request: Any, full_url: str, headers: Any
-    ) -> TYPE_RESPONSE:
-        method = request.method
-        if hasattr(request, "body"):
-            body = request.body
-        else:
-            body = request.data
-        account_id = full_url.split("/")[1]
-        vault_name = full_url.split("/")[-2]
-
-        if method == "GET":
-            jobs = self.glacier_backend.list_jobs(vault_name)
-            headers["content-type"] = "application/json"
-            return (
-                200,
-                headers,
-                json.dumps(
-                    {"JobList": [job.to_dict() for job in jobs], "Marker": None}
-                ),
-            )
-        elif method == "POST":
-            json_body = json.loads(body.decode("utf-8"))
-            job_type = json_body["Type"]
-            archive_id = None
-            if "ArchiveId" in json_body:
-                archive_id = json_body["ArchiveId"]
-            if "Tier" in json_body:
-                tier = json_body["Tier"]
-            else:
-                tier = "Standard"
-            job_id = self.glacier_backend.initiate_job(
-                vault_name, job_type, tier, archive_id
-            )
-            headers["x-amz-job-id"] = job_id
-            headers["Location"] = f"/{account_id}/vaults/{vault_name}/jobs/{job_id}"
-            return 202, headers, ""
-
-    def vault_jobs_individual_response(
-        self, request: Any, full_url: str, headers: Any
-    ) -> TYPE_RESPONSE:
-        self.setup_class(request, full_url, headers)
-        return self._vault_jobs_individual_response(full_url, headers)
-
-    def _vault_jobs_individual_response(
-        self, full_url: str, headers: Any
-    ) -> TYPE_RESPONSE:
-        vault_name = full_url.split("/")[-3]
-        archive_id = full_url.split("/")[-1]
+    def delete_archive(self) -> TYPE_RESPONSE:
+        vault_name = self.parsed_url.path.split("/")[-3]
+        archive_id = self.parsed_url.path.split("/")[-1]
 
-        job = self.glacier_backend.describe_job(vault_name, archive_id)
-        return 200, headers, json.dumps(job.to_dict())  # type: ignore
+        self.glacier_backend.delete_archive(vault_name, archive_id)
+        return 204, {"status": 204}, ""
 
-    def vault_jobs_output_response(
-        self, request: Any, full_url: str, headers: Any
-    ) -> TYPE_RESPONSE:
-        self.setup_class(request, full_url, headers)
-        return self._vault_jobs_output_response(full_url, headers)
+    def list_jobs(self) -> TYPE_RESPONSE:
+        vault_name = self.parsed_url.path.split("/")[-2]
+        jobs = self.glacier_backend.list_jobs(vault_name)
+        headers = {"content-type": "application/json"}
+        response = json.dumps(
+            {"JobList": [job.to_dict() for job in jobs], "Marker": None}
+        )
+        return 200, headers, response
+
+    def initiate_job(self) -> TYPE_RESPONSE:
+        account_id = self.uri.split("/")[1]
+        vault_name = self.parsed_url.path.split("/")[-2]
+        json_body = json.loads(self.body.decode("utf-8"))
+        job_type = json_body["Type"]
+        archive_id = json_body.get("ArchiveId")
+        tier = json_body.get("Tier") or "Standard"
+        job_id = self.glacier_backend.initiate_job(
+            vault_name, job_type, tier, archive_id
+        )
+        headers = {
+            "x-amz-job-id": job_id,
+            "Location": f"/{account_id}/vaults/{vault_name}/jobs/{job_id}",
+            "status": 202,
+        }
+        return 202, headers, ""
+
+    def describe_job(self) -> str:
+        vault_name = self.parsed_url.path.split("/")[-3]
+        archive_id = self.parsed_url.path.split("/")[-1]
+
+        job = self.glacier_backend.describe_job(vault_name, archive_id)
+        return json.dumps(job.to_dict())  # type: ignore
 
-    def _vault_jobs_output_response(self, full_url: str, headers: Any) -> TYPE_RESPONSE:
-        vault_name = full_url.split("/")[-4]
-        job_id = full_url.split("/")[-2]
+    def get_job_output(self) -> TYPE_RESPONSE:
+        vault_name = self.parsed_url.path.split("/")[-4]
+        job_id = self.parsed_url.path.split("/")[-2]
         output = self.glacier_backend.get_job_output(vault_name, job_id)
         if output is None:
-            return 404, headers, "404 Not Found"
+            return 404, {"status": 404}, "404 Not Found"
         if isinstance(output, dict):
-            headers["content-type"] = "application/json"
+            headers = {"content-type": "application/json"}
             return 200, headers, json.dumps(output)
         else:
-            headers["content-type"] = "application/octet-stream"
+            headers = {"content-type": "application/octet-stream"}
             return 200, headers, output

moto/glacier/urls.py

@@ -3,25 +3,11 @@
 url_bases = [r"https?://glacier\.(.+)\.amazonaws.com"]
 
 url_paths = {
-    "{0}/(?P<account_number>.+)/vaults$": GlacierResponse.method_dispatch(
-        GlacierResponse.all_vault_response
-    ),
-    "{0}/(?P<account_number>.+)/vaults/(?P<vault_name>[^/]+)$": GlacierResponse.method_dispatch(
-        GlacierResponse.vault_response
-    ),
-    "{0}/(?P<account_number>.+)/vaults/(?P<vault_name>.+)/archives$": GlacierResponse.method_dispatch(
-        GlacierResponse.vault_archive_response
-    ),
-    "{0}/(?P<account_number>.+)/vaults/(?P<vault_name>.+)/archives/(?P<archive_id>.+)$": GlacierResponse.method_dispatch(
-        GlacierResponse.vault_archive_individual_response
-    ),
-    "{0}/(?P<account_number>.+)/vaults/(?P<vault_name>.+)/jobs$": GlacierResponse.method_dispatch(
-        GlacierResponse.vault_jobs_response
-    ),
-    "{0}/(?P<account_number>.+)/vaults/(?P<vault_name>.+)/jobs/(?P<job_id>[^/.]+)$": GlacierResponse.method_dispatch(
-        GlacierResponse.vault_jobs_individual_response
-    ),
-    "{0}/(?P<account_number>.+)/vaults/(?P<vault_name>.+)/jobs/(?P<job_id>.+)/output$": GlacierResponse.method_dispatch(
-        GlacierResponse.vault_jobs_output_response
-    ),
+    "{0}/(?P<account_number>.+)/vaults$": GlacierResponse.dispatch,
+    "{0}/(?P<account_number>.+)/vaults/(?P<vault_name>[^/]+)$": GlacierResponse.dispatch,
+    "{0}/(?P<account_number>.+)/vaults/(?P<vault_name>.+)/archives$": GlacierResponse.dispatch,
+    "{0}/(?P<account_number>.+)/vaults/(?P<vault_name>.+)/archives/(?P<archive_id>.+)$": GlacierResponse.dispatch,
+    "{0}/(?P<account_number>.+)/vaults/(?P<vault_name>.+)/jobs$": GlacierResponse.dispatch,
+    "{0}/(?P<account_number>.+)/vaults/(?P<vault_name>.+)/jobs/(?P<job_id>[^/.]+)$": GlacierResponse.dispatch,
+    "{0}/(?P<account_number>.+)/vaults/(?P<vault_name>.+)/jobs/(?P<job_id>.+)/output$": GlacierResponse.dispatch,
 }