Add typed listing entries and core list/HEAD operations (step 2 of #1053) #1059

Use case

This is step 2 of #1053. The design decisions are recorded in #1053 (comment): the core is public API, the core is synchronous (no aiobotocore), and the cache stays in fsspec's DirCache.

Today every listing and lookup request is made inside the fsspec-shaped methods of S3FileSystem, which turn the responses straight into fsspec entries (S3Object, a MutableMapping with name/type/size). Line numbers are from pyathena/filesystem/s3.py on master 6d11865.

• ListObjectsV2: _ls_dirs() (:578-656) pages through the listing and builds S3Objects. Directory entries are synthesized from CommonPrefixes by _directory_object() (:366-381), with the trailing slash stripped to match fsspec names. info() sends its own MaxKeys=1 request to probe for a key prefix (:879-900).
• ListObjectVersions: _list_object_versions_pages() (:735-763) yields raw pages. _ls_object_versions() (:702-733) turns them into S3Objects with _versioned_file_object() (:383-406). object_version_info() (:3107-) turns the same pages into the typed S3ObjectVersion.
• ListBuckets: _ls_buckets() (:543-576) builds directory S3Objects from one response. It ignores the ContinuationToken that ListBuckets returns since botocore added its pagination.
• HeadObject: _head_object() (:464-541) builds an S3Object and caches it. metadata() (:2812-) sends its own request and returns the typed S3Metadata, and so does the multipart copy (:2140-2160).
• HeadBucket: _head_bucket() (:408-462) returns a synthesized directory S3Object.

The same S3 response is therefore parsed in several places into different shapes, and the request logic (pagination, operation parameters, error translation) cannot be used or tested apart from fsspec's entry and cache conventions.

Proposed change

Add the first operations of the public core, pyathena.filesystem.s3_core.S3Core, together with typed listing entries. S3FileSystem keeps building its fsspec entries, names and cache from the core's results.

No behavior change: the requests, return values, exceptions and cache keys of S3FileSystem/AioS3FileSystem stay as they are. The one exception is listing every bucket page, which _ls_buckets() should do (see below).

S3Core

S3Core(client: BaseClient, retry_config: RetryConfig | None = None, request_kwargs: Mapping[str, Any] | None = None)

• The core never builds a client. S3FileSystem.__init__() keeps building it, in either of its two ways (from the connection, or from s3fs-compatible arguments), and passes it in.
• client is a public read-only property, so Stubber(core.client) works.
• call(method: str, **request) is the core's request function: the retry policy (retry_api_call) and the error translation (S3ClientError.os_error: 404/NoSuchKey/NoSuchBucket/NoSuchVersion → FileNotFoundError, 403 → PermissionError, others → OSError).
• request_kwargs are constructor-level parameters, such as RequestPayer, sent with every operation that accepts them. They are filtered per operation by the botocore input shape, as _call() filters request_kwargs today. Per-call parameters are passed as given, so botocore validates them and a mistyped name is not silently dropped. The filter is public (for example operation_params(method, params)), so that the adapter can use it for its inherited lookup parameters and S3File._get_request_kwargs().
• s3_additional_kwargs stays in the adapter. It is documented for object requests only.

Operations (synchronous, one request per call, no cache):

• head_object(path: S3Path, **params) -> S3Metadata.
• head_bucket(bucket: str, **params).
• One page of ListObjectsV2 → S3ListObjectsPage, and an iterator over the pages.
• One page of ListObjectVersions → S3ListObjectVersionsPage, and an iterator over the pages.
• One page of ListBuckets, and an iterator over the pages.

A missing object or bucket raises FileNotFoundError, which is also fsspec's contract. The adapter converts it to None where its helpers need that, and exists() keeps answering True for 403 in the adapter.

The iterators yield pages, not flattened entries. This keeps the adapter's current rules: an empty listing evicts the cache, and object_version_info() stops when IsTruncated is false or there is no NextKeyMarker, with NextVersionIdMarker defaulting to "". Continuation tokens are sequential, so pages are not fetched in parallel. The page-level operations serve info()'s MaxKeys=1 probe and _ls_dirs(next_token=, max_keys=). The aio adapter keeps calling the sync filesystem in threads.

Entries (in s3_core.py)

New entries are frozen dataclasses with a from_response(...) classmethod, following S3Path. Later steps follow the same convention. Existing classes keep their form.

• S3ObjectSummary: a ListObjectsV2 Contents entry (bucket, key, size, etag, last_modified, storage_class, and path as an S3Path). It is the same concept as boto3's ObjectSummary, and distinct from the fsspec entry S3Object.
• S3CommonPrefix: bucket and prefix, with the prefix keeping its trailing delimiter.
• S3Bucket: name, creation_date, and bucket_region when returned.
• S3ListObjectsPage (objects, common_prefixes, is_truncated, next_continuation_token) and S3ListObjectVersionsPage (versions, delete_markers, common_prefixes, is_truncated, next_key_marker, next_version_id_marker). There is one type per operation, because the two continuation tokens differ in shape. The ListBuckets page has buckets and continuation_token.

Existing classes are reused:

• S3ObjectVersion for the Versions and DeleteMarkers entries.
• S3Metadata for HeadObject. It gains the fields that info() returns today but S3Metadata lacks: object_lock_mode, object_lock_retain_until_date, object_lock_legal_hold_status, and the path (S3Path) of the object. Without them, building the adapter's S3Object from S3Metadata would drop those fields.

Adapter changes

• S3FileSystem holds the core as a public read-only attribute, so that core users can reuse the filesystem's client and credentials. fs._client and fs._retry_config read from the core, so Stubber(fs._client) keeps working.
• S3FileSystem._call(method: str | Callable, **kwargs) stays as a thin delegate to the core. S3File and most tests use it.
• _ls_dirs(), _ls_buckets(), _ls_object_versions(), _head_object(), _head_bucket(), info()'s prefix probe, object_version_info(), metadata() and the multipart copy's HEAD use the core.
• One adapter function turns the core's entries into S3Objects, replacing the inline construction and _versioned_file_object(). _directory_object() stays for synthesized entries. The DirCache layout and invalidation do not change.
• The tests that mock fs._call for the list and HEAD paths move to mocking the core, or to botocore's Stubber (test_s3.py already uses it). The other tests keep their fs._call mocks.
• _ls_buckets() follows the ListBuckets continuation token, so accounts with more buckets than one page are listed in full. This is the one intended behavior change, and it goes in the release notes.

Out of scope: find()'s directory synthesis and depth semantics, copy/move/delete pairing, the multipart writer (a separate class such as S3MultipartWriter that takes an S3Core), and the cache. These are later steps of #1053.

Validation plan (if implementing)

• Unit tests for each entry's from_response(), built from the botocore response shapes and including missing optional fields.
• Unit tests for each S3Core operation with botocore's Stubber: pagination tokens (including ListBuckets), per-operation filtering of request_kwargs with per-call parameters passed as given, and error translation for 404, 403 and other errors.
• The existing offline and S3 integration tests in tests/pyathena/filesystem/ pass. They are unchanged, except where list/HEAD request mocks move. The request sequences of ls, info, exists, find, object_version_info and metadata (sync and aio) stay as the existing request-recording tests assert, and info() returns the ObjectLock fields as before.