Translate delete-marker versions to FileNotFoundError and fix two S3Core edge cases (GH-1338)

docs/filesystem.md

@@ -243,6 +243,7 @@ operations raise natural errors instead of botocore's `ClientError`:
 | S3 error | Python exception |
 | --- | --- |
 | `404` / `NoSuchKey` / `NoSuchBucket` | `FileNotFoundError` |
+| `405` of a version that is a delete marker (`x-amz-delete-marker: true`) | `FileNotFoundError` |
 | `403` / `AccessDenied` | `PermissionError` |
 | `BucketAlreadyExists` / `BucketAlreadyOwnedByYou` | `FileExistsError` |
 | `PreconditionFailed` of an `If-None-Match` condition | `FileExistsError` |

pyathena/filesystem/s3_core.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import contextlib
+import functools
 import logging
 import math
 import traceback
@@ -620,7 +621,11 @@ def call(self, method: str | Callable[..., Any], **request) -> dict[str, Any]:
         """Send a request with the retry policy and translate its errors.
 
         ``request_kwargs`` that the operation accepts are added; a parameter
-        of the request takes precedence.
+        of the request takes precedence. The operation is identified by the
+        name of the method; a ``functools.partial`` is identified by the
+        method it wraps, and its bound keywords are parameters of the request.
+        A callable whose ``__name__`` is not that of a client method is sent
+        without them.
 
         Args:
             method: The name of the client method, or the method itself.
@@ -641,7 +646,12 @@ def call(self, method: str | Callable[..., Any], **request) -> dict[str, Any]:
         """
         func = getattr(self._client, method) if isinstance(method, str) else method
         if self._request_kwargs:
-            request = {**self.operation_params(func.__name__, self._request_kwargs), **request}
+            target, bound = func, set(request)
+            while isinstance(target, functools.partial):
+                bound.update(target.keywords)
+                target = target.func
+            params = self.operation_params(getattr(target, "__name__", ""), self._request_kwargs)
+            request.update({k: v for k, v in params.items() if k not in bound})
         try:
             response = retry_api_call(func, config=self._retry_config, logger=_logger, **request)
         except BaseException as e:
@@ -1115,8 +1125,10 @@ def part_ranges(self, size: int, block_size: int) -> list[tuple[int, int]]:
 
         Returns:
             The ``(start, end)`` byte ranges, with an exclusive end, that
-            cover the whole object in order.
+            cover the whole object in order; empty for an empty object.
         """
+        if size == 0:
+            return []
         block_size = max(block_size, math.ceil(size / self.MULTIPART_UPLOAD_MAX_PARTS))
         starts = list(range(0, size, block_size))
         if len(starts) > 1 and size - starts[-1] < self.MULTIPART_UPLOAD_MIN_PART_SIZE:

pyathena/filesystem/s3_errors.py

@@ -35,7 +35,9 @@ class S3ClientError:
     by its HTTP status code; if neither is recognized, a generic ``OSError``
     with the original error message is used. A ``PreconditionFailed`` error
     of an ``If-None-Match`` condition, from a write that must not replace an
-    existing object, is mapped to ``FileExistsError``.
+    existing object, is mapped to ``FileExistsError``. A ``405`` error with
+    the ``x-amz-delete-marker: true`` header, from a request for a version
+    that is a delete marker, is mapped to ``FileNotFoundError``.
 
     Example:
         >>> try:
@@ -108,14 +110,21 @@ def __init__(self, error: botocore.exceptions.ClientError) -> None:
         self._code: str = str(error_info.get("Code", ""))
         self._message: str = str(error_info.get("Message", error))
         self._condition: str = str(error_info.get("Condition", ""))
-        status_code = error.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
+        response_metadata = error.response.get("ResponseMetadata", {})
+        status_code = response_metadata.get("HTTPStatusCode")
         self._http_status_code: int | None = int(status_code) if status_code is not None else None
+        headers = response_metadata.get("HTTPHeaders", {})
+        self._delete_marker: bool = headers.get("x-amz-delete-marker") == "true"
         self._os_error: OSError = self._translate()
 
     def _translate(self) -> OSError:
         if self._code == "PreconditionFailed" and self._condition == "If-None-Match":
             # A conditional write (IfNoneMatch="*") found an existing object.
             return FileExistsError(self._message)
+        if self._http_status_code == 405 and self._delete_marker:
+            # HeadObject/GetObject of a version that is a delete marker.
+            # https://docs.aws.amazon.com/AmazonS3/latest/userguide/DeleteMarker.html
+            return FileNotFoundError(self._message)
         exception = self._ERROR_CODE_TO_EXCEPTION.get(self._code)
         if exception:
             return exception(self._message)

pyathena/filesystem/s3_writer.py

@@ -179,8 +179,6 @@ def iter_copy_parts(
             raise ValueError("size must be nonnegative.")
         if first_part_number < 1:
             raise ValueError("part_number must be at least 1.")
-        if size == 0:
-            return
         ranges = self._core.part_ranges(size, self._core.MULTIPART_UPLOAD_MAX_PART_SIZE)
         for part_number, range_ in enumerate(ranges, first_part_number):
             self._check_part_number(part_number)

tests/pyathena/filesystem/test_s3_core.py

@@ -6,6 +6,7 @@
 # SPDX-License-Identifier: MIT
 
 import copy
+import functools
 import gc
 import io
 import re
@@ -119,6 +120,31 @@ def test_call_translates_errors(self, status, code, error):
         assert isinstance(e.value.__cause__, botocore.exceptions.ClientError)
         assert e.value.__cause__.response["Error"]["Code"] == code
 
+    def test_call_partial(self):
+        # GH-1338: a functools.partial of a client method has no __name__; it
+        # is sent with the request_kwargs of the method it wraps, and its
+        # bound keywords take precedence as parameters of the request.
+        core, stubber = _make_core(
+            request_kwargs={"RequestPayer": "requester", "ExpectedBucketOwner": "111122223333"}
+        )
+        stubber.add_response(
+            "head_object",
+            {"ContentLength": 1},
+            expected_params={
+                "Bucket": "bucket",
+                "Key": "key",
+                "RequestPayer": "requester",
+                "ExpectedBucketOwner": "444455556666",
+            },
+        )
+        head_object = functools.partial(
+            core.client.head_object, Bucket="bucket", ExpectedBucketOwner="444455556666"
+        )
+        with stubber:
+            response = core.call(head_object, Key="key")
+        stubber.assert_no_pending_responses()
+        assert response["ContentLength"] == 1
+
     @pytest.mark.parametrize("translated", [True, False])
     def test_call_failure_frees_request(self, translated):
         # GH-1186: the frames of the retries and of botocore below call() in
@@ -1082,6 +1108,8 @@ def test_upload_part_uses_creation_algorithm_and_identity(self):
     @pytest.mark.parametrize(
         ("size", "block_size", "ranges"),
         [
+            # GH-1338: an empty object has no ranges.
+            (0, 5 * 2**20, []),
             # A single range.
             (5 * 2**20, 5 * 2**20, [(0, 5 * 2**20)]),
             # The size is an exact multiple of the block size.

tests/pyathena/filesystem/test_s3_errors.py

@@ -13,12 +13,15 @@
 from pyathena.filesystem.s3_errors import S3ClientError
 
 
-def _client_error(code, message="error message", status_code=None, condition=None):
+def _client_error(code, message="error message", status_code=None, condition=None, headers=None):
     error_response = {"Error": {"Code": code, "Message": message}}
     if condition is not None:
         error_response["Error"]["Condition"] = condition
     if status_code is not None:
         error_response["ResponseMetadata"] = {"HTTPStatusCode": status_code}
+    if headers is not None:
+        # botocore lowercases the names of the response headers.
+        error_response.setdefault("ResponseMetadata", {})["HTTPHeaders"] = headers
     return botocore.exceptions.ClientError(error_response, "TestOperation")
 
 
@@ -109,3 +112,21 @@ def test_os_error_precondition_failed(self, condition, expected):
         if expected is OSError:
             assert actual.errno == errno.EINVAL
         assert "error message" in str(actual)
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            # HEAD responses have no body, so the code is the status code.
+            "405",
+            "MethodNotAllowed",
+        ],
+    )
+    def test_os_error_delete_marker(self, code):
+        # GH-1338: HeadObject/GetObject of a version that is a delete marker
+        # is answered with 405 and the x-amz-delete-marker header, and the
+        # version does not exist as an object. Other 405 errors map to EPERM.
+        actual = S3ClientError(
+            _client_error(code, status_code=405, headers={"x-amz-delete-marker": "true"})
+        ).os_error
+        assert type(actual) is FileNotFoundError
+        assert "error message" in str(actual)