[Git][debian-gis-team/python-s3fs][upstream] New upstream version 2026.9.0
Antonio Valentino (@antonio.valentino)
gitlab at salsa.debian.org
Wed Sep 23 07:21:11 BST 2026
Antonio Valentino pushed to branch upstream at Debian GIS Project / python-s3fs
Commits:
181bd815 by Antonio Valentino at 2026-09-23T06:17:22+00:00
New upstream version 2026.9.0
- - - - -
6 changed files:
- docs/source/changelog.rst
- requirements.txt
- s3fs/_version.py
- s3fs/core.py
- s3fs/tests/test_custom_error_handler.py
- s3fs/tests/test_s3fs.py
Changes:
=====================================
docs/source/changelog.rst
=====================================
@@ -1,6 +1,15 @@
Changelog
=========
+2026.9.0
+--------
+
+- retry transient errors on each page of a listing instead of failing or truncating it (#982, 1042)
+- Preserve the original coroutine-related error when its traceback has no response (#1011)
+- version-guard the big exclusive-write xfails (#1040)
+· Fix requester pays issues (#1039)
+
+
2026.7.0
--------
=====================================
requirements.txt
=====================================
@@ -1,3 +1,3 @@
aiobotocore>=2.19.0,<4.0.0
-fsspec>=2026.7.0,<2026.7.1
+fsspec>=2026.9.0,<2026.9.1
aiohttp!=4.0.0a0, !=4.0.0a1, >=3.9.0
=====================================
s3fs/_version.py
=====================================
@@ -25,9 +25,9 @@ def get_keywords() -> Dict[str, str]:
# setup.py/versioneer.py will grep for the variable names, so they must
# each be defined on a line of their own. _version.py will just call
# get_keywords().
- git_refnames = " (HEAD -> main, tag: 2026.7.0)"
- git_full = "609950a67e1d2f26bd98b5053b016200272a35dd"
- git_date = "2026-07-28 13:11:33 -0400"
+ git_refnames = " (tag: 2026.9.0)"
+ git_full = "62f45e6f667e1fe745b2f613d0217b296b2de0db"
+ git_date = "2026-09-18 15:01:58 -0400"
keywords = {"refnames": git_refnames, "full": git_full, "date": git_date}
return keywords
=====================================
s3fs/core.py
=====================================
@@ -207,10 +207,11 @@ async def _error_wrapper(func, *, args=(), kwargs=None, retries):
tb = err.__traceback__
while tb.tb_next:
tb = tb.tb_next
- try:
- await tb.tb_frame.f_locals["response"]
- except Exception as e:
- err = e
+ if "response" in tb.tb_frame.f_locals:
+ try:
+ await tb.tb_frame.f_locals["response"]
+ except Exception as e:
+ err = e
err = translate_boto_error(err)
raise err
@@ -420,6 +421,7 @@ class S3FileSystem(AsyncFileSystem):
self.version_aware = version_aware
self.client_kwargs = client_kwargs or {}
self.config_kwargs = config_kwargs or {}
+ self.requester_pays = requester_pays
self.req_kw = {"RequestPayer": "requester"} if requester_pays else {}
self.s3_additional_kwargs = s3_additional_kwargs or {}
self.use_ssl = use_ssl
@@ -469,10 +471,18 @@ class S3FileSystem(AsyncFileSystem):
def _get_s3_method_kwargs(self, method, *akwarglist, **kwargs):
additional_kwargs = self.s3_additional_kwargs.copy()
+ # Apply request-wide defaults; unsupported parameters are filtered below.
+ additional_kwargs.update(self.req_kw)
for akwargs in akwarglist:
additional_kwargs.update(akwargs)
# Add the normal kwargs in
additional_kwargs.update(kwargs)
+ # A narrower scope can explicitly disable the request-wide default.
+ if (
+ "RequestPayer" in additional_kwargs
+ and additional_kwargs["RequestPayer"] is None
+ ):
+ additional_kwargs.pop("RequestPayer")
# filter all kwargs
return self._filter_kwargs(method, additional_kwargs)
@@ -808,7 +818,7 @@ class S3FileSystem(AsyncFileSystem):
if fill_cache is None:
fill_cache = self.default_fill_cache
if requester_pays is None:
- requester_pays = bool(self.req_kw)
+ requester_pays = self.requester_pays
acl = (
acl
@@ -896,26 +906,17 @@ class S3FileSystem(AsyncFileSystem):
raise ValueError(
"versions cannot be specified if the filesystem is not version aware"
)
- await self.set_session()
- s3 = await self.get_s3(bucket)
if self.version_aware:
method = "list_object_versions"
contents_key = "Versions"
else:
method = "list_objects_v2"
contents_key = "Contents"
- pag = s3.get_paginator(method)
- config = {}
+ kwargs = dict(Bucket=bucket, Prefix=prefix, Delimiter=delimiter, **self.req_kw)
if max_items is not None:
- config.update(MaxItems=max_items, PageSize=2 * max_items)
- it = pag.paginate(
- Bucket=bucket,
- Prefix=prefix,
- Delimiter=delimiter,
- PaginationConfig=config,
- **self.req_kw,
- )
- async for i in it:
+ kwargs["MaxKeys"] = 2 * max_items
+ remaining = max_items
+ async for i in self._list_pages(method, **kwargs):
for l in i.get("CommonPrefixes", []):
c = {
"Key": l["Prefix"][:-1],
@@ -925,12 +926,33 @@ class S3FileSystem(AsyncFileSystem):
}
self._fill_info(c, bucket, versions=False)
yield c
- for c in i.get(contents_key, []):
+ contents = i.get(contents_key, [])
+ if remaining is not None:
+ contents = contents[:remaining]
+ remaining -= len(contents)
+ for c in contents:
if not self.version_aware or c.get("IsLatest") or versions:
c["type"] = "file"
c["size"] = c["Size"]
self._fill_info(c, bucket, versions=versions)
yield c
+ if remaining is not None and remaining <= 0:
+ return
+
+ async def _list_pages(self, method, **kwargs):
+ # Each page goes through _call_s3, so a transient error on one page is
+ # retried like any other request instead of failing (or silently
+ # truncating) the whole listing, as the paginator did (#982).
+ while True:
+ out = await self._call_s3(method, **kwargs)
+ yield out
+ if not out.get("IsTruncated"):
+ return
+ if method == "list_object_versions":
+ kwargs["KeyMarker"] = out.get("NextKeyMarker", "")
+ kwargs["VersionIdMarker"] = out.get("NextVersionIdMarker", "")
+ else:
+ kwargs["ContinuationToken"] = out["NextContinuationToken"]
@staticmethod
def _fill_info(f, bucket, versions=False):
@@ -1258,16 +1280,16 @@ class S3FileSystem(AsyncFileSystem):
except FileNotFoundError:
# might still be a bucket we can access but don't own
pass
- try:
- await self._call_s3("head_bucket", Bucket=bucket, **self.req_kw)
- return True
- except Exception:
- pass
- try:
- await self._call_s3("get_bucket_location", Bucket=bucket, **self.req_kw)
- return True
- except Exception:
- return False
+ probes = [("head_bucket", {}), ("get_bucket_location", {})]
+ if self.requester_pays:
+ probes.append(("list_objects_v2", {"MaxKeys": 1}))
+ for method, extra in probes:
+ try:
+ await self._call_s3(method, Bucket=bucket, **extra)
+ return True
+ except Exception as e:
+ logger.debug("bucket probe %s failed for %s: %s", method, bucket, e)
+ return False
exists = sync_wrapper(_exists)
@@ -1311,7 +1333,6 @@ class S3FileSystem(AsyncFileSystem):
Key=key,
**version_id_kw(version_id or vers),
**head,
- **self.req_kw,
)
try:
return await resp["Body"].read()
@@ -1743,7 +1764,6 @@ class S3FileSystem(AsyncFileSystem):
Bucket=bucket,
Key=key,
**version_id_kw(version_id),
- **self.req_kw,
)
return {
"ETag": out.get("ETag", ""),
@@ -1761,7 +1781,7 @@ class S3FileSystem(AsyncFileSystem):
raise translate_boto_error(e, set_cause=False)
else:
try:
- out = await self._call_s3("head_bucket", Bucket=bucket, **self.req_kw)
+ out = await self._call_s3("head_bucket", Bucket=bucket)
return {
"name": bucket,
"type": "directory",
@@ -1782,7 +1802,6 @@ class S3FileSystem(AsyncFileSystem):
Prefix=key.rstrip("/") + "/" if key else "",
Delimiter="/",
MaxKeys=1,
- **self.req_kw,
)
if (
out.get("KeyCount", 0) > 0
@@ -1880,7 +1899,6 @@ class S3FileSystem(AsyncFileSystem):
kwargs,
Bucket=bucket,
Prefix=key,
- **self.req_kw,
)
versions.extend(out["Versions"])
kwargs.update(
@@ -1912,7 +1930,6 @@ class S3FileSystem(AsyncFileSystem):
Bucket=bucket,
Key=key,
**version_id_kw(version_id),
- **self.req_kw,
)
meta = {k.replace("_", "-"): v for k, v in response["Metadata"].items()}
return meta
@@ -2410,10 +2427,9 @@ class S3FileSystem(AsyncFileSystem):
async def _rm_versioned_bucket_contents(self, bucket):
"""Remove a versioned bucket and all contents"""
- await self.set_session()
- s3 = await self.get_s3(bucket)
- pag = s3.get_paginator("list_object_versions")
- async for plist in pag.paginate(Bucket=bucket):
+ async for plist in self._list_pages(
+ "list_object_versions", Bucket=bucket, **self.req_kw
+ ):
obs = plist.get("Versions", []) + plist.get("DeleteMarkers", [])
delete_keys = {
"Objects": [
@@ -2546,6 +2562,8 @@ class S3File(AbstractBufferedFile):
self.fill_cache = fill_cache
self.s3_additional_kwargs = s3_additional_kwargs or {}
self.req_kw = {"RequestPayer": "requester"} if requester_pays else {}
+ self._request_payer_kw = {"RequestPayer": self.req_kw.get("RequestPayer")}
+ self.read_kw = self._request_payer_kw
if "r" not in mode:
if block_size < 5 * 2**20:
raise ValueError("Block size must be >=5MB")
@@ -2589,7 +2607,6 @@ class S3File(AbstractBufferedFile):
Bucket=bucket,
Key=key,
**version_id_kw(version_id),
- **self.req_kw,
)
head = {
@@ -2615,10 +2632,19 @@ class S3File(AbstractBufferedFile):
and "ETag" in self.details
and not s3.local_expiry_check
):
- self.req_kw["IfMatch"] = self.details["ETag"]
+ self.read_kw = {
+ **self._request_payer_kw,
+ "IfMatch": self.details["ETag"],
+ }
def _call_s3(self, method, *kwarglist, **kwargs):
- return self.fs.call_s3(method, self.s3_additional_kwargs, *kwarglist, **kwargs)
+ return self.fs.call_s3(
+ method,
+ self.s3_additional_kwargs,
+ self._request_payer_kw,
+ *kwarglist,
+ **kwargs,
+ )
def _initiate_upload(self):
if self.autocommit and not self.append_block and self.tell() < self.blocksize:
@@ -2694,7 +2720,7 @@ class S3File(AbstractBufferedFile):
self.version_id,
start,
end,
- req_kw=self.req_kw,
+ req_kw=self.read_kw,
details=self.details,
)
=====================================
s3fs/tests/test_custom_error_handler.py
=====================================
@@ -26,6 +26,32 @@ class CustomNonRetryableError(Exception):
pass
+def test_coroutine_error_without_response_preserves_original():
+ error = TypeError("'coroutine' object is not subscriptable")
+
+ async def failing_func():
+ raise error
+
+ with pytest.raises(TypeError) as exc:
+ asyncio.run(_error_wrapper(failing_func, retries=1))
+ assert exc.value is error
+
+
+def test_coroutine_error_recovers_response_exception():
+ async def response_error():
+ raise ClientError(
+ {"Error": {"Code": "AccessDenied", "Message": "Access denied"}},
+ "PutObject",
+ )
+
+ async def failing_func():
+ response = response_error()
+ return response["Body"]
+
+ with pytest.raises(PermissionError, match="Access denied"):
+ asyncio.run(_error_wrapper(failing_func, retries=1))
+
+
@pytest.fixture(autouse=True)
def reset_error_handler():
"""Reset the custom error handler and retryable errors after each test."""
=====================================
s3fs/tests/test_s3fs.py
=====================================
@@ -7,6 +7,7 @@ from concurrent.futures import ProcessPoolExecutor
import io
import os
import random
+import socket
import requests
import time
import sys
@@ -1149,6 +1150,25 @@ def test_errors_cause_preservings(monkeypatch, s3):
assert exc.value.__cause__ is None
+def test_listing_retries_transient_errors(monkeypatch, s3):
+ # a transient error on a listing page is retried like any other call (#982)
+ expected = s3.find(test_bucket_name)
+ original = type(s3.s3).list_objects_v2
+ calls = []
+
+ async def list_objects_v2(self, *args, **kwargs):
+ calls.append(kwargs)
+ if len(calls) == 1:
+ raise socket.timeout
+ return await original(self, *args, **kwargs)
+
+ monkeypatch.setattr(type(s3.s3), "list_objects_v2", list_objects_v2)
+ s3.invalidate_cache()
+
+ assert s3.find(test_bucket_name) == expected
+ assert len(calls) == 2
+
+
def test_local_expiry_check(s3):
s3 = S3FileSystem(
local_expiry_check=True,
@@ -2074,11 +2094,75 @@ def test_requester_pays(s3):
fn = test_bucket_name + "/myfile"
s3 = S3FileSystem(requester_pays=True, client_kwargs={"endpoint_url": endpoint_uri})
assert s3.req_kw["RequestPayer"] == "requester"
+ put_kwargs = s3._get_s3_method_kwargs(
+ s3.s3.put_object, Bucket=test_bucket_name, Key="myfile"
+ )
+ assert put_kwargs["RequestPayer"] == "requester"
+ list_kwargs = s3._get_s3_method_kwargs(
+ s3.s3.list_objects_v2, Bucket=test_bucket_name, MaxKeys=1
+ )
+ assert list_kwargs["RequestPayer"] == "requester"
+ assert "RequestPayer" not in s3._get_s3_method_kwargs(s3.s3.list_buckets)
s3.touch(fn)
with s3.open(fn, "rb") as f:
assert f.req_kw["RequestPayer"] == "requester"
+ at pytest.mark.parametrize(
+ "filesystem_requester_pays,file_requester_pays,expected",
+ [(False, True, True), (True, False, False)],
+)
+ at pytest.mark.parametrize("method", ["get_object", "put_object"])
+def test_requester_pays_per_file_override(
+ filesystem_requester_pays,
+ file_requester_pays,
+ expected,
+ method,
+):
+ fn = test_bucket_name + "/myfile"
+ s3 = S3FileSystem(
+ requester_pays=filesystem_requester_pays,
+ client_kwargs={"endpoint_url": endpoint_uri},
+ skip_instance_cache=True,
+ )
+ f = s3fs.core.S3File(s3, fn, mode="rb", requester_pays=file_requester_pays, size=0)
+ request_kwargs = s3._get_s3_method_kwargs(
+ getattr(s3.s3, method),
+ f.s3_additional_kwargs,
+ f._request_payer_kw,
+ Bucket=test_bucket_name,
+ Key="myfile",
+ )
+ f.close()
+
+ assert ("RequestPayer" in request_kwargs) is expected
+ if expected:
+ assert request_kwargs["RequestPayer"] == "requester"
+
+
+def test_requester_pays_makedirs_uses_object_listing():
+ class MetadataRestrictedS3FileSystem(S3FileSystem):
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.calls = []
+
+ async def _call_s3(self, method, *akwarglist, **kwargs):
+ self.calls.append(method)
+ if method == "list_objects_v2":
+ return {"Contents": []}
+ raise PermissionError("Bucket metadata is not accessible")
+
+ s3 = MetadataRestrictedS3FileSystem(
+ requester_pays=True,
+ client_kwargs={"endpoint_url": endpoint_uri},
+ skip_instance_cache=True,
+ )
+ s3.makedirs("requester-pays-bucket/checkpoints/run", exist_ok=True)
+
+ assert s3.calls == ["head_bucket", "get_bucket_location", "list_objects_v2"]
+ assert "create_bucket" not in s3.calls
+
+
def test_credentials():
s3 = S3FileSystem(
key="foo", secret="foo", client_kwargs={"endpoint_url": endpoint_uri}
@@ -2991,10 +3075,13 @@ def test_exist_after_delete(s3):
# condition: True if running on botocore < 1.36.0
# The below tests for exclusive writes will fail on older versions of botocore.
old_botocore = version.parse(botocore.__version__) < version.parse("1.36.0")
+old_moto = version.parse(moto.__version__) < version.parse("5.1.11")
@pytest.mark.xfail(
- reason="moto doesn't support IfNoneMatch for MPU when object created via MPU"
+ old_moto,
+ reason="moto<5.1.11 doesn't support IfNoneMatch for MPU when object created via MPU",
+ strict=True,
)
def test_pipe_exclusive_big(s3):
chunksize = 5 * 2**20 # minimum allowed
@@ -3034,7 +3121,9 @@ def test_pipe_exclusive_big_after_small(s3):
@pytest.mark.xfail(
- reason="moto doesn't support IfNoneMatch for MPU when object created via MPU"
+ old_moto,
+ reason="moto<5.1.11 doesn't support IfNoneMatch for MPU when object created via MPU",
+ strict=True,
)
def test_put_exclusive_big(s3, tmpdir):
chunksize = 5 * 2**20 # minimum allowed
View it on GitLab: https://salsa.debian.org/debian-gis-team/python-s3fs/-/commit/181bd815568a8b2f4b563456d8585cef4b34560c
--
View it on GitLab: https://salsa.debian.org/debian-gis-team/python-s3fs/-/commit/181bd815568a8b2f4b563456d8585cef4b34560c
You're receiving this email because of your account on salsa.debian.org. Manage all notifications: https://salsa.debian.org/-/profile/notifications | Help: https://salsa.debian.org/help
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://alioth-lists.debian.net/pipermail/pkg-grass-devel/attachments/20260923/b368d733/attachment-0001.htm>
More information about the Pkg-grass-devel
mailing list