[Git][debian-gis-team/python-s3fs][upstream] New upstream version 2026.7.0

Antonio Valentino (@antonio.valentino) gitlab at salsa.debian.org
Sat Aug 1 11:13:10 BST 2026



Antonio Valentino pushed to branch upstream at Debian GIS Project / python-s3fs


Commits:
ca03e160 by Antonio Valentino at 2026-08-01T09:59:18+00:00
New upstream version 2026.7.0
- - - - -


7 changed files:

- .github/workflows/ci.yml
- ci/env.yaml
- docs/source/changelog.rst
- requirements.txt
- s3fs/_version.py
- s3fs/core.py
- s3fs/tests/test_s3fs.py


Changes:

=====================================
.github/workflows/ci.yml
=====================================
@@ -15,7 +15,7 @@ jobs:
           - "3.12"
           - "3.13"
           - "3.14"
-        aiobotocore-version: [">=2.19.0,<2.20.0", "<3.0.0", "<4.0.0"]
+        aiobotocore-version: ["<4.0.0"]
 
     env:
       BOTO_CONFIG: /dev/null
@@ -38,7 +38,7 @@ jobs:
         shell: bash -l {0}
         run: |
           pip install git+https://github.com/fsspec/filesystem_spec
-          pip install --upgrade "aiobotocore${{ matrix.aiobotocore-version }}"
+          pip install --upgrade --force-reinstall "aiobotocore${{ matrix.aiobotocore-version }}"
           pip install . --no-deps
           pip list
 


=====================================
ci/env.yaml
=====================================
@@ -14,6 +14,6 @@ dependencies:
   - black
   - httpretty
   - aiobotocore
-  - moto
+  - moto !=5.2.0
   - flask
   - fsspec


=====================================
docs/source/changelog.rst
=====================================
@@ -1,6 +1,18 @@
 Changelog
 =========
 
+2026.7.0
+--------
+
+- don't cache prefix-filtered listings as complete directory entries (#1034)
+
+2026.6.0
+--------
+
+- cat should use info (#1029)
+- loosen bounds to allow install from git (#1026)
+- Fix CI (#1024)
+
 2026.4.0
 --------
 


=====================================
requirements.txt
=====================================
@@ -1,3 +1,3 @@
 aiobotocore>=2.19.0,<4.0.0
-fsspec==2026.4.0
+fsspec>=2026.7.0,<2026.7.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 = " (tag: 2026.4.0)"
-    git_full = "35d9eb0566bf128267613d46761894dfd25482aa"
-    git_date = "2026-04-29 16:51:56 -0400"
+    git_refnames = " (HEAD -> main, tag: 2026.7.0)"
+    git_full = "609950a67e1d2f26bd98b5053b016200272a35dd"
+    git_date = "2026-07-28 13:11:33 -0400"
     keywords = {"refnames": git_refnames, "full": git_full, "date": git_date}
     return keywords
 


=====================================
s3fs/core.py
=====================================
@@ -425,7 +425,11 @@ class S3FileSystem(AsyncFileSystem):
         self.use_ssl = use_ssl
         self.cache_regions = cache_regions
         self._s3 = None
+        self._set_session_lock = asyncio.Lock()
         self.session = session
+        self._session_is_owned = (
+            session is None
+        )  # False when the caller injected a session
         self.fixed_upload_size = fixed_upload_size
         self.local_expiry_check = local_expiry_check
         if max_concurrency < 1:
@@ -599,7 +603,9 @@ class S3FileSystem(AsyncFileSystem):
         if self._s3 is not None and not refresh:
             hsess = getattr(getattr(self._s3, "_endpoint", None), "http_session", None)
             if hsess is not None:
-                if all(_.closed for _ in hsess._sessions.values()):
+                if hsess._sessions is None or (
+                    hsess._sessions and all(_.closed for _ in hsess._sessions.values())
+                ):
                     refresh = True
             if not refresh:
                 return self._s3
@@ -638,36 +644,43 @@ class S3FileSystem(AsyncFileSystem):
             }
             config_kwargs["signature_version"] = UNSIGNED
 
-        conf = AioConfig(**config_kwargs)
-        if self.session is None or refresh:
-            self.session = aiobotocore.session.AioSession(**self.kwargs)
-
-        for parameters in (config_kwargs, self.kwargs, init_kwargs, client_kwargs):
-            for option in ("region_name", "endpoint_url"):
-                if parameters.get(option):
-                    self.cache_regions = False
-                    break
-        else:
-            cache_regions = self.cache_regions
+        async with self._set_session_lock:
+            # Re-check under the lock: a concurrent task may have set up the
+            # session while we were waiting.
+            if self._s3 is not None and not refresh:
+                return self._s3
+            conf = AioConfig(**config_kwargs)
+            if self.session is None or (refresh and self._session_is_owned):
+                # Only (re)create the AioSession when s3fs owns it
+                self.session = aiobotocore.session.AioSession(**self.kwargs)
+                self._session_is_owned = True
+
+            for parameters in (config_kwargs, self.kwargs, init_kwargs, client_kwargs):
+                for option in ("region_name", "endpoint_url"):
+                    if parameters.get(option):
+                        self.cache_regions = False
+                        break
+            else:
+                cache_regions = self.cache_regions
 
-        logger.debug(
-            "RC: caching enabled? %r (explicit option is %r)",
-            cache_regions,
-            self.cache_regions,
-        )
-        self.cache_regions = cache_regions
-        if self.cache_regions:
-            s3creator = S3BucketRegionCache(
-                self.session, config=conf, **init_kwargs, **client_kwargs
-            )
-            self._s3 = await s3creator.get_client()
-        else:
-            s3creator = self.session.create_client(
-                "s3", config=conf, **init_kwargs, **client_kwargs
+            logger.debug(
+                "RC: caching enabled? %r (explicit option is %r)",
+                cache_regions,
+                self.cache_regions,
             )
-            self._s3 = await s3creator.__aenter__()
+            self.cache_regions = cache_regions
+            if self.cache_regions:
+                s3creator = S3BucketRegionCache(
+                    self.session, config=conf, **init_kwargs, **client_kwargs
+                )
+                self._s3 = await s3creator.get_client()
+            else:
+                s3creator = self.session.create_client(
+                    "s3", config=conf, **init_kwargs, **client_kwargs
+                )
+                self._s3 = await s3creator.__aenter__()
 
-        self._s3creator = s3creator
+            self._s3creator = s3creator
         # the following actually closes the aiohttp connection; use of privates
         # might break in the future, would cause exception at gc time
         if not self.asynchronous:
@@ -839,6 +852,9 @@ class S3FileSystem(AsyncFileSystem):
         versions=False,
     ):
         bucket, key, _ = self.split_path(path)
+        # A caller-supplied prefix is a stem filter, so the listing is partial
+        # and must not be cached. Capture it before prefix is overwritten below.
+        partial = bool(prefix)
         if not prefix:
             prefix = ""
         if key:
@@ -864,7 +880,7 @@ class S3FileSystem(AsyncFileSystem):
             except ClientError as e:
                 raise translate_boto_error(e)
 
-            if delimiter and files and not versions:
+            if delimiter and files and not versions and not partial:
                 self.dircache[path] = files
             return files
         return self.dircache[path]
@@ -1308,15 +1324,7 @@ class S3FileSystem(AsyncFileSystem):
             and (max_concurrency or self.max_concurrency) > 1
         ):
             chunksize = chunksize or self.default_block_size
-            resp = await self._call_s3(
-                "get_object",
-                Bucket=bucket,
-                Key=key,
-                **version_id_kw(version_id or vers),
-                **self.req_kw,
-            )
-            content_length = resp.get("ContentLength", None)
-            resp["Body"].close()
+            content_length = await self._size(path)
 
             if content_length and content_length > chunksize:
                 return await self._cat_file_concurrent(


=====================================
s3fs/tests/test_s3fs.py
=====================================
@@ -3172,14 +3172,14 @@ def test_find_missing_ls(s3):
     assert set(listed_cached) == set(listed_no_cache)
 
 
-def test_session_close():
+def test_session_close(s3):
+    s3.pipe(f"{test_bucket_name}/dir/afile", b"small")
+
     async def run_program(run):
-        s3 = s3fs.S3FileSystem(anon=True, asynchronous=True)
+        s3 = s3fs.S3FileSystem(anon=True, asynchronous=True, endpoint_url=endpoint_uri)
+        s3.invalidate_cache()
         session = await s3.set_session()
-        files = await s3._ls(
-            "s3://noaa-hrrr-bdp-pds/hrrr.20140730/conus/"
-        )  # Random open data store
-        print(f"Number of files {len(files)}")
+        files = await s3._ls(f"{test_bucket_name}/dir")
         await session.close()
 
     import aiobotocore.httpsession
@@ -3189,6 +3189,97 @@ def test_session_close():
     asyncio.run(run_program(False))
 
 
+def test_set_session_sessions_none(s3):
+    """After the HTTP session is closed and aiobotocore sets _sessions=None
+    (aiobotocore 3.x behaviour), set_session must rebuild the client rather
+    than returning the dead one."""
+    s3.pipe(f"{test_bucket_name}/dir/afile", b"small")
+
+    async def run():
+        fs = S3FileSystem(
+            anon=False,
+            asynchronous=True,
+            client_kwargs={"endpoint_url": endpoint_uri},
+            skip_instance_cache=True,
+        )
+        await fs.set_session()
+        original_client = fs._s3
+
+        # Simulate aiobotocore 3.x behaviour: __aexit__ sets _sessions to None.
+        hsess = fs._s3._endpoint.http_session
+        hsess._sessions = None
+
+        # set_session must detect the dead client and create a new one.
+        await fs.set_session()
+        assert (
+            fs._s3 is not original_client
+        ), "set_session should have rebuilt the client when _sessions is None"
+        await fs.set_session(refresh=True)  # clean up
+
+    asyncio.run(run())
+
+
+def test_set_session_concurrent_no_leak(s3):
+    """Concurrent calls to set_session on a fresh instance must not leak
+    clients (i.e. only one client should be created, not N)."""
+    s3.pipe(f"{test_bucket_name}/dir/afile", b"small")
+
+    async def run():
+        S3FileSystem.clear_instance_cache()
+        fs = S3FileSystem(
+            anon=False,
+            asynchronous=True,
+            client_kwargs={"endpoint_url": endpoint_uri},
+            skip_instance_cache=True,
+        )
+        # All coroutines start with _s3 == None; fire them simultaneously.
+        results = await asyncio.gather(*[fs.set_session() for _ in range(8)])
+        # Every coroutine must get back the same client object.
+        assert (
+            len(set(id(r) for r in results)) == 1
+        ), "Concurrent set_session calls returned different client objects"
+        # Only one client should be alive (no leaked extras).
+        assert fs._s3 is results[0]
+        await fs.set_session(refresh=True)  # clean up
+
+    asyncio.run(run())
+
+
+def test_set_session_preserves_injected_session(s3):
+    """A session= injected at construction time must not be replaced when
+    set_session triggers a refresh due to closed HTTP connections."""
+    s3.pipe(f"{test_bucket_name}/dir/afile", b"small")
+
+    async def run():
+        import aiobotocore.session as aio_session
+
+        custom_session = aio_session.AioSession()
+        fs = S3FileSystem(
+            anon=False,
+            asynchronous=True,
+            session=custom_session,
+            client_kwargs={"endpoint_url": endpoint_uri},
+            skip_instance_cache=True,
+        )
+        await fs.set_session()
+        assert (
+            fs.session is custom_session
+        ), "session should not be replaced on first connect"
+
+        # Simulate all HTTP connections being closed so set_session will refresh.
+        hsess = fs._s3._endpoint.http_session
+        if hsess._sessions:
+            for sess in hsess._sessions.values():
+                await sess.close()
+
+        await fs.set_session()
+        assert (
+            fs.session is custom_session
+        ), "set_session must not replace a user-injected session on refresh"
+
+    asyncio.run(run())
+
+
 def test_rm_recursive_prfix(s3):
     prefix = "logs/"  # must end with "/"
 
@@ -3198,3 +3289,108 @@ def test_rm_recursive_prfix(s3):
     logs_path = f"s3://{test_bucket_name}/{prefix}"
     s3.rm(logs_path, recursive=True)
     assert not s3.isdir(logs_path)
+
+
+def test_set_session_closed_sessions_rebuilds_once(s3):
+    """Regression test for the #1019 perf regression: a populated-but-all-closed
+    _sessions dict must rebuild the client exactly once, then reuse it. The
+    rebuilt client has an empty _sessions dict, which must not count as
+    "all closed" (vacuous all([]) == True) and force a refresh on every call.
+    """
+    import aiobotocore.session as aio_session
+    from unittest import mock
+
+    s3.pipe(f"{test_bucket_name}/dir/afile", b"small")
+
+    create_client_calls = 0
+    original_create_client = aio_session.AioSession._create_client
+
+    async def counting_create_client(self, *args, **kwargs):
+        nonlocal create_client_calls
+        create_client_calls += 1
+        return await original_create_client(self, *args, **kwargs)
+
+    async def run():
+        fs = S3FileSystem(
+            anon=False,
+            asynchronous=True,
+            client_kwargs={"endpoint_url": endpoint_uri},
+            skip_instance_cache=True,
+        )
+        await fs._ls(f"{test_bucket_name}/dir")  # populates _sessions
+        sessions = fs._s3._endpoint.http_session._sessions
+        assert sessions, "expected a populated _sessions dict after an op"
+
+        for sess in sessions.values():
+            await sess.close()
+
+        baseline = create_client_calls  # ignore the warmup build above
+        iterations = 10
+        for _ in range(iterations):
+            await fs.set_session()
+
+        rebuilds = create_client_calls - baseline
+        assert rebuilds == 1, (
+            f"set_session rebuilt the client {rebuilds} times across {iterations} "
+            f"calls; expected 1. >1 means the empty-dict case forces a refresh on "
+            f"every call (#1019 regression)."
+        )
+        await fs.set_session(refresh=True)  # clean up
+
+    with mock.patch.object(
+        aio_session.AioSession, "_create_client", counting_create_client
+    ):
+        asyncio.run(run())
+
+
+def test_find_with_prefix_does_not_poison_dircache(s3):
+    data_dir = test_bucket_name + "/splits"
+    s3.touch(data_dir + "/train-00000.parquet")
+    s3.touch(data_dir + "/test-00000.parquet")
+    s3.invalidate_cache()
+
+    train_results = s3.find(data_dir, prefix="train-", maxdepth=1)
+    assert len(train_results) == 1
+    assert data_dir + "/train-00000.parquet" in train_results
+
+    all_files = {f.split("/")[-1] for f in s3.ls(data_dir)}
+    assert all_files == {"train-00000.parquet", "test-00000.parquet"}
+
+
+def test_glob_prefix_does_not_poison_dircache(s3):
+    data_dir = test_bucket_name + "/globs"
+    s3.touch(data_dir + "/train-00000.parquet")
+    s3.touch(data_dir + "/test-00000.parquet")
+    s3.invalidate_cache()
+
+    assert len(s3.glob(data_dir + "/train-*")) == 1
+    test_hits = s3.glob(data_dir + "/test-*")
+    assert len(test_hits) == 1
+    assert data_dir + "/test-00000.parquet" in test_hits
+
+
+def test_find_with_prefix_preserves_existing_full_cache(s3):
+    data_dir = test_bucket_name + "/warm"
+    s3.touch(data_dir + "/train-00000.parquet")
+    s3.touch(data_dir + "/test-00000.parquet")
+    s3.invalidate_cache()
+
+    assert len(s3.ls(data_dir)) == 2
+
+    s3.find(data_dir, prefix="train-", maxdepth=1)
+
+    cached = {f.split("/")[-1] for f in s3.ls(data_dir)}
+    assert cached == {"train-00000.parquet", "test-00000.parquet"}
+
+
+def test_normal_ls_and_find_still_populate_dircache(s3):
+    data_dir = test_bucket_name + "/normal"
+    s3.touch(data_dir + "/file-a")
+    s3.touch(data_dir + "/file-b")
+    s3.invalidate_cache()
+
+    assert data_dir not in s3.dircache
+
+    s3.ls(data_dir)
+    assert data_dir in s3.dircache
+    assert len(s3.dircache[data_dir]) == 2



View it on GitLab: https://salsa.debian.org/debian-gis-team/python-s3fs/-/commit/ca03e1601493df184c48d4f7b90f9a4ef736e8bb

-- 
View it on GitLab: https://salsa.debian.org/debian-gis-team/python-s3fs/-/commit/ca03e1601493df184c48d4f7b90f9a4ef736e8bb
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/20260801/966d0958/attachment-0001.htm>


More information about the Pkg-grass-devel mailing list