fix(storage): close the upload file handle when the request fails

src/storage/src/storage3/_async/file_api.py

@@ -65,33 +65,38 @@ async def _request(
         **kwargs: Any,
     ) -> Response:
         try:
-            url_path = self._base_url.joinpath(*path).with_query(query_params)
-            headers = headers or dict()
-            headers.update(self._headers)
-            response = await self._client.request(
-                method,
-                str(url_path),
-                headers=headers,
-                json=json,
-                files=files,
-                **kwargs,
-            )
-            response.raise_for_status()
-        except HTTPStatusError as exc:
             try:
-                resp = exc.response.json()
-                raise StorageApiError(
-                    resp["message"], resp["error"], resp["statusCode"]
-                ) from exc
-            except KeyError as err:
-                message = f"Unable to parse error message: {resp.text}"
-                raise StorageApiError(message, "InternalError", 400) from err
-
-        # close the resource before returning the response
-        if files and "file" in files and isinstance(files["file"][1], BufferedReader):
-            files["file"][1].close()
-
-        return response
+                url_path = self._base_url.joinpath(*path).with_query(query_params)
+                headers = headers or dict()
+                headers.update(self._headers)
+                response = await self._client.request(
+                    method,
+                    str(url_path),
+                    headers=headers,
+                    json=json,
+                    files=files,
+                    **kwargs,
+                )
+                response.raise_for_status()
+            except HTTPStatusError as exc:
+                try:
+                    resp = exc.response.json()
+                    raise StorageApiError(
+                        resp["message"], resp["error"], resp["statusCode"]
+                    ) from exc
+                except KeyError as err:
+                    message = f"Unable to parse error message: {resp.text}"
+                    raise StorageApiError(message, "InternalError", 400) from err
+
+            return response
+        finally:
+            # close the resource on the error path too, not just on success
+            if (
+                files
+                and "file" in files
+                and isinstance(files["file"][1], BufferedReader)
+            ):
+                files["file"][1].close()
 
     async def create_signed_upload_url(
         self,

src/storage/src/storage3/_sync/file_api.py

@@ -65,33 +65,38 @@ def _request(
         **kwargs: Any,
     ) -> Response:
         try:
-            url_path = self._base_url.joinpath(*path).with_query(query_params)
-            headers = headers or dict()
-            headers.update(self._headers)
-            response = self._client.request(
-                method,
-                str(url_path),
-                headers=headers,
-                json=json,
-                files=files,
-                **kwargs,
-            )
-            response.raise_for_status()
-        except HTTPStatusError as exc:
             try:
-                resp = exc.response.json()
-                raise StorageApiError(
-                    resp["message"], resp["error"], resp["statusCode"]
-                ) from exc
-            except KeyError as err:
-                message = f"Unable to parse error message: {resp.text}"
-                raise StorageApiError(message, "InternalError", 400) from err
-
-        # close the resource before returning the response
-        if files and "file" in files and isinstance(files["file"][1], BufferedReader):
-            files["file"][1].close()
-
-        return response
+                url_path = self._base_url.joinpath(*path).with_query(query_params)
+                headers = headers or dict()
+                headers.update(self._headers)
+                response = self._client.request(
+                    method,
+                    str(url_path),
+                    headers=headers,
+                    json=json,
+                    files=files,
+                    **kwargs,
+                )
+                response.raise_for_status()
+            except HTTPStatusError as exc:
+                try:
+                    resp = exc.response.json()
+                    raise StorageApiError(
+                        resp["message"], resp["error"], resp["statusCode"]
+                    ) from exc
+                except KeyError as err:
+                    message = f"Unable to parse error message: {resp.text}"
+                    raise StorageApiError(message, "InternalError", 400) from err
+
+            return response
+        finally:
+            # close the resource on the error path too, not just on success
+            if (
+                files
+                and "file" in files
+                and isinstance(files["file"][1], BufferedReader)
+            ):
+                files["file"][1].close()
 
     def create_signed_upload_url(
         self,

src/storage/tests/_async/test_client.py

@@ -8,10 +8,12 @@
 
 import pytest
 from httpx import AsyncClient as HttpxClient
+from httpx import Headers
 from httpx import HTTPStatusError, Response
 from storage3 import AsyncStorageClient
 from storage3.exceptions import StorageApiError
 from storage3.utils import StorageException
+from yarl import URL
 
 from .. import AsyncBucketProxy
 from ..utils import AsyncFinalizerFactory
@@ -827,3 +829,33 @@ async def test_client_list_v2_paginated(
         assert all(f.name.startswith(file.bucket_path) for f in result.objects)
         pages += 1
     assert pages == 4
+
+
+async def test_upload_closes_file_handle_on_error(tmp_path: Path) -> None:
+    """A failed upload must not leak the handle storage3 opened itself."""
+    source = tmp_path / "leak.txt"
+    source.write_bytes(b"payload")
+
+    error_response = Mock(spec=Response)
+    error_response.json.return_value = {
+        "message": "Duplicate",
+        "error": "Duplicate",
+        "statusCode": 409,
+    }
+
+    captured: dict = {}
+
+    async def fail(*args: object, **kwargs: object) -> Response:
+        captured["files"] = kwargs.get("files")
+        raise HTTPStatusError("409", request=Mock(), response=error_response)
+
+    client = AsyncMock()
+    client.headers = Headers()
+    client.request = fail
+
+    proxy = AsyncBucketProxy("bucket", URL("http://localhost"), Headers(), client)
+
+    with pytest.raises(StorageApiError):
+        await proxy.upload("leak.txt", str(source))
+
+    assert captured["files"]["file"][1].closed

src/storage/tests/_sync/test_client.py

@@ -1,17 +1,19 @@
 from __future__ import annotations
 
-from collections.abc import Generator
+from collections.abc import Generator, Generator
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 from unittest.mock import Mock, patch
 from uuid import uuid4
 
 import pytest
 from httpx import Client as HttpxClient
+from httpx import Headers
 from httpx import HTTPStatusError, Response
 from storage3 import SyncStorageClient
 from storage3.exceptions import StorageApiError
 from storage3.utils import StorageException
+from yarl import URL
 
 from .. import SyncBucketProxy
 from ..utils import SyncFinalizerFactory
@@ -825,3 +827,33 @@ def test_client_list_v2_paginated(
         assert all(f.name.startswith(file.bucket_path) for f in result.objects)
         pages += 1
     assert pages == 4
+
+
+def test_upload_closes_file_handle_on_error(tmp_path: Path) -> None:
+    """A failed upload must not leak the handle storage3 opened itself."""
+    source = tmp_path / "leak.txt"
+    source.write_bytes(b"payload")
+
+    error_response = Mock(spec=Response)
+    error_response.json.return_value = {
+        "message": "Duplicate",
+        "error": "Duplicate",
+        "statusCode": 409,
+    }
+
+    captured: dict = {}
+
+    def fail(*args: object, **kwargs: object) -> Response:
+        captured["files"] = kwargs.get("files")
+        raise HTTPStatusError("409", request=Mock(), response=error_response)
+
+    client = Mock()
+    client.headers = Headers()
+    client.request = fail
+
+    proxy = SyncBucketProxy("bucket", URL("http://localhost"), Headers(), client)
+
+    with pytest.raises(StorageApiError):
+        proxy.upload("leak.txt", str(source))
+
+    assert captured["files"]["file"][1].closed