Techdebt: Extract S3 DELETE responses into separate methods (#8044)

moto/s3/models.py

@@ -2767,7 +2767,7 @@ def delete_object(
         key_name: str,
         version_id: Optional[str] = None,
         bypass: bool = False,
-    ) -> Tuple[bool, Optional[Dict[str, Any]]]:
+    ) -> Tuple[bool, Dict[str, Any]]:
         bucket = self.get_bucket(bucket_name)
 
         response_meta = {}
@@ -2831,7 +2831,7 @@ def delete_object(
 
             return True, response_meta
         except KeyError:
-            return False, None
+            return False, response_meta
 
     def delete_objects(
         self,

moto/s3/responses.py

@@ -199,6 +199,11 @@ def backend(self) -> S3Backend:
     def should_autoescape(self) -> bool:
         return True
 
+    def abort_multipart_upload(self) -> TYPE_RESPONSE:
+        upload_id = self._get_param("uploadId")
+        self.backend.abort_multipart_upload(self.bucket_name, upload_id)
+        return 204, {}, ""
+
     def all_buckets(self) -> str:
         self.data["Action"] = "ListAllMyBuckets"
         self._authenticate_and_authorize_s3_action()
@@ -1069,35 +1074,28 @@ def _bucket_response_delete(
         self._authenticate_and_authorize_s3_action(bucket_name=bucket_name)
 
         if "policy" in querystring:
-            self.backend.delete_bucket_policy(bucket_name)
-            return 204, {}, ""
+            return self.delete_bucket_policy()
         elif "tagging" in querystring:
-            self.backend.delete_bucket_tagging(bucket_name)
-            return 204, {}, ""
+            return self.delete_bucket_tagging()
         elif "website" in querystring:
-            self.backend.delete_bucket_website(bucket_name)
-            return 204, {}, ""
+            return self.delete_bucket_website()
         elif "cors" in querystring:
-            self.backend.delete_bucket_cors(bucket_name)
-            return 204, {}, ""
+            return self.delete_bucket_cors()
         elif "lifecycle" in querystring:
-            self.backend.delete_bucket_lifecycle(bucket_name)
-            return 204, {}, ""
+            return self.delete_bucket_lifecycle()
         elif "publicAccessBlock" in querystring:
-            self.backend.delete_public_access_block(bucket_name)
-            return 204, {}, ""
+            return self.delete_public_access_block()
         elif "encryption" in querystring:
-            self.backend.delete_bucket_encryption(bucket_name)
-            return 204, {}, ""
+            return self.delete_bucket_encryption()
         elif "replication" in querystring:
-            self.backend.delete_bucket_replication(bucket_name)
-            return 204, {}, ""
+            return self.delete_bucket_replication()
         elif "ownershipControls" in querystring:
-            self.backend.delete_bucket_ownership_controls(bucket_name)
-            return 204, {}, ""
+            return self.delete_bucket_ownership_controls()
 
-        removed_bucket = self.backend.delete_bucket(bucket_name)
+        return self.delete_bucket()
 
+    def delete_bucket(self) -> TYPE_RESPONSE:
+        removed_bucket = self.backend.delete_bucket(self.bucket_name)
         if removed_bucket:
             # Bucket exists
             template = self.response_template(S3_DELETE_BUCKET_SUCCESS)
@@ -1107,6 +1105,42 @@ def _bucket_response_delete(
             template = self.response_template(S3_DELETE_BUCKET_WITH_ITEMS_ERROR)
             return 409, {}, template.render(bucket=removed_bucket)
 
+    def delete_bucket_ownership_controls(self) -> TYPE_RESPONSE:
+        self.backend.delete_bucket_ownership_controls(self.bucket_name)
+        return 204, {}, ""
+
+    def delete_bucket_replication(self) -> TYPE_RESPONSE:
+        self.backend.delete_bucket_replication(self.bucket_name)
+        return 204, {}, ""
+
+    def delete_bucket_encryption(self) -> TYPE_RESPONSE:
+        self.backend.delete_bucket_encryption(self.bucket_name)
+        return 204, {}, ""
+
+    def delete_public_access_block(self) -> TYPE_RESPONSE:
+        self.backend.delete_public_access_block(self.bucket_name)
+        return 204, {}, ""
+
+    def delete_bucket_lifecycle(self) -> TYPE_RESPONSE:
+        self.backend.delete_bucket_lifecycle(self.bucket_name)
+        return 204, {}, ""
+
+    def delete_bucket_cors(self) -> TYPE_RESPONSE:
+        self.backend.delete_bucket_cors(self.bucket_name)
+        return 204, {}, ""
+
+    def delete_bucket_website(self) -> TYPE_RESPONSE:
+        self.backend.delete_bucket_website(self.bucket_name)
+        return 204, {}, ""
+
+    def delete_bucket_tagging(self) -> TYPE_RESPONSE:
+        self.backend.delete_bucket_tagging(self.bucket_name)
+        return 204, {}, ""
+
+    def delete_bucket_policy(self) -> TYPE_RESPONSE:
+        self.backend.delete_bucket_policy(self.bucket_name)
+        return 204, {}, ""
+
     def _bucket_response_post(self, request: Any, bucket_name: str) -> TYPE_RESPONSE:
         response_headers = {}
         if not request.headers.get("Content-Length"):
@@ -1401,11 +1435,11 @@ def _key_response(
         elif method == "PUT":
             return self._key_response_put(request, body, bucket_name, query, key_name)
         elif method == "HEAD":
-            return self._key_response_head(
+            return self.head_object(
                 bucket_name, query, key_name, headers=request.headers
             )
         elif method == "DELETE":
-            return self._key_response_delete(headers, bucket_name, query, key_name)
+            return self._key_response_delete(bucket_name, query, key_name)
         elif method == "POST":
             return self._key_response_post(request, body, bucket_name, query, key_name)
         elif method == "OPTIONS":
@@ -1849,7 +1883,7 @@ def _key_response_put(
         response_headers.pop("content-length", None)
         return 200, response_headers, ""
 
-    def _key_response_head(
+    def head_object(
         self,
         bucket_name: str,
         query: Dict[str, Any],
@@ -2277,34 +2311,40 @@ def _replication_config_from_xml(self, xml: str) -> Dict[str, Any]:
         return config
 
     def _key_response_delete(
-        self, headers: Any, bucket_name: str, query: Dict[str, Any], key_name: str
+        self, bucket_name: str, query: Dict[str, Any], key_name: str
     ) -> TYPE_RESPONSE:
         self._set_action("KEY", "DELETE", query)
         self._authenticate_and_authorize_s3_action(
             bucket_name=bucket_name, key_name=key_name
         )
 
         if query.get("uploadId"):
-            upload_id = query["uploadId"][0]
-            self.backend.abort_multipart_upload(bucket_name, upload_id)
-            return 204, {}, ""
-        version_id = query.get("versionId", [None])[0]
+            return self.abort_multipart_upload()
         if "tagging" in query:
-            self.backend.delete_object_tagging(
-                bucket_name, key_name, version_id=version_id
-            )
-            template = self.response_template(S3_DELETE_KEY_TAGGING_RESPONSE)
-            return 204, {}, template.render(version_id=version_id)
-        bypass = headers.get("X-Amz-Bypass-Governance-Retention")
+            return self.delete_object_tagging()
+        return self.delete_object()
+
+    def delete_object(self) -> TYPE_RESPONSE:
+        bypass = self.headers.get("X-Amz-Bypass-Governance-Retention")
+        key_name = self.parse_key_name()
+        version_id = self._get_param("versionId")
         _, response_meta = self.backend.delete_object(
-            bucket_name, key_name, version_id=version_id, bypass=bypass
+            self.bucket_name, key_name, version_id=version_id, bypass=bypass
         )
         response_headers = {}
-        if response_meta is not None:
-            for k in response_meta:
-                response_headers[f"x-amz-{k}"] = response_meta[k]
+        for k in response_meta:
+            response_headers[f"x-amz-{k}"] = response_meta[k]
         return 204, response_headers, ""
 
+    def delete_object_tagging(self) -> TYPE_RESPONSE:
+        key_name = self.parse_key_name()
+        version_id = self._get_param("versionId")
+        self.backend.delete_object_tagging(
+            self.bucket_name, key_name, version_id=version_id
+        )
+        template = self.response_template(S3_DELETE_KEY_TAGGING_RESPONSE)
+        return 204, {}, template.render(version_id=version_id)
+
     def _complete_multipart_body(self, body: bytes) -> Iterator[Tuple[int, str]]:
         ps = minidom.parseString(body).getElementsByTagName("Part")
         prev = 0