S3 filesystem¶
PyAthena ships its own fsspec-compatible
filesystem implementation for Amazon S3 (S3FileSystem), built on boto3, with an API
surface compatible with s3fs for users migrating from it.
The filesystem is used internally by the pandas, Polars, and S3FS result sets to read query results from S3, and can also be used independently for S3 file operations.
fsspec registration¶
Importing pyathena.pandas or pyathena.polars registers S3FileSystem as the fsspec
s3 / s3a protocols via pyathena.filesystem.register_s3_filesystem. This replaces
fsspec’s default lazy mapping of the s3 protocol to s3fs, which means
fsspec.filesystem("s3") returns PyAthena’s implementation and s3fs-specific settings
(such as the S3FS_LOGGING_LEVEL environment variable) have no effect.
A filesystem class that has already been registered explicitly is also overwritten, with a warning log, so that the replacement is diagnosable. To restore another implementation, re-register it afterwards:
import fsspec
import s3fs
import pyathena.pandas # Registers PyAthena's S3FileSystem.
fsspec.register_implementation("s3", s3fs.S3FileSystem, clobber=True)
Basic usage¶
The filesystem can be constructed from a PyAthena connection, whose S3 client it then uses (see “S3 client” in Usage), or directly with s3fs-compatible credential arguments:
from pyathena import connect
from pyathena.filesystem.s3 import S3FileSystem
fs = S3FileSystem(connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
region_name="us-west-2"))
# Or with direct credentials (s3fs-compatible arguments).
fs = S3FileSystem(key="YOUR_ACCESS_KEY", secret="YOUR_SECRET_KEY")
# Or with a named profile.
fs = S3FileSystem(profile="YOUR_PROFILE")
# Or anonymously for public buckets.
fs = S3FileSystem(anon=True)
Standard fsspec operations work as expected:
fs.ls("s3://YOUR_S3_BUCKET/path/to/")
fs.find("s3://YOUR_S3_BUCKET/path/to/")
fs.exists("s3://YOUR_S3_BUCKET/path/to/object")
fs.info("s3://YOUR_S3_BUCKET/path/to/object")
with fs.open("s3://YOUR_S3_BUCKET/path/to/object", "rb") as f:
data = f.read()
fs.pipe("s3://YOUR_S3_BUCKET/path/to/object", b"data")
fs.cat("s3://YOUR_S3_BUCKET/path/to/object")
fs.cp("s3://YOUR_S3_BUCKET/src", "s3://YOUR_S3_BUCKET/dst")
fs.rm("s3://YOUR_S3_BUCKET/path/to/", recursive=True)
Writes with pipe/pipe_file issue a single PutObject request for data up to the
block size (5 MiB by default); larger data is uploaded as a parallel multipart upload
through the buffered file path. Inside an
fsspec transaction,
writes are deferred until the transaction commits and are discarded on rollback.
With the compression argument (a codec of fsspec, or "infer" from the extension of
the path), pipe/pipe_file compress the data before uploading it, and the sizes in
this section apply to the compressed data.
A multipart write keeps at most max_workers parts of the written data in flight and
waits for one of them to finish before it uploads another. The parts take about
max_workers × the block size of memory, in addition to the buffered data of the file.
If one of these parts has failed when the write waits, or a part copied from the
existing object in an append has failed before the write uploads another part, the
write raises its error instead of uploading the rest of the data, and aborts the
multipart upload after the parts being uploaded or copied finish. A part that fails
later raises its error when the file is closed, or when the transaction commits, and
the multipart upload is aborted.
max_workers is an argument of S3FileSystem, AioS3FileSystem, open, put, and
pipe.
The block size for writing, given by the block_size argument of open or by the
filesystem’s default_block_size, must be between 5 MiB and 5 GiB, inclusive, the part
size limits of a multipart upload. Otherwise, open raises ValueError.
A multipart upload consists of at most 10,000 parts. put and pipe upload one part
per block, so with the default block size they can upload up to about 48.8 GiB
(10,000 × 5 MiB). To upload a larger object, use a block size of at least its size
divided by 10,000, either with the block_size argument of put, pipe, and open
or with the default_block_size argument of S3FileSystem. put and pipe check
the size before uploading anything and raise ValueError with the minimum block size
if the data needs more parts. A file written with open can take more parts, because
each write that fills the buffer uploads the data beyond its last full block as a
separate part when that data is at least 5 MiB. In an append, the parts copied from
the existing object also count toward the limit. A write with open that reaches the
limit raises ValueError and aborts its multipart upload. Multipart copies with cp
use parts large enough to stay within the limit.
cp copies an object larger than 5 GiB with a multipart upload instead of a single
CopyObject request, with the same result as CopyObject. If HeadObject reports a size
of at most 5 GiB when the copy starts, for example because the size that cp found
came from a cached listing, the object is copied with CopyObject instead. CopyObject parameters given as
keyword arguments are sent to the multipart requests that accept them, such as
CopySourceIfMatch to each part copy. With the default COPY value of
MetadataDirective, TaggingDirective, and AnnotationDirective, the content headers
(such as ContentType) and user-defined metadata, the tags, and the annotations of the
source are copied, and the values given for them are ignored, as CopyObject does. A
REPLACE directive uses the given values instead, and AnnotationDirective="EXCLUDE"
skips the annotations. Copying the tags needs s3:GetObjectTagging on the source, and
copying the annotations needs s3:ListObjectAnnotations and s3:GetObjectAnnotation on
the source and s3:PutObjectAnnotation on the destination. A source without a
?versionId= suffix whose HeadObject reports a version ID other than null is copied
from that version, which needs s3:GetObjectVersion and, to copy the tags,
s3:GetObjectVersionTagging on the source. A null version is not pinned. The annotations are listed before anything is written and copied after the
upload completes, so the destination exists without them until the last one is
written. If an annotation fails to copy, the error is raised and the destination is
kept. A failed part copy aborts the multipart upload. So does an interrupt, or the
cancellation of an AioS3FileSystem copy, unless the upload has already completed. A
CreateMultipartUpload request and the part copies in flight finish first, and so does
the CompleteMultipartUpload request of an AioS3FileSystem copy. A further interrupt
while the CreateMultipartUpload request is waited for is ignored. One while the part
copies are waited for or the upload is aborted stops the cleanup, and the upload ID is
logged.
Paths are normalized as in fsspec, which drops a trailing slash, so info, isfile,
and open treat s3://YOUR_S3_BUCKET/dir/ as s3://YOUR_S3_BUCKET/dir: the object
dir if it exists, and otherwise the directory dir. An object whose key ends in a
slash, such as a folder marker, is therefore not a file for these methods. Opening
dir/ for reading reads the object dir or raises FileNotFoundError. Opening it for
writing, pipe, pipe_file, and put_file write the object dir. A path with a
?versionId= suffix keeps the slash and refers to the object. find, and ls of the directory, list the object as a file
entry. cat_file uses the key as written. Without a ?versionId= suffix, it reads
such an object without a range, with a non-empty range of non-negative offsets, or with
a negative start and no end, and raises FileNotFoundError for other ranges.
S3 request parameters, such as ContentType, ServerSideEncryption, or RequestPayer,
can be given to open, pipe, and put as keyword arguments or in
s3_additional_kwargs, and to all of them through the s3_additional_kwargs argument
of S3FileSystem. The parameters of a call take precedence over those of the
filesystem. A file sends each of its requests only the parameters that the S3 operation
accepts, so, for example, ServerSideEncryption for writes is not sent with reads. A
pipe of data up to the block size sends its parameters with a single PutObject
request as given. put sets ContentType from the file extension unless the call or
the filesystem gives one.
fs = S3FileSystem(s3_additional_kwargs={"ServerSideEncryption": "AES256"})
with fs.open("s3://YOUR_S3_BUCKET/path/to/data.csv", "wb", ContentType="text/csv") as f:
f.write(b"col1\n1\n")
info and exists accept ExpectedBucketOwner, RequestPayer, and the
SSECustomer* parameters of an object encrypted with a customer-provided key (SSE-C)
as keyword arguments, and ignore other request parameters. The lookups of the object
by open, and the existence check of pipe with mode="create", send these
parameters of the file or the write. A lookup with them uses only the cached results of
lookups with the same values, not cached listings.
sse_c = {"SSECustomerAlgorithm": "AES256", "SSECustomerKey": YOUR_32_BYTE_KEY}
with fs.open("s3://YOUR_S3_BUCKET/path/to/encrypted.csv", "rb", **sse_c) as f:
data = f.read()
Read caching¶
A file opened for reading fetches its data through an fsspec cache, chosen by the
cache_type argument of open or by the default_cache_type argument of
S3FileSystem, and "bytes" by default. The value is the name of a cache registered in
fsspec.caching.caches, so the available names depend on the installed fsspec version.
The common ones are:
"none": keeps nothing; each read requests the bytes it reads."bytes": reads one block beyond each read and keeps one contiguous range, which later reads extend."readahead": reads one block beyond each read and keeps only the last range read, which suits many small sequential reads."blockcache": requests whole blocks and keeps the most recently used ones, up tomaxblocks(32 by default)."background": like"blockcache", and also reads the next block in a thread while the current one is consumed.
The block size is the block_size argument of open or the filesystem’s
default_block_size (5 MiB by default). Options of the cache, such as maxblocks, are
given in the cache_options argument of open. See
Read Buffering
in the fsspec documentation for the other cache types.
fs = S3FileSystem(default_cache_type="readahead")
with fs.open(
"s3://YOUR_S3_BUCKET/path/to/data.csv",
"rb",
block_size=32 * 2**20,
cache_type="background",
cache_options={"maxblocks": 2},
) as f:
data = f.read()
The cache_type option of PandasCursor and PolarsCursor sets the default_cache_type
of the filesystem that reads their results; see S3 read options and
S3 read options. For the cache that S3FSCursor uses, see
S3 read options.
Error translation¶
S3 error responses are translated into standard Python exceptions, so filesystem
operations raise natural errors instead of botocore’s ClientError:
S3 error |
Python exception |
|---|---|
|
|
|
|
|
|
|
|
|
|
Others |
|
Multipart upload management¶
Incomplete multipart uploads continue to accrue storage costs until they are completed or aborted. The filesystem can discover and abort them:
uploads = fs.list_multipart_uploads("s3://YOUR_S3_BUCKET")
for upload in uploads:
print(upload.key, upload.upload_id, upload.initiated)
# Abort all incomplete uploads to a key and the keys under it.
fs.clear_multipart_uploads("s3://YOUR_S3_BUCKET/path/to/")
Versioning¶
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. 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.
fs = S3FileSystem(
connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/", region_name="us-west-2"),
version_aware=True,
)
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. Each version is named
# "bucket/key?versionId=<id>", 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")
for version in versions:
print(version.version_id, version.is_latest, version.last_modified)
Version-aware operations require the s3:GetObjectVersion and
s3:ListBucketVersions permissions.
Moving a ?versionId=null source onto its own key copies that version and
deletes the source version when bucket versioning is enabled. If versioning has
never been enabled or is suspended, the move leaves the source in place: copying
onto the key would replace the null version that the subsequent deletion removes.
The same distinction applies to move conflict checks.
A move compares a null version with its unversioned key using GetBucketVersioning,
which requires s3:GetBucketVersioning. Each relevant bucket is looked up once
during planning; other moves do not make this request. The result is not cached
between moves. A failed lookup stops the move before any copy or deletion.
Directory buckets do not support versioning, so their null versions are treated
as the key itself without a versioning lookup.
The opt-in S3 versioning integration tests create dedicated temporary buckets and check unversioned, enabled, and suspended states. See Testing for the command and required test permissions.
Bucket lifecycle¶
Bucket creation and deletion are infrastructure-level changes and are disabled by
default: mkdir/makedirs and rmdir raise PermissionError when they would
create or delete a bucket. Pass the opt-in flags to enable them:
fs = S3FileSystem(
connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/", region_name="us-west-2"),
allow_bucket_creation=True,
allow_bucket_deletion=True,
)
fs.mkdir("s3://YOUR_NEW_BUCKET")
fs.rmdir("s3://YOUR_NEW_BUCKET") # The bucket must be empty.
Creating a key prefix under an existing bucket requires no operation (S3 has no real directories below the bucket level) and is always a no-op.
Typed S3 operations¶
Parse an S3 path¶
Use S3Path.parse() to obtain the bucket, key and version ID of a path:
from pyathena.filesystem.s3_path import S3Path
path = S3Path.parse("s3://bucket/key?versionId=v1")
print(path.bucket, path.key, path.version_id)
In 4.0, S3FileSystem.PATTERN_PATH and the parse_path() methods of
S3FileSystem and AioS3FileSystem are removed.
S3Path.parse() returns an S3Path object; access its attributes instead of
unpacking the tuple returned by parse_path().
For direct access to the regular expression, use S3Path.PATTERN instead of
S3FileSystem.PATTERN_PATH.
Send S3 requests¶
S3FileSystem.core is an S3Core, the typed operations that the filesystem sends
its listing, lookup, read, write, delete, multipart upload, copy, tagging, ACL and
metadata replacement requests with. It can also be built on a boto3 S3 client. Each
operation sends one request (one per page for the iterators and
list_object_annotations()); plan_multipart_copy() and copy_object_annotation(),
described below, send several, and generate_presigned_url() signs a URL locally
without a request. The requests are sent with the retry policy. An operation raises
FileNotFoundError for a missing bucket or multipart upload, or for a missing object or
version that it reads, and caches nothing. Requests sent through fs.core do not
invalidate the filesystem’s cache: call
fs.invalidate_cache() after a change, or make it through the filesystem.
import boto3
from pyathena.filesystem.s3_core import S3Core
from pyathena.filesystem.s3_path import S3Path
core = fs.core # or S3Core(boto3.client("s3"))
metadata = core.head_object(S3Path.parse("s3://YOUR_S3_BUCKET/path/to/object"))
print(metadata.content_length, metadata.version_id)
for page in core.list_objects("YOUR_S3_BUCKET", prefix="path/to/", delimiter="/"):
print([o.key for o in page.objects], [p.prefix for p in page.common_prefixes])
list_object_versions(), list_multipart_uploads() and list_buckets() return pages in
the same way. The core sends the prefix of list_multipart_uploads() as given, and S3
matches it as a plain string prefix, so it also lists the uploads to sibling keys that
start with it, which fs.list_multipart_uploads() excludes.
get_object() reads an object, a version of it, or a byte range and returns the bytes.
A range has an exclusive end; None as the end reads to the end of the object, and a
negative start without an end reads the last bytes. The response body is closed after
the read; a failure to read it raises the botocore exception and is not retried.
put_object() writes bytes and returns an S3PutObject.
data = core.get_object(S3Path.parse("s3://YOUR_S3_BUCKET/path/to/object"), (0, 1024))
result = core.put_object(S3Path("YOUR_S3_BUCKET", "path/to/copy"), data)
print(result.etag, result.version_id)
delete_objects() deletes an S3DeleteBatch, the objects of one bucket that one
DeleteObjects request accepts. S3DeleteBatch.from_paths() groups paths into batches
of up to 1,000 objects per bucket. The objects that S3 could not delete are in the
errors of the returned S3DeleteResult, not raised; as in S3, deleting a key that
does not exist is not an error.
from pyathena.filesystem.s3_core import S3DeleteBatch
paths = [
S3Path.parse("s3://YOUR_S3_BUCKET/path/to/a"),
S3Path.parse("s3://YOUR_S3_BUCKET/path/to/b"),
]
for batch in S3DeleteBatch.from_paths(paths):
result = core.delete_objects(batch)
for error in result.errors:
print(error) # path (code: message)
get_object_tagging() returns the tags of an object as a dictionary, and
put_object_tagging() replaces all of its tags with the given ones. Both act on the
version ID of the path, if any, including null. put_object_acl() and
put_bucket_acl() apply a canned ACL, which must be in S3Core.OBJECT_ACLS or
S3Core.BUCKET_ACLS, respectively; another value raises ValueError, and an object
ACL also applies to the version ID of the path.
replace_object_metadata() replaces the user-defined metadata of an object, given the
head_object() result of the same path, by copying the object onto itself. The copy
rewrites the object, or creates a new version in a versioned bucket, so the path must
not have a version ID. The copy retains the content headers (CacheControl,
ContentDisposition, ContentEncoding, ContentLanguage, ContentType), Expires,
WebsiteRedirectLocation, StorageClass, and the ServerSideEncryption,
SSEKMSKeyId and BucketKeyEnabled of the object. Additional CopyObject parameters
take precedence over the retained fields, and any encryption parameter replaces all of
the retained encryption fields.
path = S3Path.parse("s3://YOUR_S3_BUCKET/path/to/object")
core.put_object_tagging(path, {**core.get_object_tagging(path), "tag2": "value2"})
head = core.head_object(path)
core.replace_object_metadata(path, head, {**head, "attr1": "value1"})
generate_presigned_url() signs a URL locally, for get_object unless another client
method is given, and sends no request. Its parameters take precedence over the
Bucket, Key and VersionId of the path. The request_kwargs of the core, such as
RequestPayer, are not included in the signed parameters; pass them as parameters
of generate_presigned_url() to sign them.
url = core.generate_presigned_url(path, expires_in=600)
upload_url = core.generate_presigned_url(path, "put_object", ContentType="text/csv")
create_bucket() and delete_bucket() create and delete a bucket when called. The
allow_bucket_creation and allow_bucket_deletion options apply only to the
filesystem’s mkdir/makedirs and rmdir, not to calls through the core.
get_bucket_versioning() returns the versioning state of a bucket, Enabled or
Suspended, or None if versioning has never been enabled on it.
S3Path.is_directory_bucket tells whether the bucket of a path is a directory bucket.
create_bucket() sends a LocationConstraint for the region_name argument, or for
the client’s region by default, except in us-east-1. The fields of a
CreateBucketConfiguration passed as a parameter are kept, and the LocationConstraint
is added to it only if it has none of LocationConstraint, Location and Bucket; an
empty or None configuration is treated as not passed. To
create a directory bucket, pass its Location and Bucket in
CreateBucketConfiguration; no LocationConstraint is added.
Multipart writer¶
S3MultipartWriter provides synchronous multipart requests and part planning without fsspec.
It is the writer used by S3File; applications can also construct it on an S3Core.
The constructor validates the destination and block size without sending a request.
The caller owns the buffer and executor, collects part results in number order, and waits for running requests before completing or aborting.
Serialize initiation and finalization; part requests can run in parallel after initiation.
from io import BytesIO
from pyathena.filesystem.s3_writer import S3MultipartWriter
writer = S3MultipartWriter(
core,
S3Path("YOUR_S3_BUCKET", "path/to/object"),
block_size=5 * 2**20,
request_kwargs={"ChecksumAlgorithm": "SHA256"},
)
writer.initiate()
try:
parts = [
writer.upload_part(number, body)
for number, body in writer.iter_parts(BytesIO(b"your data"))
]
result = writer.complete(parts)
except BaseException:
writer.abort()
raise
iter_parts() consumes a blocking binary stream from its current position and produces numbered part bodies without sending requests.
It merges a short tail into the preceding block and splits the result if it reaches the maximum part size.
A stream shorter than the minimum part size must be the final part.
iter_copy_parts(size) produces numbered source ranges for copying an existing object before appending data.
A source smaller than the minimum part size must be the final part; S3File reads a small existing object into its buffer when appending.
Both planners accept first_part_number to include previously copied or uploaded parts in the 10,000-part limit.
upload_part_copy() sends a planned range with an exclusive end; None sends no range of its own, as with the core primitive.
The writer filters inherited and per-request parameters by operation; per-request parameters take precedence.
It retains the upload identity after completion and after an abort failure, and clears it only after a successful abort.
The caller can therefore retry a failed abort.
The writer does not invalidate the filesystem cache; call fs.invalidate_cache() after a direct write.
When an S3File wait for multipart creation is interrupted, it waits for any creation that has already started and recovers the successful response before aborting.
It drops the buffer and closes the file before waiting, so a later close or deferred commit cannot upload the interrupted write.
It suppresses Ctrl-C during the blocking recovery wait and re-raises the original error when cleanup finishes.
An additional interruption outside that wait can stop cleanup; once the upload identity is known, discard() can retry the abort.
An abort failure retains the upload identity for discard() to retry.
Cancelling an async buffered write through asyncio.to_thread still allows its writer thread to finish.
Multipart requests¶
create_multipart_upload(), upload_part(), upload_part_copy(),
complete_multipart_upload() and abort_multipart_upload() send the requests of a
multipart upload. part_ranges() sends no request: it splits an object into the byte
ranges of the parts that copy it, by the part limits MULTIPART_UPLOAD_MIN_PART_SIZE (5 MiB),
MULTIPART_UPLOAD_MAX_PART_SIZE (5 GiB) and MULTIPART_UPLOAD_MAX_PARTS (10,000) of
S3Core. check_multipart_upload_size() sends no request either: it raises
ValueError with the minimum block size if data of the given size takes more than
MULTIPART_UPLOAD_MAX_PARTS parts of the block size, the check that put and pipe
make before uploading.
copy_object() copies an object with one CopyObject request, which accepts objects
up to MULTIPART_UPLOAD_MAX_PART_SIZE. For a larger object, plan_multipart_copy()
reads the source and returns an S3MultipartCopyPlan: the version to copy, the byte
ranges of the parts, the parameters of each multipart upload request, and the
annotations to copy, so that the multipart upload writes the metadata, tags and
annotations that CopyObject would. It sends HeadObject, then GetObjectTagging and
ListObjectAnnotations unless the directives or the source exclude them, and writes
nothing. If HeadObject reports a size that fits in one CopyObject request, nothing else
is read, and the plan’s fits_single_request says to copy with copy_object()
instead. copy_object_annotation() copies one annotation onto the destination after
the upload completes, with GetObjectAnnotation and PutObjectAnnotation. The
filesystems’ cp_file(), copy() and mv() run these plans.
The multipart primitives take the S3MultipartUpload returned by creation,
keeping its bucket, key, upload ID and checksum configuration together.
The core retains no upload state.
upload_part() passes the upload’s checksum algorithm to the SDK;
completion selects the matching part checksum and sends the upload’s
ChecksumType when present, including FULL_OBJECT.
Without a creation algorithm, completion sends the ETag and part number without
checksums that the SDK may add to part uploads.
abort_multipart_upload() also accepts uploads returned by
list_multipart_uploads().
upload = core.create_multipart_upload(
S3Path("YOUR_S3_BUCKET", "path/to/object"), ChecksumAlgorithm="SHA256"
)
try:
part = core.upload_part(upload, 1, b"data")
result = core.complete_multipart_upload(upload, [part])
except Exception:
core.abort_multipart_upload(upload)
raise
Path pairing¶
S3PathPairing pairs the paths of one copy(), get() or mv(): mv() always, and
copy() and get() when a source has a version ID and the destination is one path
(fsspec pairs the others). Its delete_paths() splits the paths of an rm() into
those deleted as given and those expanded. The pairing is fsspec’s, except that a
path with a version ID names that version, and its destination is named after its key.
A pairing is a frozen dataclass of path1, path2, recursive and maxdepth, and
holds no filesystem. The filesystems read from it what to look up (expands,
skips_directories, looks_up_destination and conflict_candidates()), expand the
sources, leave out the directories when skips_directories is true, look up the
paths, and pass the results to copy_pairs() and move_pairs(). A sources,
destination_is_dir or missing that a rule needs and that is not passed raises
ValueError.
For moves that compare a null version with its unversioned key, the caller also
looks up bucket versioning and passes the names of the versioning-enabled buckets
as versioning_enabled_buckets to both conflict_candidates() and move_pairs().
Pass a collection of bucket names, such as a set, rather than a single string.
The default empty collection treats null versions as their keys, as in unversioned
or suspended buckets. The model makes no AWS requests.
from pyathena.filesystem.s3_path_pairing import S3PathPairing
pairing = S3PathPairing("s3://YOUR_S3_BUCKET/src/", "s3://YOUR_S3_BUCKET/dst/", recursive=True)
sources = fs.expand_path(pairing.path1, recursive=pairing.recursive)
for source, destination in pairing.copy_pairs(sources):
print(source, "->", destination)
Async filesystem¶
AioS3FileSystem provides the same functionality on top of fsspec’s
AsyncFileSystem, running its blocking calls in thread pools sized by max_workers.
Its open_async() returns a file whose reads and writes are coroutines.
See AioS3FileSystem for details.