From 460c4cbe5d4ac623111efe1564044466f81f334e Mon Sep 17 00:00:00 2001 From: laughingman7743 Date: Sun, 4 Oct 2026 09:18:43 +0900 Subject: [PATCH 1/5] Allow "?" in keys and address versions in listings, files and copies Only a version ID query at the end of a path is a version now, so keys containing "?" can be read, listed and deleted (S3Path.PATTERN and the new S3Path.split_version_id(), also used by invalidate_cache()). A path with a version names that version: expand_path() no longer globs its "?", and copy(), get() and mv() (sync and aio) pair it with a destination named after its key instead of the version-qualified name. ls(versions=True) names each version "bucket/key?versionId=", except the "null" version, as s3fs does, and a version-aware S3File carries the version pinned at open time in its path, so that metadata(), getxattr() and url() describe that version. Co-Authored-By: Claude Opus 5.5 --- docs/filesystem.md | 14 +- pyathena/filesystem/s3.py | 231 ++++++++++++++++++--- pyathena/filesystem/s3_async.py | 94 ++++++++- pyathena/filesystem/s3_path.py | 33 ++- tests/pyathena/filesystem/test_s3.py | 170 ++++++++++++++- tests/pyathena/filesystem/test_s3_async.py | 35 +++- tests/pyathena/filesystem/test_s3_path.py | 26 ++- 7 files changed, 546 insertions(+), 57 deletions(-) diff --git a/docs/filesystem.md b/docs/filesystem.md index e6c7521f..cc961306 100644 --- a/docs/filesystem.md +++ b/docs/filesystem.md @@ -227,8 +227,13 @@ fs.clear_multipart_uploads("s3://YOUR_S3_BUCKET/path/to/") With `version_aware=True`, reads pin the object version observed at open time, so a file handle keeps returning consistent data even if the object is overwritten while -reading. Explicit versions can always be read with the `?versionId=` suffix or the -`version_id` argument. +reading. The file path carries the pinned version as a `?versionId=` suffix, which +`metadata()`, `getxattr()` and `url()` of the file also use. Explicit versions can +always be read with the `?versionId=` suffix or the `version_id` argument. Only a +`?versionId=` (or `?versionID=`, `?versionid=`, `?version_id=`) query at the end of a +path is a version; any other `?` is part of the key. A path with a version is not a +glob pattern: `copy()`, `mv()` and `get()` copy that version to a destination named +after its key. ```python fs = S3FileSystem( @@ -239,8 +244,9 @@ fs = S3FileSystem( with fs.open("s3://YOUR_S3_BUCKET/path/to/object", "rb") as f: data = f.read() # Pinned to the version observed at open time. -# List all versions of the objects under a prefix. -fs.ls("s3://YOUR_S3_BUCKET/path/to/", versions=True, detail=True) +# List all versions of the objects under a prefix. Each version is named +# "bucket/key?versionId=", except the "null" version, which is named by its key. +fs.ls("s3://YOUR_S3_BUCKET/path/to/", versions=True) # Typed version information, including delete markers if requested. versions = fs.object_version_info("s3://YOUR_S3_BUCKET/path/to/object") diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py index 042488db..0ee2afe8 100644 --- a/pyathena/filesystem/s3.py +++ b/pyathena/filesystem/s3.py @@ -29,7 +29,7 @@ from fsspec.callbacks import _DEFAULT_CALLBACK, Callback from fsspec.compression import compr from fsspec.core import get_compression -from fsspec.implementations.local import trailing_sep +from fsspec.implementations.local import LocalFileSystem, make_path_posix, trailing_sep from fsspec.spec import AbstractBufferedFile from fsspec.utils import isfilelike, other_paths, tokenize @@ -69,6 +69,23 @@ _LOOKUPS_CACHE_KEY = "lookups" +def _has_version_id(path: str | list[str]) -> bool: + """Return whether a path, or a path of a list, ends with a version ID query. + + fsspec's ``copy()`` and ``get()`` take ``?`` for a glob character and + name the destinations after the sources, so sources with a version ID are + paired with their destinations by ``S3FileSystem._copy_paths``. + + Args: + path: A path or a list of paths. + + Returns: + Whether a path has a version ID query. + """ + paths = [path] if isinstance(path, str) else path + return any(S3Path.split_version_id(str(p))[1] for p in paths) + + class CompressedBuffer(BytesIO): """An in-memory buffer of data compressed with a codec of fsspec. @@ -381,8 +398,19 @@ def _directory_object(bucket: str, key: str | None, version_id: str | None = Non @staticmethod def _versioned_file_object(bucket: str, version: dict[str, Any]) -> S3Object: - """Build an S3Object from a ListObjectVersions Versions entry.""" - return S3Object( + """Build an S3Object from a ListObjectVersions Versions entry. + + Its name addresses the version with a ``?versionId=`` query, except + for the ``null`` version, which a write to the key replaces. + + Args: + bucket: The bucket of the version. + version: The ``Versions`` entry of the ListObjectVersions response. + + Returns: + The file object of the version. + """ + file = S3Object( init=version, type=S3ObjectType.S3_OBJECT_TYPE_FILE, bucket=bucket, @@ -390,6 +418,9 @@ def _versioned_file_object(bucket: str, version: dict[str, Any]) -> S3Object: version_id=version.get("VersionId"), is_latest=version.get("IsLatest", False), ) + if file.version_id != "null": + file.name = str(S3Path(bucket, version["Key"], file.version_id)) + return file def _head_bucket( self, @@ -654,8 +685,10 @@ def ls( detail: If True, return S3Object instances; if False, return paths as strings. refresh: If True, bypass cache and fetch fresh results from S3. **kwargs: Additional arguments including: - versions: If True, list all versions of the objects. Requires - the filesystem to be constructed with ``version_aware=True``. + versions: If True, list all versions of the objects, named + ``bucket/key?versionId=`` except for the ``null`` + version. Requires the filesystem to be constructed with + ``version_aware=True``. Returns: List of S3Object instances (if detail=True) or paths as strings (if detail=False). @@ -1059,6 +1092,48 @@ def find( return {f.name: f for f in files} return [f.name for f in files] + def expand_path(self, path, recursive=False, maxdepth=None, **kwargs) -> list[str]: + """Expand glob patterns and directories into the paths they match. + + As in fsspec, except that a path with a version ID names that version + of an object: it is not a glob pattern, although its ``?`` is one in + fsspec, nor is anything expanded below it. With ``recursive``, it is + included only if the version exists. + + Args: + path: S3 path, glob pattern, or list of them. + recursive: Whether to include the paths below the directories. + maxdepth: Maximum depth of the expansion, at least 1. + **kwargs: Additional arguments passed to ``glob`` and ``find``. + + Returns: + The sorted matching paths. + + Raises: + ValueError: If ``maxdepth`` is less than 1. + FileNotFoundError: If nothing matches. + """ + if maxdepth is not None and maxdepth < 1: + raise ValueError("maxdepth must be at least 1") + paths = [ + self._strip_protocol(p) + for p in ([path] if isinstance(path, (str, os.PathLike)) else path) + ] + versions = [p for p in paths if S3Path.split_version_id(p)[1]] + others = [p for p in paths if not S3Path.split_version_id(p)[1]] + out = {p for p in versions if not recursive or self.exists(p)} + if others: + try: + out.update( + super().expand_path(others, recursive=recursive, maxdepth=maxdepth, **kwargs) + ) + except FileNotFoundError: + if not out: + raise + if not out: + raise FileNotFoundError(path) + return sorted(out) + def exists(self, path: str, **kwargs) -> bool: """Check if an S3 path exists. @@ -1182,7 +1257,8 @@ def _expand_delete_paths( unversioned_paths.append(p) if unversioned_paths: - # expand_path treats "?" as a wildcard, so versioned paths skip it. + # Versioned paths are deleted as given, without the lookup that + # expand_path makes for them with recursive. unversioned_paths = self.expand_path( unversioned_paths, recursive=recursive, maxdepth=maxdepth ) @@ -1575,27 +1651,7 @@ def _move_paths( except for a directory with no object at its key, which is not copied. """ - if isinstance(path1, list) and isinstance(path2, list): - paths1, paths2 = path1, path2 - else: - source_is_str = isinstance(path1, str) - paths1 = self.expand_path(path1, recursive=recursive, maxdepth=maxdepth) - if source_is_str and (not recursive or maxdepth is not None): - # Non-recursive glob does not copy directories. - paths1 = [p for p in paths1 if not (trailing_sep(p) or self.isdir(p))] - if not paths1: - return [] - # The destination is looked up only when it decides the mapping. - exists = source_is_str and ( - (has_magic(path1) and len(paths1) == 1) - or ( - not has_magic(path1) - and not trailing_sep(path1) - and isinstance(path2, str) - and (trailing_sep(path2) or self.isdir(path2)) - ) - ) - paths2 = other_paths(paths1, path2, exists=exists, flatten=not source_is_str) + paths1, paths2 = self._copy_paths(path1, path2, recursive=recursive, maxdepth=maxdepth) # The paths are copied as given, and compared by what they name. named = [ (p1, p2, self._move_target(p1), self._move_target(p2)) @@ -1634,6 +1690,114 @@ def _move_paths( raise ValueError("Cannot move a path onto another path that is moved.") return pairs + def _copy_paths( + self, + path1: str | list[str], + path2: str | list[str], + recursive: bool = False, + maxdepth: int | None = None, + isdir: Callable[[str], bool] | None = None, + ) -> tuple[list[str], list[str]]: + """Pair the sources of a copy with their destinations as fsspec's ``copy()`` does. + + A source with a version ID names that version of an object: it is not + a glob pattern, and its destination is named after its key without + the version. + + Args: + path1: Source S3 path, glob pattern, or list of them. + path2: Destination path, or list of paths when ``path1`` is a + list. + recursive: Whether to include the contents of the directories. + maxdepth: Maximum depth of the expansion. + isdir: Whether a destination path is a directory, by default + ``self.isdir``; ``get()`` passes the local filesystem's. + + Returns: + The sources and their destinations. Both are empty if ``path1`` + is a string that matches only directories without ``recursive``. + """ + if isinstance(path1, list) and isinstance(path2, list): + return path1, path2 + source_is_str = isinstance(path1, str) + paths1 = self.expand_path(path1, recursive=recursive, maxdepth=maxdepth) + if source_is_str and (not recursive or maxdepth is not None): + # Non-recursive glob does not copy directories. + paths1 = [p for p in paths1 if not (trailing_sep(p) or self.isdir(p))] + if not paths1: + return [], [] + glob = isinstance(path1, str) and has_magic(path1) and not _has_version_id(path1) + # The destination is looked up only when it decides the mapping. + exists = source_is_str and ( + (glob and len(paths1) == 1) + or ( + not glob + and not trailing_sep(path1) + and isinstance(path2, str) + and (trailing_sep(path2) or (isdir or self.isdir)(path2)) + ) + ) + names = [S3Path.split_version_id(p)[0] for p in paths1] + return paths1, other_paths(names, path2, exists=exists, flatten=not source_is_str) + + def copy(self, path1, path2, recursive=False, maxdepth=None, on_error=None, **kwargs) -> None: + """Copy files within S3. + + As fsspec's ``copy()``, except that a source with a version ID copies + that version of the object to a destination named after its key, as + ``_copy_paths`` pairs them. + + Args: + path1: Source S3 path, glob pattern, or list of them. + path2: Destination S3 path, or list of paths when ``path1`` is a + list. + recursive: Whether to copy the directories with their contents. + maxdepth: Maximum depth of a recursive copy. + on_error: ``"raise"`` or ``"ignore"`` for a missing source; by + default ``"ignore"`` with ``recursive`` and ``"raise"`` + otherwise. + **kwargs: Additional S3 copy parameters passed to ``cp_file()``. + """ + if isinstance(path2, str) and _has_version_id(path1): + path1, path2 = self._copy_paths(path1, path2, recursive=recursive, maxdepth=maxdepth) + if not path1: + return + super().copy( + path1, path2, recursive=recursive, maxdepth=maxdepth, on_error=on_error, **kwargs + ) + + def get( + self, rpath, lpath, recursive=False, callback=_DEFAULT_CALLBACK, maxdepth=None, **kwargs + ) -> None: + """Copy files from S3 to the local filesystem. + + As fsspec's ``get()``, except that a source with a version ID + downloads that version of the object to a local path named after its + key, as ``_copy_paths`` pairs them. + + Args: + rpath: Source S3 path, glob pattern, or list of them. + lpath: Local destination path, or list of paths when ``rpath`` is + a list. + recursive: Whether to copy the directories with their contents. + callback: Progress callback. + maxdepth: Maximum depth of a recursive copy. + **kwargs: Additional parameters passed to ``get_file()``. + """ + if isinstance(lpath, str) and _has_version_id(rpath): + rpath, lpath = self._copy_paths( + rpath, + make_path_posix(lpath), + recursive=recursive, + maxdepth=maxdepth, + isdir=LocalFileSystem().isdir, + ) + if not rpath: + return + super().get( + rpath, lpath, recursive=recursive, callback=callback, maxdepth=maxdepth, **kwargs + ) + def _move_target(self, path: str) -> str: """Return what a path of a move names, for comparing the paths. @@ -3063,12 +3227,9 @@ def invalidate_cache(self, path: str | None = None) -> None: if not path: self._evict_cache("") while path: - # parse_path does not accept "?" in keys, so it starts the - # versionId query. - base, _, query = path.partition("?") + base, version_id = S3Path.split_version_id(path) cache_paths = [path] - if query: - version_id = query.partition("=")[2] + if version_id: cache_paths.extend( f"{base}?{name}={version_id}" for name in ("versionId", "versionID", "versionid", "version_id") @@ -3086,7 +3247,7 @@ def invalidate_cache(self, path: str | None = None) -> None: self._evict_cache(cache_key) # A version-qualified path continues with the path without # the version. - path = self._strip_protocol(base) if query else self._parent(path) + path = self._strip_protocol(base) if version_id else self._parent(path) def _evict_cache(self, key: str | tuple[str, str]) -> None: """Remove a dircache entry if it exists. @@ -3614,6 +3775,10 @@ def __init__( # consistent even if the object is overwritten. info() heads # the object when the cached entry carries no version. self.version_id = info.get("version_id") + if self.version_id: + # Carried in the path as an explicit version is, so that + # the methods of the file use it too. + path = f"{path}?versionId={self.version_id}" if etag := info.get("etag"): self.s3_additional_kwargs.update({"IfMatch": etag}) self._details = info diff --git a/pyathena/filesystem/s3_async.py b/pyathena/filesystem/s3_async.py index 0d98a28a..55122d88 100644 --- a/pyathena/filesystem/s3_async.py +++ b/pyathena/filesystem/s3_async.py @@ -19,8 +19,9 @@ from fsspec.asyn import AsyncFileSystem, sync from fsspec.callbacks import _DEFAULT_CALLBACK from fsspec.core import get_compression +from fsspec.implementations.local import LocalFileSystem, make_path_posix -from pyathena.filesystem.s3 import CompressedBuffer, S3File, S3FileSystem +from pyathena.filesystem.s3 import CompressedBuffer, S3File, S3FileSystem, _has_version_id from pyathena.filesystem.s3_executor import S3AioExecutor, S3Executor, S3ThreadPoolExecutor from pyathena.filesystem.s3_object import ( S3Metadata, @@ -392,6 +393,79 @@ def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> None: """ sync(self.loop, self._mv, path1, path2, recursive=recursive, maxdepth=maxdepth, **kwargs) + async def _copy( + self, + path1, + path2, + recursive=False, + on_error=None, + maxdepth=None, + batch_size=None, + **kwargs, + ) -> None: + """Copy files within S3. + + See :meth:`S3FileSystem.copy`. The copies run as in fsspec's + ``_copy()``. + + Args: + path1: Source S3 path, glob pattern, or list of them. + path2: Destination S3 path, or list of paths when ``path1`` is a + list. + recursive: Whether to copy the directories with their contents. + on_error: ``"raise"`` or ``"ignore"`` for a missing source. + maxdepth: Maximum depth of a recursive copy. + batch_size: Number of copies to run at the same time. + **kwargs: Additional S3 copy parameters passed to ``_cp_file()``. + """ + if isinstance(path2, str) and _has_version_id(path1): + path1, path2 = await asyncio.to_thread( + self._sync_fs._copy_paths, path1, path2, recursive=recursive, maxdepth=maxdepth + ) + if not path1: + return + await super()._copy( + path1, + path2, + recursive=recursive, + on_error=on_error, + maxdepth=maxdepth, + batch_size=batch_size, + **kwargs, + ) + + async def _get( + self, rpath, lpath, recursive=False, callback=_DEFAULT_CALLBACK, maxdepth=None, **kwargs + ) -> None: + """Copy files from S3 to the local filesystem. + + See :meth:`S3FileSystem.get`. The downloads run as in fsspec's + ``_get()``. + + Args: + rpath: Source S3 path, glob pattern, or list of them. + lpath: Local destination path, or list of paths when ``rpath`` is + a list. + recursive: Whether to copy the directories with their contents. + callback: Progress callback. + maxdepth: Maximum depth of a recursive copy. + **kwargs: Additional parameters passed to ``_get_file()``. + """ + if isinstance(lpath, str) and _has_version_id(rpath): + rpath, lpath = await asyncio.to_thread( + self._sync_fs._copy_paths, + rpath, + make_path_posix(lpath), + recursive=recursive, + maxdepth=maxdepth, + isdir=LocalFileSystem().isdir, + ) + if not rpath: + return + await super()._get( + rpath, lpath, recursive=recursive, callback=callback, maxdepth=maxdepth, **kwargs + ) + async def _cp_file(self, path1: str, path2: str, **kwargs) -> None: """Copy an S3 object, using async parallel multipart upload for large files. @@ -657,6 +731,24 @@ async def _find( return {f.name: f for f in files} return [f.name for f in files] + async def _expand_path(self, path, recursive=False, maxdepth=None, **kwargs) -> list[str]: + """Expand glob patterns and directories into the paths they match. + + See :meth:`S3FileSystem.expand_path`. + + Args: + path: S3 path, glob pattern, or list of them. + recursive: Whether to include the paths below the directories. + maxdepth: Maximum depth of the expansion, at least 1. + **kwargs: Additional arguments passed to ``glob`` and ``find``. + + Returns: + The sorted matching paths. + """ + return await asyncio.to_thread( + self._sync_fs.expand_path, path, recursive=recursive, maxdepth=maxdepth, **kwargs + ) + def _create_executor(self, max_workers: int) -> S3Executor: """Create the executor for the parallel operations of a file. diff --git a/pyathena/filesystem/s3_path.py b/pyathena/filesystem/s3_path.py index 3013092c..7d546df8 100644 --- a/pyathena/filesystem/s3_path.py +++ b/pyathena/filesystem/s3_path.py @@ -24,8 +24,9 @@ class S3Path: Paths are parsed from strings such as ``s3://bucket/key``, ``s3a://bucket/key`` or ``bucket/key``, optionally followed by a version ID query (``?versionId=``, ``?versionID=``, ``?versionid=`` or - ``?version_id=``). The root path, which names no bucket, is not an - ``S3Path``. + ``?version_id=``). Only a query at the end of the path is a version ID; + any other ``?`` is part of the key. The root path, which names no bucket, + is not an ``S3Path``. Attributes: bucket: The name of the bucket. @@ -46,9 +47,13 @@ class S3Path: 's3://bucket/dir/key?versionId=v1' """ + # Version IDs do not contain "?", so only the last query can be one. + VERSION_QUERY: ClassVar[Pattern[str]] = re.compile( + r"\?version(Id|ID|id|_id)=(?P[^?]+)$" + ) PATTERN: ClassVar[Pattern[str]] = re.compile( - r"(^s3://|^s3a://|^)(?P[a-zA-Z0-9.\-_]+)(/(?P[^?]+)|/)?" - r"($|\?version(Id|ID|id|_id)=(?P.+)$)" + r"(^s3://|^s3a://|^)(?P[a-zA-Z0-9.\-_]+)(/(?P.+?)|/)?" + rf"($|{VERSION_QUERY.pattern})" ) bucket: str @@ -74,6 +79,26 @@ def parse(cls, path: str) -> S3Path: raise ValueError(f"Invalid S3 path format {path}.") return cls(match.group("bucket"), match.group("key"), match.group("version_id")) + @classmethod + def split_version_id(cls, path: str) -> tuple[str, str | None]: + """Split the version ID query from the end of a path string. + + Unlike :meth:`parse`, any string is accepted, such as the names that + fsspec builds. + + Args: + path: The path string. + + Returns: + Tuple of the string without the version ID query and the version + ID, or of the string itself and None if it ends with no version + ID query. + """ + match = cls.VERSION_QUERY.search(path) + if not match: + return path, None + return path[: match.start()], match.group("version_id") + @property def is_bucket(self) -> bool: """Whether the path names the bucket: it has no key, or a key of only slashes.""" diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py index 49ee05eb..135090dc 100644 --- a/tests/pyathena/filesystem/test_s3.py +++ b/tests/pyathena/filesystem/test_s3.py @@ -38,6 +38,7 @@ from pyathena.filesystem.s3_errors import S3ClientError from pyathena.filesystem.s3_executor import S3AioExecutor, S3ThreadPoolExecutor from pyathena.filesystem.s3_object import S3Object, S3ObjectType, S3StorageClass +from pyathena.filesystem.s3_path import S3Path from pyathena.util import RetryConfig from tests import ENV from tests.pyathena.conftest import connect @@ -148,17 +149,16 @@ def test_parse_path_invalid(self): with pytest.raises(ValueError, match="Invalid S3 path format"): S3FileSystem.parse_path("s3://bucket?foo=bar") - with pytest.raises(ValueError, match="Invalid S3 path format"): - S3FileSystem.parse_path("s3://bucket/path/to/obj?foo=bar") - with pytest.raises(ValueError, match="Invalid S3 path format"): S3FileSystem.parse_path("s3a://bucket?") with pytest.raises(ValueError, match="Invalid S3 path format"): S3FileSystem.parse_path("s3a://bucket?foo=bar") - with pytest.raises(ValueError, match="Invalid S3 path format"): - S3FileSystem.parse_path("s3a://bucket/path/to/obj?foo=bar") + # GH-979: a "?" in a key that does not start a trailing version ID + # query is part of the key. + for path in ("s3://bucket/path/to/obj?foo=bar", "s3a://bucket/path/to/obj?foo=bar"): + assert S3FileSystem.parse_path(path) == ("bucket", "path/to/obj?foo=bar", None) @staticmethod def _make_fs(): @@ -1654,6 +1654,105 @@ def test_mv_versions_onto_their_key(self): ] assert deletes == [[{"Key": "b", "VersionId": "v1"}]] + def test_question_mark_keys(self): + # GH-979: keys containing "?" are keys, not version ID queries. + fs = self._make_fs() + keys = self._serve_keys(fs, {"dir/a.txt", "dir/what?.txt", "dir/q?version_id"}) + + assert fs.ls("s3://bucket/dir") == [ + "bucket/dir/a.txt", + "bucket/dir/q?version_id", + "bucket/dir/what?.txt", + ] + assert fs.info("s3://bucket/dir/what?.txt")["name"] == "bucket/dir/what?.txt" + fs.invalidate_cache() + assert fs.exists("s3://bucket/dir/q?version_id") + heads = [c.kwargs for c in fs._call.call_args_list if c.args[0] is fs._client.head_object] + assert heads[-1] == {"Bucket": "bucket", "Key": "dir/q?version_id"} + + fs.rm("s3://bucket/dir", recursive=True) + assert keys == set() + + def test_invalidate_cache_question_mark_key(self): + # A "?" that does not start a trailing version ID query is part of + # the key, whose parents are invalidated. + fs = self._make_fs() + for key in ("bucket/dir/what?.txt", ("bucket/dir", "/"), "bucket/dir/what"): + fs.dircache[key] = [] + + fs.invalidate_cache("s3://bucket/dir/what?.txt") + assert list(fs.dircache) == ["bucket/dir/what"] + + @pytest.mark.parametrize("recursive", [False, True]) + def test_expand_path_version(self, recursive): + # GH-979: "?" of a version ID query is not a glob character, and a + # version is not expanded below its key. + fs = self._make_fs() + self._serve_keys(fs, {"b", "bc", "c"}) + + assert fs.expand_path("s3://bucket/b?versionId=v1", recursive=recursive) == [ + "bucket/b?versionId=v1" + ] + assert fs.expand_path( + ["s3://bucket/b?versionId=v1", "s3://bucket/c*"], recursive=recursive + ) == ["bucket/b?versionId=v1", "bucket/c"] + with pytest.raises(ValueError, match="maxdepth"): + fs.expand_path("s3://bucket/b?versionId=v1", maxdepth=0) + + def test_expand_path_recursive_missing_version(self): + # With recursive, a version is included only if it exists, as fsspec + # does for the other paths. + fs = self._make_fs() + self._serve_keys(fs, set()) + + with pytest.raises(FileNotFoundError): + fs.expand_path("s3://bucket/b?versionId=v1", recursive=True) + + @pytest.mark.parametrize( + ("path1", "path2", "expected"), + [ + ("s3://bucket/b?versionId=v1", "s3://bucket/out", "out"), + ("s3://bucket/b?version_id=v1", "s3://bucket/d/", "d/b"), + (["s3://bucket/b?versionId=v1"], "s3://bucket/d", "d/b"), + # The key may contain "?" too. + ("s3://bucket/q?x?versionId=v1", "s3://bucket/d/", "d/q?x"), + ], + ) + @pytest.mark.parametrize("method", ["copy", "mv"]) + def test_copy_version(self, method, path1, path2, expected): + # GH-979: a version is copied to a destination named after its key, + # not globbed with "?" as a wildcard. + fs = self._make_fs() + self._serve_keys(fs, {"b", "q?x", "d/x"}) + + getattr(fs, method)(path1, path2) + copies = [c.kwargs for c in fs._call.call_args_list if c.args[0] is fs._client.copy_object] + source = S3Path.parse(path1 if isinstance(path1, str) else path1[0]) + assert [(c["CopySource"], c["Key"]) for c in copies] == [ + ({"Bucket": "bucket", "Key": source.key, "VersionId": "v1"}, expected) + ] + + @pytest.mark.parametrize( + ("rpath", "lpath", "expected"), + [ + ("s3://bucket/key?versionId=v1", "f.txt", "f.txt"), + ("s3://bucket/key?versionId=v1", "d/", "d/key"), + (["s3://bucket/key?versionId=v1"], "d", "d/key"), + ], + ) + def test_get_version(self, tmp_path, monkeypatch, rpath, lpath, expected): + # GH-979: a version is downloaded to a local path named after its + # key, which used to be the version-qualified name of the source. + monkeypatch.chdir(tmp_path) + (tmp_path / "d").mkdir() + fs, _ = self._make_object_fs(b"data") + + fs.get(rpath, lpath) + assert sorted(str(p.relative_to(tmp_path)) for p in tmp_path.rglob("*")) == sorted( + {"d", expected} + ) + assert (tmp_path / expected).read_bytes() == b"data" + def test_mv_nothing_within_maxdepth(self): # Only directories within maxdepth: nothing is moved, as with copy(). fs = self._make_fs() @@ -2429,6 +2528,32 @@ def test_open_version_id(self): with pytest.raises(ValueError, match="do not match"): fs.open("s3://bucket/key?versionId=v2", "rb", version_id="v1") + def test_open_version_aware_pins_version_in_path(self): + # GH-979: the version observed at open time is carried in the path as + # an explicit version is, so that the metadata, attributes and URL of + # the file describe that version, not the latest one. + fs = self._make_fs() + fs.version_aware = True + fs.default_cache_type = "bytes" + fs._call.side_effect = lambda method, **request: ( + "https://signed" + if method is fs._client.generate_presigned_url + else {"ContentLength": 4, "ETag": '"e"', "VersionId": "v1", "Metadata": {"a": "1"}} + ) + + with fs.open("s3://bucket/key", "rb") as f: + assert f.version_id == "v1" + assert f.path == "bucket/key?versionId=v1" + assert f.metadata()["a"] == "1" + assert f.getxattr("a") == "1" + assert f.url() == "https://signed" + requests = [c.kwargs for c in fs._call.call_args_list] + # The open-time lookup, then metadata(), getxattr() and url(). + assert requests[1:3] == [{"Bucket": "bucket", "Key": "key", "VersionId": "v1"}] * 2 + assert requests[3]["Params"] == {"Bucket": "bucket", "Key": "key", "VersionId": "v1"} + # A reopened (e.g., unpickled) file reads the same version. + assert f.__reduce__()[1][1] == "bucket/key?versionId=v1" + @pytest.mark.parametrize("mode", ["wb", "ab", "xb"]) @pytest.mark.parametrize( ("path", "kwargs"), @@ -3510,6 +3635,7 @@ def test_ls_versions(self): "Versions": [ {"Key": "path/key", "VersionId": "v2", "IsLatest": True, "Size": 4}, {"Key": "path/key", "VersionId": "v1", "IsLatest": False, "Size": 2}, + {"Key": "path/other", "VersionId": "null", "IsLatest": True, "Size": 1}, ], "IsTruncated": False, } @@ -3518,13 +3644,17 @@ def test_ls_versions(self): fs._call.assert_called_once_with( fs._client.list_object_versions, Bucket="bucket", Prefix="path/", Delimiter="/" ) + # GH-979: each version is named so that it can be addressed, except + # the "null" version, which a write to the key replaces. assert [(f.name, f.version_id, f.is_latest) for f in actual] == [ ("bucket/path/dir", None, None), - ("bucket/path/key", "v2", True), - ("bucket/path/key", "v1", False), + ("bucket/path/key?versionId=v2", "v2", True), + ("bucket/path/key?versionId=v1", "v1", False), + ("bucket/path/other", "null", True), ] assert actual[1].size == 4 assert actual[2].size == 2 + assert fs.ls("s3://bucket/path", versions=True) == [f.name for f in actual] def test_ls_versions_object_path_falls_back_to_the_key(self): fs = self._make_fs() @@ -3548,8 +3678,8 @@ def test_ls_versions_object_path_falls_back_to_the_key(self): fs._client.list_object_versions, Bucket="bucket", Prefix="path/key", Delimiter="/" ) assert [(f.name, f.version_id, f.size) for f in actual] == [ - ("bucket/path/key", "v2", 4), - ("bucket/path/key", "v1", 2), + ("bucket/path/key?versionId=v2", "v2", 4), + ("bucket/path/key?versionId=v1", "v1", 2), ] def test_dir_filesystem(self): @@ -4936,6 +5066,28 @@ def test_read_null_version(self, fs): with fs.open(f"{path}?versionId=null", "rb") as f: assert f.read() == data + def test_question_mark_keys_and_null_version(self, fs, tmp_path): + # GH-979: keys containing "?" can be written, listed, read and + # deleted, and a version path, here the "null" version of an + # unversioned bucket, is copied and downloaded under its key. + base = ( + f"{ENV.s3_staging_bucket}/{ENV.s3_staging_key}{ENV.schema}/" + f"filesystem/test_question_mark_keys/{uuid.uuid4()}" + ) + fs.pipe(f"s3://{base}/what?.txt", b"1") + assert fs.ls(f"s3://{base}") == [f"{base}/what?.txt"] + assert fs.info(f"s3://{base}/what?.txt")["size"] == 1 + assert fs.cat(f"s3://{base}/what?.txt") == b"1" + + fs.copy(f"s3://{base}/what?.txt?versionId=null", f"s3://{base}/copy/") + assert fs.cat(f"s3://{base}/copy/what?.txt") == b"1" + fs.get(f"s3://{base}/what?.txt?versionId=null", f"{tmp_path}/") + assert (tmp_path / "what?.txt").read_bytes() == b"1" + + fs.rm(f"s3://{base}", recursive=True) + assert not fs.exists(f"s3://{base}/what?.txt") + assert not fs.exists(f"s3://{base}/copy/what?.txt") + def test_file_url_metadata_getxattr_setxattr(self, fs): path = ( f"s3://{ENV.s3_staging_bucket}/{ENV.s3_staging_key}{ENV.schema}/" diff --git a/tests/pyathena/filesystem/test_s3_async.py b/tests/pyathena/filesystem/test_s3_async.py index b35eb1c2..5ce3c606 100644 --- a/tests/pyathena/filesystem/test_s3_async.py +++ b/tests/pyathena/filesystem/test_s3_async.py @@ -136,17 +136,16 @@ def test_parse_path_invalid(self): with pytest.raises(ValueError, match="Invalid S3 path format"): AioS3FileSystem.parse_path("s3://bucket?foo=bar") - with pytest.raises(ValueError, match="Invalid S3 path format"): - AioS3FileSystem.parse_path("s3://bucket/path/to/obj?foo=bar") - with pytest.raises(ValueError, match="Invalid S3 path format"): AioS3FileSystem.parse_path("s3a://bucket?") with pytest.raises(ValueError, match="Invalid S3 path format"): AioS3FileSystem.parse_path("s3a://bucket?foo=bar") - with pytest.raises(ValueError, match="Invalid S3 path format"): - AioS3FileSystem.parse_path("s3a://bucket/path/to/obj?foo=bar") + # GH-979: a "?" in a key that does not start a trailing version ID + # query is part of the key. + for path in ("s3://bucket/path/to/obj?foo=bar", "s3a://bucket/path/to/obj?foo=bar"): + assert AioS3FileSystem.parse_path(path) == ("bucket", "path/to/obj?foo=bar", None) @pytest.mark.parametrize("max_workers", [1, 4]) @pytest.mark.asyncio @@ -728,6 +727,32 @@ async def test_cp_file_directory(self): await fs._cp_file("s3://bucket/src", "s3://bucket/dst") fs._sync_fs._call.assert_not_called() + @pytest.mark.asyncio + async def test_copy_version(self): + # GH-979: a version is copied to a destination named after its key, + # not globbed with "?" as a wildcard. + fs = AioS3FileSystem(connection=mock.MagicMock(), skip_instance_cache=True) + fs._sync_fs.isdir = mock.MagicMock(return_value=False) + fs._cp_file = mock.AsyncMock() + + await fs._copy("s3://bucket/b?versionId=v1", "s3://bucket/d/", RequestPayer="requester") + fs._cp_file.assert_awaited_once_with( + "bucket/b?versionId=v1", "s3://bucket/d/b", RequestPayer="requester" + ) + + @pytest.mark.asyncio + async def test_get_version(self, tmp_path): + # GH-979: a version is downloaded to a local path named after its key. + fs = AioS3FileSystem(connection=mock.MagicMock(), skip_instance_cache=True) + fs._sync_fs.isdir = mock.MagicMock(return_value=False) + fs._get_file = mock.AsyncMock() + + await fs._get("s3://bucket/b?versionId=v1", f"{tmp_path}/") + assert fs._get_file.await_args.args[:2] == ( + "bucket/b?versionId=v1", + f"{tmp_path.as_posix()}/b", + ) + @pytest.mark.asyncio async def test_mv(self): # GH-1008: the files are copied in parallel, the directories are diff --git a/tests/pyathena/filesystem/test_s3_path.py b/tests/pyathena/filesystem/test_s3_path.py index 56908b2f..52578770 100644 --- a/tests/pyathena/filesystem/test_s3_path.py +++ b/tests/pyathena/filesystem/test_s3_path.py @@ -26,12 +26,20 @@ class TestS3Path: ("bucket/obj?versionid=v1", S3Path("bucket", "obj", "v1")), ("bucket/obj?version_id=v1", S3Path("bucket", "obj", "v1")), ("bucket?versionId=v1", S3Path("bucket", None, "v1")), + # Only a trailing version ID query is a version; any other "?" is + # part of the key. + ("bucket/dir/what?.txt", S3Path("bucket", "dir/what?.txt")), + ("bucket/obj?x=1", S3Path("bucket", "obj?x=1")), + ("bucket/a?b?versionId=v1", S3Path("bucket", "a?b", "v1")), + ("bucket/obj?versionId=", S3Path("bucket", "obj?versionId=")), + ("bucket/obj?versionId=a?versionId=b", S3Path("bucket", "obj?versionId=a", "b")), + ("bucket/?", S3Path("bucket", "?")), ], ) def test_parse(self, path, expected): assert S3Path.parse(path) == expected - @pytest.mark.parametrize("path", ["", "s3://", "http://bucket", "bucket/obj?x=1"]) + @pytest.mark.parametrize("path", ["", "s3://", "http://bucket", "bucket?x=1"]) def test_parse_invalid(self, path): with pytest.raises(ValueError, match="Invalid S3 path format"): S3Path.parse(path) @@ -70,6 +78,22 @@ def test_names(self, path, name, string, uri): def test_str_spells_the_version_query_as_version_id(self): assert str(S3Path.parse("s3a://bucket/key?version_id=v1")) == "bucket/key?versionId=v1" + @pytest.mark.parametrize( + ("path", "expected"), + [ + ("bucket/key", ("bucket/key", None)), + ("s3://bucket/key?versionId=v1", ("s3://bucket/key", "v1")), + ("bucket/dir/?version_id=v1", ("bucket/dir/", "v1")), + ("bucket/what?.txt", ("bucket/what?.txt", None)), + ("bucket/a?b?versionID=v1", ("bucket/a?b", "v1")), + # Not an S3 path, which parse() would reject. + ("", ("", None)), + ("name?versionid=v1", ("name", "v1")), + ], + ) + def test_split_version_id(self, path, expected): + assert S3Path.split_version_id(path) == expected + def test_with_version_id(self): path = S3Path("bucket", "key", "v1") assert path.with_version_id("v2") == S3Path("bucket", "key", "v2") From 43ab8f999116e4efdfdb829908cee0e2bd74e42e Mon Sep 17 00:00:00 2001 From: laughingman7743 Date: Sun, 4 Oct 2026 09:48:55 +0900 Subject: [PATCH 2/5] Fix regressions found in the review of version paths and "?" keys - S3Path.PATTERN matches keys with newlines again, also at the end, and reads "bucket/?versionId=..." as a version of the bucket path, as before. - get() checks that the destinations it names for version paths lie under lpath, as fsspec does for the destinations that it names, and accepts a PathLike destination; _has_version_id() accepts PathLike sources. - AioS3FileSystem._expand_path() keeps fsspec's async expansion for the paths without a version, whose glob lists only the keys under the stem. - A version path is kept by a recursive expansion only if it is a file, so a key prefix of the same name does not count as the version. Co-Authored-By: Claude Opus 5.5 --- pyathena/filesystem/s3.py | 73 ++++++++++++++++------ pyathena/filesystem/s3_async.py | 49 ++++++++++++--- pyathena/filesystem/s3_path.py | 9 ++- tests/pyathena/filesystem/test_s3.py | 48 ++++++++++++-- tests/pyathena/filesystem/test_s3_async.py | 25 ++++++++ tests/pyathena/filesystem/test_s3_path.py | 6 ++ 6 files changed, 176 insertions(+), 34 deletions(-) diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py index 0ee2afe8..c2e5b41f 100644 --- a/pyathena/filesystem/s3.py +++ b/pyathena/filesystem/s3.py @@ -69,7 +69,7 @@ _LOOKUPS_CACHE_KEY = "lookups" -def _has_version_id(path: str | list[str]) -> bool: +def _has_version_id(path: str | os.PathLike[str] | list[str]) -> bool: """Return whether a path, or a path of a list, ends with a version ID query. fsspec's ``copy()`` and ``get()`` take ``?`` for a glob character and @@ -82,8 +82,30 @@ def _has_version_id(path: str | list[str]) -> bool: Returns: Whether a path has a version ID query. """ - paths = [path] if isinstance(path, str) else path - return any(S3Path.split_version_id(str(p))[1] for p in paths) + paths = [path] if isinstance(path, (str, os.PathLike)) else path + return any(S3Path.split_version_id(os.fspath(p))[1] for p in paths) + + +def _check_contained(root: str, paths: list[str]) -> None: + """Raise if a local destination lies outside the destination root. + + The destinations are named after the source keys, whose ``..`` segments + would otherwise place a download above the root, as fsspec's ``get()`` + checks for the destinations that it names. + + Args: + root: The local destination root. + paths: The local destinations named below it. + + Raises: + ValueError: If a destination lies outside the root. + """ + root_key = os.path.normcase(os.path.abspath(root)) + prefix = root_key.rstrip(os.sep) + os.sep + for path in paths: + key = os.path.normcase(os.path.abspath(path)) + if key != root_key and not key.startswith(prefix): + raise ValueError(f"The destination {path!r} is outside {root!r}.") class CompressedBuffer(BytesIO): @@ -1098,7 +1120,7 @@ def expand_path(self, path, recursive=False, maxdepth=None, **kwargs) -> list[st As in fsspec, except that a path with a version ID names that version of an object: it is not a glob pattern, although its ``?`` is one in fsspec, nor is anything expanded below it. With ``recursive``, it is - included only if the version exists. + included only if it is a file, as fsspec includes a path that exists. Args: path: S3 path, glob pattern, or list of them. @@ -1115,13 +1137,8 @@ def expand_path(self, path, recursive=False, maxdepth=None, **kwargs) -> list[st """ if maxdepth is not None and maxdepth < 1: raise ValueError("maxdepth must be at least 1") - paths = [ - self._strip_protocol(p) - for p in ([path] if isinstance(path, (str, os.PathLike)) else path) - ] - versions = [p for p in paths if S3Path.split_version_id(p)[1]] - others = [p for p in paths if not S3Path.split_version_id(p)[1]] - out = {p for p in versions if not recursive or self.exists(p)} + versions, others = self._split_version_paths(path) + out = {p for p in versions if not recursive or self.isfile(p)} if others: try: out.update( @@ -1134,6 +1151,23 @@ def expand_path(self, path, recursive=False, maxdepth=None, **kwargs) -> list[st raise FileNotFoundError(path) return sorted(out) + def _split_version_paths(self, path) -> tuple[list[str], list[str]]: + """Split paths into those with a version ID and the others. + + Args: + path: S3 path, glob pattern, or list of them. + + Returns: + Tuple of the paths with a version ID and the other paths, without + the protocol. + """ + paths = [ + self._strip_protocol(p) + for p in ([path] if isinstance(path, (str, os.PathLike)) else path) + ] + versions = [p for p in paths if S3Path.split_version_id(p)[1]] + return versions, [p for p in paths if p not in versions] + def exists(self, path: str, **kwargs) -> bool: """Check if an S3 path exists. @@ -1773,7 +1807,8 @@ def get( As fsspec's ``get()``, except that a source with a version ID downloads that version of the object to a local path named after its - key, as ``_copy_paths`` pairs them. + key, as ``_copy_paths`` pairs them. Those destinations are checked to + lie under ``lpath``. Args: rpath: Source S3 path, glob pattern, or list of them. @@ -1783,15 +1818,17 @@ def get( callback: Progress callback. maxdepth: Maximum depth of a recursive copy. **kwargs: Additional parameters passed to ``get_file()``. + + Raises: + ValueError: If a source with a version ID is paired, and a + destination lies outside ``lpath``. """ - if isinstance(lpath, str) and _has_version_id(rpath): + if not isinstance(lpath, list) and _has_version_id(rpath): + root = make_path_posix(lpath) rpath, lpath = self._copy_paths( - rpath, - make_path_posix(lpath), - recursive=recursive, - maxdepth=maxdepth, - isdir=LocalFileSystem().isdir, + rpath, root, recursive=recursive, maxdepth=maxdepth, isdir=LocalFileSystem().isdir ) + _check_contained(root, lpath) if not rpath: return super().get( diff --git a/pyathena/filesystem/s3_async.py b/pyathena/filesystem/s3_async.py index 55122d88..277afcba 100644 --- a/pyathena/filesystem/s3_async.py +++ b/pyathena/filesystem/s3_async.py @@ -21,7 +21,13 @@ from fsspec.core import get_compression from fsspec.implementations.local import LocalFileSystem, make_path_posix -from pyathena.filesystem.s3 import CompressedBuffer, S3File, S3FileSystem, _has_version_id +from pyathena.filesystem.s3 import ( + CompressedBuffer, + S3File, + S3FileSystem, + _check_contained, + _has_version_id, +) from pyathena.filesystem.s3_executor import S3AioExecutor, S3Executor, S3ThreadPoolExecutor from pyathena.filesystem.s3_object import ( S3Metadata, @@ -450,16 +456,22 @@ async def _get( callback: Progress callback. maxdepth: Maximum depth of a recursive copy. **kwargs: Additional parameters passed to ``_get_file()``. + + Raises: + ValueError: If a source with a version ID is paired, and a + destination lies outside ``lpath``. """ - if isinstance(lpath, str) and _has_version_id(rpath): + if not isinstance(lpath, list) and _has_version_id(rpath): + root = make_path_posix(lpath) rpath, lpath = await asyncio.to_thread( self._sync_fs._copy_paths, rpath, - make_path_posix(lpath), + root, recursive=recursive, maxdepth=maxdepth, isdir=LocalFileSystem().isdir, ) + _check_contained(root, lpath) if not rpath: return await super()._get( @@ -734,20 +746,41 @@ async def _find( async def _expand_path(self, path, recursive=False, maxdepth=None, **kwargs) -> list[str]: """Expand glob patterns and directories into the paths they match. - See :meth:`S3FileSystem.expand_path`. + See :meth:`S3FileSystem.expand_path`. The other paths are expanded by + fsspec's ``_expand_path()``, whose glob lists only the keys under the + literal part of the pattern. Args: path: S3 path, glob pattern, or list of them. recursive: Whether to include the paths below the directories. maxdepth: Maximum depth of the expansion, at least 1. - **kwargs: Additional arguments passed to ``glob`` and ``find``. + **kwargs: Additional arguments passed to fsspec's + ``_expand_path()``, such as ``assume_literal``. Returns: The sorted matching paths. + + Raises: + ValueError: If ``maxdepth`` is less than 1. + FileNotFoundError: If nothing matches. """ - return await asyncio.to_thread( - self._sync_fs.expand_path, path, recursive=recursive, maxdepth=maxdepth, **kwargs - ) + if maxdepth is not None and maxdepth < 1: + raise ValueError("maxdepth must be at least 1") + versions, others = self._sync_fs._split_version_paths(path) + out = {p for p in versions if not recursive or await self._isfile(p)} + if others: + try: + out.update( + await super()._expand_path( + others, recursive=recursive, maxdepth=maxdepth, **kwargs + ) + ) + except FileNotFoundError: + if not out: + raise + if not out: + raise FileNotFoundError(path) + return sorted(out) def _create_executor(self, max_workers: int) -> S3Executor: """Create the executor for the parallel operations of a file. diff --git a/pyathena/filesystem/s3_path.py b/pyathena/filesystem/s3_path.py index 7d546df8..40f18f65 100644 --- a/pyathena/filesystem/s3_path.py +++ b/pyathena/filesystem/s3_path.py @@ -49,11 +49,14 @@ class S3Path: # Version IDs do not contain "?", so only the last query can be one. VERSION_QUERY: ClassVar[Pattern[str]] = re.compile( - r"\?version(Id|ID|id|_id)=(?P[^?]+)$" + r"\?version(Id|ID|id|_id)=(?P[^?]+)\Z" ) + # Keys may contain any character, including newlines. A bare "/" is tried + # before a key so that "bucket/?versionId=..." names the bucket. PATTERN: ClassVar[Pattern[str]] = re.compile( - r"(^s3://|^s3a://|^)(?P[a-zA-Z0-9.\-_]+)(/(?P.+?)|/)?" - rf"($|{VERSION_QUERY.pattern})" + r"(^s3://|^s3a://|^)(?P[a-zA-Z0-9.\-_]+)(/|/(?P.+?))?" + rf"(\Z|{VERSION_QUERY.pattern})", + re.DOTALL, ) bucket: str diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py index 135090dc..c5ce66d9 100644 --- a/tests/pyathena/filesystem/test_s3.py +++ b/tests/pyathena/filesystem/test_s3.py @@ -34,7 +34,7 @@ import pyathena from pyathena.filesystem import register_s3_filesystem -from pyathena.filesystem.s3 import CompressedBuffer, S3File, S3FileSystem +from pyathena.filesystem.s3 import CompressedBuffer, S3File, S3FileSystem, _has_version_id from pyathena.filesystem.s3_errors import S3ClientError from pyathena.filesystem.s3_executor import S3AioExecutor, S3ThreadPoolExecutor from pyathena.filesystem.s3_object import S3Object, S3ObjectType, S3StorageClass @@ -1700,13 +1700,17 @@ def test_expand_path_version(self, recursive): fs.expand_path("s3://bucket/b?versionId=v1", maxdepth=0) def test_expand_path_recursive_missing_version(self): - # With recursive, a version is included only if it exists, as fsspec - # does for the other paths. + # With recursive, a version is included only if it is a file, as + # fsspec includes a path that exists; a key prefix of the same name + # is not a version. fs = self._make_fs() - self._serve_keys(fs, set()) + self._serve_keys(fs, {"b", "dir/child"}) + assert fs.expand_path( + ["s3://bucket/b?versionId=v1", "s3://bucket/dir?versionId=v2"], recursive=True + ) == ["bucket/b?versionId=v1"] with pytest.raises(FileNotFoundError): - fs.expand_path("s3://bucket/b?versionId=v1", recursive=True) + fs.expand_path("s3://bucket/dir?versionId=v2", recursive=True) @pytest.mark.parametrize( ("path1", "path2", "expected"), @@ -1738,9 +1742,12 @@ def test_copy_version(self, method, path1, path2, expected): ("s3://bucket/key?versionId=v1", "f.txt", "f.txt"), ("s3://bucket/key?versionId=v1", "d/", "d/key"), (["s3://bucket/key?versionId=v1"], "d", "d/key"), + (Path("bucket/key?versionId=v1"), "d/", "d/key"), + ("s3://bucket/key?versionId=v1", Path("f.txt"), "f.txt"), ], ) def test_get_version(self, tmp_path, monkeypatch, rpath, lpath, expected): + # Path sources and destinations are accepted as fsspec accepts them. # GH-979: a version is downloaded to a local path named after its # key, which used to be the version-qualified name of the source. monkeypatch.chdir(tmp_path) @@ -1753,6 +1760,37 @@ def test_get_version(self, tmp_path, monkeypatch, rpath, lpath, expected): ) assert (tmp_path / expected).read_bytes() == b"data" + @pytest.mark.parametrize( + "rpath", + [ + "s3://bucket/a/..?versionId=v1", + # The sources without a version are named by the same pairing. + ["s3://bucket/x/..", "s3://bucket/key?versionId=v1"], + ], + ) + def test_get_version_outside_destination(self, tmp_path, rpath): + # A destination named after a key must stay under lpath, which + # fsspec checks for the destinations that it names. + (tmp_path / "d").mkdir() + fs, _ = self._make_object_fs(b"data") + + with pytest.raises(ValueError, match="outside"): + fs.get(rpath, f"{tmp_path}/d/") + assert sorted(p.name for p in tmp_path.rglob("*")) == ["d"] + + @pytest.mark.parametrize( + ("path", "expected"), + [ + ("bucket/key", False), + ("bucket/key?versionId=v1", True), + (Path("bucket/key"), False), + (Path("bucket/key?versionId=v1"), True), + (["bucket/a", Path("bucket/b?versionId=v1")], True), + ], + ) + def test_has_version_id(self, path, expected): + assert _has_version_id(path) is expected + def test_mv_nothing_within_maxdepth(self): # Only directories within maxdepth: nothing is moved, as with copy(). fs = self._make_fs() diff --git a/tests/pyathena/filesystem/test_s3_async.py b/tests/pyathena/filesystem/test_s3_async.py index 5ce3c606..ec9f5e63 100644 --- a/tests/pyathena/filesystem/test_s3_async.py +++ b/tests/pyathena/filesystem/test_s3_async.py @@ -740,6 +740,31 @@ async def test_copy_version(self): "bucket/b?versionId=v1", "s3://bucket/d/b", RequestPayer="requester" ) + @pytest.mark.asyncio + async def test_get_version_path_destination(self, tmp_path): + # A Path destination is paired too: the version is downloaded to the + # file, not into a directory of that name. + fs = AioS3FileSystem(connection=mock.MagicMock(), skip_instance_cache=True) + fs._sync_fs.isdir = mock.MagicMock(return_value=False) + fs._get_file = mock.AsyncMock() + + await fs._get("s3://bucket/b?versionId=v1", tmp_path / "out.bin") + assert fs._get_file.await_args.args[:2] == ( + "bucket/b?versionId=v1", + (tmp_path / "out.bin").as_posix(), + ) + + @pytest.mark.asyncio + async def test_expand_path_glob_lists_stem_prefix(self): + # Unversioned globs keep fsspec's async expansion, which lists only + # the keys that start with the stem before the first wildcard. + fs = AioS3FileSystem(connection=mock.MagicMock(), skip_instance_cache=True) + fs._sync_fs._find = mock.MagicMock(return_value=[]) + + with pytest.raises(FileNotFoundError): + await fs._expand_path("s3://bucket/reports/2026-*.csv") + assert fs._sync_fs._find.call_args.kwargs["prefix"] == "2026-" + @pytest.mark.asyncio async def test_get_version(self, tmp_path): # GH-979: a version is downloaded to a local path named after its key. diff --git a/tests/pyathena/filesystem/test_s3_path.py b/tests/pyathena/filesystem/test_s3_path.py index 52578770..09a46a6a 100644 --- a/tests/pyathena/filesystem/test_s3_path.py +++ b/tests/pyathena/filesystem/test_s3_path.py @@ -34,6 +34,12 @@ class TestS3Path: ("bucket/obj?versionId=", S3Path("bucket", "obj?versionId=")), ("bucket/obj?versionId=a?versionId=b", S3Path("bucket", "obj?versionId=a", "b")), ("bucket/?", S3Path("bucket", "?")), + # A version right after the bucket names a version of the bucket path. + ("bucket/?versionId=v1", S3Path("bucket", None, "v1")), + # Keys may contain newlines, also at the end. + ("bucket/a\nb", S3Path("bucket", "a\nb")), + ("bucket/a\n", S3Path("bucket", "a\n")), + ("bucket/a\n?versionId=v1", S3Path("bucket", "a\n", "v1")), ], ) def test_parse(self, path, expected): From a16f0630a05678ffd0c7207aee490c1de2df6497 Mon Sep 17 00:00:00 2001 From: laughingman7743 Date: Sun, 4 Oct 2026 09:57:18 +0900 Subject: [PATCH 3/5] Do not look up a missing version as a key prefix in info() A path with a version names an object, so info() raises FileNotFoundError when HeadObject does not find that version, instead of reporting a key prefix of the same name as a directory. exists() is False for it, and the ListObjectsV2 request is no longer sent. Co-Authored-By: Claude Opus 5.5 --- pyathena/filesystem/s3.py | 7 +++++-- tests/pyathena/filesystem/test_s3.py | 22 +++++++++++++++++++--- 2 files changed, 24 insertions(+), 5 deletions(-) diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py index c2e5b41f..8446a11b 100644 --- a/pyathena/filesystem/s3.py +++ b/pyathena/filesystem/s3.py @@ -818,10 +818,11 @@ def info(self, path: str, **kwargs) -> S3Object: missing, or find a key prefix, the cached listing of the parent is removed. With ``version_aware``, a cached file entry without a version ID is looked up again. With an explicit - version, the cached entries of the path are skipped, and the + version, the cached entries of the path are skipped, the HeadObject result is cached under the version-qualified path apart from other versions, except for the ``null`` version, which an - overwrite replaces. With request parameters on which the + overwrite replaces, and a missing version is not looked up as a key + prefix, since a version names an object. With request parameters on which the authorization of the requests depends (``ExpectedBucketOwner``, ``RequestPayer``, and the ``SSECustomer*`` parameters of an object encrypted with a customer-provided key), each request receives those @@ -906,6 +907,8 @@ def info(self, path: str, **kwargs) -> S3Object: ) if object_info: return object_info + if version_id: + raise FileNotFoundError(path) else: bucket_info = self._head_bucket(path, refresh=refresh, lookup_kwargs=lookup_kwargs) if bucket_info: diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py index c5ce66d9..1c151040 100644 --- a/tests/pyathena/filesystem/test_s3.py +++ b/tests/pyathena/filesystem/test_s3.py @@ -1683,6 +1683,23 @@ def test_invalidate_cache_question_mark_key(self): fs.invalidate_cache("s3://bucket/dir/what?.txt") assert list(fs.dircache) == ["bucket/dir/what"] + @pytest.mark.parametrize( + "kwargs", + [{"path": "s3://bucket/dir?versionId=v1"}, {"path": "s3://bucket/dir", "version_id": "v1"}], + ) + def test_info_missing_version_is_not_a_prefix(self, kwargs): + # GH-979: a version names an object, so a missing version is not + # found even if its key is a key prefix, which info() used to return + # as a directory. + fs = self._make_fs() + self._serve_keys(fs, {"dir/child"}) + + with pytest.raises(FileNotFoundError): + fs.info(**kwargs) + assert not fs.exists("s3://bucket/dir?versionId=v1") + methods = {c.args[0] for c in fs._call.call_args_list} + assert fs._client.list_objects_v2 not in methods + @pytest.mark.parametrize("recursive", [False, True]) def test_expand_path_version(self, recursive): # GH-979: "?" of a version ID query is not a glob character, and a @@ -3507,12 +3524,11 @@ def test_info_version_spellings_share_cache(self, lookup): # so a missing version evicts it for every spelling. fs = self._make_fs() kwargs = self.LOOKUP_KWARGS if lookup else {} + # A missing version is not looked up as a key prefix. fs._call.side_effect = [ {"ContentLength": 4, "ETag": '"etag"', "VersionId": "v1"}, FileNotFoundError("key"), - {"KeyCount": 0}, FileNotFoundError("key"), - {"KeyCount": 0}, ] assert fs.info("s3://bucket/key?versionId=v1", **kwargs).size == 4 @@ -3523,7 +3539,7 @@ def test_info_version_spellings_share_cache(self, lookup): fs.info("s3://bucket/key?version_id=v1", refresh=True, **kwargs) with pytest.raises(FileNotFoundError): fs.info("s3://bucket/key?versionId=v1", **kwargs) - assert fs._call.call_count == 5 + assert fs._call.call_count == 3 def test_info_does_not_cache_null_version(self): fs = self._make_fs() From b3b7d2f3a16d5c5d43111390c00b75cbb572e85e Mon Sep 17 00:00:00 2001 From: laughingman7743 Date: Sun, 4 Oct 2026 10:03:21 +0900 Subject: [PATCH 4/5] Raise failed lookups of version paths, and leave destination sequences to fsspec - A recursive expansion checks a version path with exists() again, which raises lookup errors other than FileNotFoundError; info() no longer takes a key prefix for a missing version, so the prefix case still excludes it. isfile() hid those errors and dropped the path. - get() and _get() pair version paths only for a str or PathLike lpath, so a tuple of destinations is mapped by fsspec as before. Co-Authored-By: Claude Opus 5.5 --- pyathena/filesystem/s3.py | 7 ++++--- pyathena/filesystem/s3_async.py | 4 ++-- tests/pyathena/filesystem/test_s3.py | 19 +++++++++++++++++++ 3 files changed, 25 insertions(+), 5 deletions(-) diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py index 8446a11b..2619a7f1 100644 --- a/pyathena/filesystem/s3.py +++ b/pyathena/filesystem/s3.py @@ -1123,7 +1123,8 @@ def expand_path(self, path, recursive=False, maxdepth=None, **kwargs) -> list[st As in fsspec, except that a path with a version ID names that version of an object: it is not a glob pattern, although its ``?`` is one in fsspec, nor is anything expanded below it. With ``recursive``, it is - included only if it is a file, as fsspec includes a path that exists. + included only if the version exists, as fsspec includes a path that + exists. Args: path: S3 path, glob pattern, or list of them. @@ -1141,7 +1142,7 @@ def expand_path(self, path, recursive=False, maxdepth=None, **kwargs) -> list[st if maxdepth is not None and maxdepth < 1: raise ValueError("maxdepth must be at least 1") versions, others = self._split_version_paths(path) - out = {p for p in versions if not recursive or self.isfile(p)} + out = {p for p in versions if not recursive or self.exists(p)} if others: try: out.update( @@ -1826,7 +1827,7 @@ def get( ValueError: If a source with a version ID is paired, and a destination lies outside ``lpath``. """ - if not isinstance(lpath, list) and _has_version_id(rpath): + if isinstance(lpath, (str, os.PathLike)) and _has_version_id(rpath): root = make_path_posix(lpath) rpath, lpath = self._copy_paths( rpath, root, recursive=recursive, maxdepth=maxdepth, isdir=LocalFileSystem().isdir diff --git a/pyathena/filesystem/s3_async.py b/pyathena/filesystem/s3_async.py index 277afcba..7b48da7c 100644 --- a/pyathena/filesystem/s3_async.py +++ b/pyathena/filesystem/s3_async.py @@ -461,7 +461,7 @@ async def _get( ValueError: If a source with a version ID is paired, and a destination lies outside ``lpath``. """ - if not isinstance(lpath, list) and _has_version_id(rpath): + if isinstance(lpath, (str, os.PathLike)) and _has_version_id(rpath): root = make_path_posix(lpath) rpath, lpath = await asyncio.to_thread( self._sync_fs._copy_paths, @@ -767,7 +767,7 @@ async def _expand_path(self, path, recursive=False, maxdepth=None, **kwargs) -> if maxdepth is not None and maxdepth < 1: raise ValueError("maxdepth must be at least 1") versions, others = self._sync_fs._split_version_paths(path) - out = {p for p in versions if not recursive or await self._isfile(p)} + out = {p for p in versions if not recursive or await self._exists(p)} if others: try: out.update( diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py index 1c151040..c3ddbfde 100644 --- a/tests/pyathena/filesystem/test_s3.py +++ b/tests/pyathena/filesystem/test_s3.py @@ -1716,6 +1716,23 @@ def test_expand_path_version(self, recursive): with pytest.raises(ValueError, match="maxdepth"): fs.expand_path("s3://bucket/b?versionId=v1", maxdepth=0) + def test_expand_path_recursive_version_lookup_error(self): + # A failed lookup is raised, not taken for a missing version. + fs = self._make_fs() + self._serve_keys(fs, {"a", "b"}) + serve = fs._call.side_effect + + def call(method, **kwargs): + if method is fs._client.head_object and kwargs["Key"] == "b": + raise PermissionError("b") + return serve(method, **kwargs) + + fs._call.side_effect = call + with pytest.raises(PermissionError): + fs.expand_path( + ["s3://bucket/a?versionId=v1", "s3://bucket/b?versionId=v2"], recursive=True + ) + def test_expand_path_recursive_missing_version(self): # With recursive, a version is included only if it is a file, as # fsspec includes a path that exists; a key prefix of the same name @@ -1761,6 +1778,8 @@ def test_copy_version(self, method, path1, path2, expected): (["s3://bucket/key?versionId=v1"], "d", "d/key"), (Path("bucket/key?versionId=v1"), "d/", "d/key"), ("s3://bucket/key?versionId=v1", Path("f.txt"), "f.txt"), + # A sequence of destinations is paired by fsspec, as before. + ("s3://bucket/key?versionId=v1", ("f.txt",), "f.txt"), ], ) def test_get_version(self, tmp_path, monkeypatch, rpath, lpath, expected): From 491e567d14706ed28db2de5b390dd12a26b1e877 Mon Sep 17 00:00:00 2001 From: laughingman7743 Date: Sun, 4 Oct 2026 10:32:15 +0900 Subject: [PATCH 5/5] Require fsspec>=2026.9.0 and move the version check to S3Path - S3Path.has_version_id() tells whether a path string ends with a version ID query; copy()/get() and their aio forms apply it to each source. - get() and _get() check the destinations that they name with fsspec.utils.check_contained(), which fsspec 2026.9.0 added, instead of a copy of it. The fsspec requirement is raised accordingly (4.0.0). - The "onerror" keyword that fsspec < 2026.6.0 leaked from mv() into cp_file is no longer removed, as the new requirement excludes it. Co-Authored-By: Claude Opus 5.5 --- pyathena/filesystem/s3.py | 58 +++-------------------- pyathena/filesystem/s3_async.py | 20 +++----- pyathena/filesystem/s3_path.py | 16 +++++++ pyproject.toml | 2 +- tests/pyathena/filesystem/test_s3.py | 15 +----- tests/pyathena/filesystem/test_s3_path.py | 15 ++++++ uv.lock | 2 +- 7 files changed, 48 insertions(+), 80 deletions(-) diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py index 2619a7f1..c8dc8f3e 100644 --- a/pyathena/filesystem/s3.py +++ b/pyathena/filesystem/s3.py @@ -31,7 +31,7 @@ from fsspec.core import get_compression from fsspec.implementations.local import LocalFileSystem, make_path_posix, trailing_sep from fsspec.spec import AbstractBufferedFile -from fsspec.utils import isfilelike, other_paths, tokenize +from fsspec.utils import check_contained, isfilelike, other_paths, tokenize import pyathena from pyathena.connection import Connection @@ -69,45 +69,6 @@ _LOOKUPS_CACHE_KEY = "lookups" -def _has_version_id(path: str | os.PathLike[str] | list[str]) -> bool: - """Return whether a path, or a path of a list, ends with a version ID query. - - fsspec's ``copy()`` and ``get()`` take ``?`` for a glob character and - name the destinations after the sources, so sources with a version ID are - paired with their destinations by ``S3FileSystem._copy_paths``. - - Args: - path: A path or a list of paths. - - Returns: - Whether a path has a version ID query. - """ - paths = [path] if isinstance(path, (str, os.PathLike)) else path - return any(S3Path.split_version_id(os.fspath(p))[1] for p in paths) - - -def _check_contained(root: str, paths: list[str]) -> None: - """Raise if a local destination lies outside the destination root. - - The destinations are named after the source keys, whose ``..`` segments - would otherwise place a download above the root, as fsspec's ``get()`` - checks for the destinations that it names. - - Args: - root: The local destination root. - paths: The local destinations named below it. - - Raises: - ValueError: If a destination lies outside the root. - """ - root_key = os.path.normcase(os.path.abspath(root)) - prefix = root_key.rstrip(os.sep) + os.sep - for path in paths: - key = os.path.normcase(os.path.abspath(path)) - if key != root_key and not key.startswith(prefix): - raise ValueError(f"The destination {path!r} is outside {root!r}.") - - class CompressedBuffer(BytesIO): """An in-memory buffer of data compressed with a codec of fsspec. @@ -1764,7 +1725,7 @@ def _copy_paths( paths1 = [p for p in paths1 if not (trailing_sep(p) or self.isdir(p))] if not paths1: return [], [] - glob = isinstance(path1, str) and has_magic(path1) and not _has_version_id(path1) + glob = isinstance(path1, str) and has_magic(path1) and not S3Path.has_version_id(path1) # The destination is looked up only when it decides the mapping. exists = source_is_str and ( (glob and len(paths1) == 1) @@ -1796,7 +1757,8 @@ def copy(self, path1, path2, recursive=False, maxdepth=None, on_error=None, **kw otherwise. **kwargs: Additional S3 copy parameters passed to ``cp_file()``. """ - if isinstance(path2, str) and _has_version_id(path1): + sources = [path1] if isinstance(path1, (str, os.PathLike)) else path1 + if isinstance(path2, str) and any(S3Path.has_version_id(p) for p in sources): path1, path2 = self._copy_paths(path1, path2, recursive=recursive, maxdepth=maxdepth) if not path1: return @@ -1827,12 +1789,13 @@ def get( ValueError: If a source with a version ID is paired, and a destination lies outside ``lpath``. """ - if isinstance(lpath, (str, os.PathLike)) and _has_version_id(rpath): + sources = [rpath] if isinstance(rpath, (str, os.PathLike)) else rpath + if isinstance(lpath, (str, os.PathLike)) and any(S3Path.has_version_id(p) for p in sources): root = make_path_posix(lpath) rpath, lpath = self._copy_paths( rpath, root, recursive=recursive, maxdepth=maxdepth, isdir=LocalFileSystem().isdir ) - _check_contained(root, lpath) + check_contained(root, lpath) if not rpath: return super().get( @@ -1906,13 +1869,6 @@ def _copy_file(self, path1: str, path2: str, **kwargs) -> bool: Raises: ValueError: If trying to copy to a versioned file or copy buckets. """ - # fsspec < 2026.6.0: AbstractFileSystem.mv() passed the typo'd - # "onerror" keyword (instead of "on_error", which copy() consumes), - # so it leaked through copy(**kwargs) into cp_file and must not - # reach the S3 API. Remove this once the fsspec requirement is - # >= 2026.6.0, where mv() passes on_error correctly. - # https://github.com/fsspec/filesystem_spec/commit/346a589fef9308550ffa3d0d510f2db67281bb05 - kwargs.pop("onerror", None) # Parameters of the multipart copy, not of the S3 requests. block_size = kwargs.pop("block_size", None) max_workers = kwargs.pop("max_workers", None) diff --git a/pyathena/filesystem/s3_async.py b/pyathena/filesystem/s3_async.py index 7b48da7c..ea107823 100644 --- a/pyathena/filesystem/s3_async.py +++ b/pyathena/filesystem/s3_async.py @@ -20,14 +20,9 @@ from fsspec.callbacks import _DEFAULT_CALLBACK from fsspec.core import get_compression from fsspec.implementations.local import LocalFileSystem, make_path_posix +from fsspec.utils import check_contained -from pyathena.filesystem.s3 import ( - CompressedBuffer, - S3File, - S3FileSystem, - _check_contained, - _has_version_id, -) +from pyathena.filesystem.s3 import CompressedBuffer, S3File, S3FileSystem from pyathena.filesystem.s3_executor import S3AioExecutor, S3Executor, S3ThreadPoolExecutor from pyathena.filesystem.s3_object import ( S3Metadata, @@ -424,7 +419,8 @@ async def _copy( batch_size: Number of copies to run at the same time. **kwargs: Additional S3 copy parameters passed to ``_cp_file()``. """ - if isinstance(path2, str) and _has_version_id(path1): + sources = [path1] if isinstance(path1, (str, os.PathLike)) else path1 + if isinstance(path2, str) and any(S3Path.has_version_id(p) for p in sources): path1, path2 = await asyncio.to_thread( self._sync_fs._copy_paths, path1, path2, recursive=recursive, maxdepth=maxdepth ) @@ -461,7 +457,8 @@ async def _get( ValueError: If a source with a version ID is paired, and a destination lies outside ``lpath``. """ - if isinstance(lpath, (str, os.PathLike)) and _has_version_id(rpath): + sources = [rpath] if isinstance(rpath, (str, os.PathLike)) else rpath + if isinstance(lpath, (str, os.PathLike)) and any(S3Path.has_version_id(p) for p in sources): root = make_path_posix(lpath) rpath, lpath = await asyncio.to_thread( self._sync_fs._copy_paths, @@ -471,7 +468,7 @@ async def _get( maxdepth=maxdepth, isdir=LocalFileSystem().isdir, ) - _check_contained(root, lpath) + check_contained(root, lpath) if not rpath: return await super()._get( @@ -509,9 +506,6 @@ async def _copy_file(self, path1: str, path2: str, **kwargs) -> bool: Raises: ValueError: If trying to copy to a versioned file or copy buckets. """ - # fsspec < 2026.6.0 leaks the typo'd "onerror" keyword from mv(); - # see S3FileSystem.cp_file. - kwargs.pop("onerror", None) # Parameters of the multipart copy, not of the S3 requests. block_size = kwargs.pop("block_size", None) max_workers = kwargs.pop("max_workers", None) diff --git a/pyathena/filesystem/s3_path.py b/pyathena/filesystem/s3_path.py index 40f18f65..6b2af357 100644 --- a/pyathena/filesystem/s3_path.py +++ b/pyathena/filesystem/s3_path.py @@ -9,6 +9,7 @@ from __future__ import annotations +import os import re from dataclasses import dataclass, replace from re import Pattern @@ -102,6 +103,21 @@ def split_version_id(cls, path: str) -> tuple[str, str | None]: return path, None return path[: match.start()], match.group("version_id") + @classmethod + def has_version_id(cls, path: str | os.PathLike[str]) -> bool: + """Return whether a path string ends with a version ID query. + + Unlike :meth:`parse`, any string is accepted, as with + :meth:`split_version_id`. + + Args: + path: The path. + + Returns: + Whether the path ends with a version ID query. + """ + return cls.split_version_id(os.fspath(path))[1] is not None + @property def is_bucket(self) -> bool: """Whether the path names the bucket: it has no key, or a key of only slashes.""" diff --git a/pyproject.toml b/pyproject.toml index 37465714..bddc691a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,7 @@ dependencies = [ "boto3>=1.43.31", "botocore>=1.43.31", "tenacity>=4.1.0", - "fsspec", + "fsspec>=2026.9.0", "python-dateutil", ] requires-python = ">=3.11" diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py index c3ddbfde..fbb40e6a 100644 --- a/tests/pyathena/filesystem/test_s3.py +++ b/tests/pyathena/filesystem/test_s3.py @@ -34,7 +34,7 @@ import pyathena from pyathena.filesystem import register_s3_filesystem -from pyathena.filesystem.s3 import CompressedBuffer, S3File, S3FileSystem, _has_version_id +from pyathena.filesystem.s3 import CompressedBuffer, S3File, S3FileSystem from pyathena.filesystem.s3_errors import S3ClientError from pyathena.filesystem.s3_executor import S3AioExecutor, S3ThreadPoolExecutor from pyathena.filesystem.s3_object import S3Object, S3ObjectType, S3StorageClass @@ -1814,19 +1814,6 @@ def test_get_version_outside_destination(self, tmp_path, rpath): fs.get(rpath, f"{tmp_path}/d/") assert sorted(p.name for p in tmp_path.rglob("*")) == ["d"] - @pytest.mark.parametrize( - ("path", "expected"), - [ - ("bucket/key", False), - ("bucket/key?versionId=v1", True), - (Path("bucket/key"), False), - (Path("bucket/key?versionId=v1"), True), - (["bucket/a", Path("bucket/b?versionId=v1")], True), - ], - ) - def test_has_version_id(self, path, expected): - assert _has_version_id(path) is expected - def test_mv_nothing_within_maxdepth(self): # Only directories within maxdepth: nothing is moved, as with copy(). fs = self._make_fs() diff --git a/tests/pyathena/filesystem/test_s3_path.py b/tests/pyathena/filesystem/test_s3_path.py index 09a46a6a..cf143f33 100644 --- a/tests/pyathena/filesystem/test_s3_path.py +++ b/tests/pyathena/filesystem/test_s3_path.py @@ -5,6 +5,8 @@ # # SPDX-License-Identifier: MIT +from pathlib import Path + import pytest from pyathena.filesystem.s3_path import S3Path @@ -100,6 +102,19 @@ def test_str_spells_the_version_query_as_version_id(self): def test_split_version_id(self, path, expected): assert S3Path.split_version_id(path) == expected + @pytest.mark.parametrize( + ("path", "expected"), + [ + ("bucket/key", False), + ("bucket/key?versionId=v1", True), + ("bucket/what?.txt", False), + (Path("bucket/key"), False), + (Path("bucket/key?version_id=v1"), True), + ], + ) + def test_has_version_id(self, path, expected): + assert S3Path.has_version_id(path) is expected + def test_with_version_id(self): path = S3Path("bucket", "key", "v1") assert path.with_version_id("v2") == S3Path("bucket", "key", "v2") diff --git a/uv.lock b/uv.lock index f736824b..9651cca0 100644 --- a/uv.lock +++ b/uv.lock @@ -978,7 +978,7 @@ dev = [ requires-dist = [ { name = "boto3", specifier = ">=1.43.31" }, { name = "botocore", specifier = ">=1.43.31" }, - { name = "fsspec" }, + { name = "fsspec", specifier = ">=2026.9.0" }, { name = "pandas", marker = "extra == 'pandas'", specifier = ">=3.0.0" }, { name = "polars", marker = "extra == 'polars'", specifier = ">=1.39.0" }, { name = "pyarrow", marker = "extra == 'arrow'", specifier = ">=22.0.0" },