drivendataorg / drivendataorg/cloudpathlib

Add download/upload progress callback support

Open
#488 1 comment 1 reaction 0 assignees View on GitHub
design decision
Dominant language
Python
Stars
628
Forks
88
Avg merge
17h 28m
Merged PRs (30d)
2

Description

I've started using cloudpathlib recently and its made what I'm working on so much easier. I would like to report download progress to the users if any of my cloud files need downloaded.

I'm using S3 and got it working for my needs in my personal PR https://github.com/blurstudio-forks/cloudpathlib/pull/1. It's fairly easy to configure by passing a callable to the new `transfer_callable` argument of `S3Client`. The callable must accept the arguments `direction, state, cloud_path, bytes_sent)`. This uses the `S3Client` to enable callbacks for any `CloudPath` used if it happens to trigger download/uploads.

The first 3 arguments allow your callable to understand what is happening and update your UI as needed including keeping track of multiple transfers happening at the same time. It lets you know when a transfer starts and stops so you can show/hide UI.

A simple implementation example:
```py
from itertools import islice
from cloudpathlib import CloudPath, S3Client

def progress_callback(direction, state, cloud_path, bytes_sent):
print("Progress", direction, state, cloud_path, bytes_sent)

client = S3Client(transfer_callable=progress_callback, no_sign_request=True)
ladi = CloudPath("s3://ladi/Images/FEMA_CAP/2020/70349", client=client)
a_file = next(islice(ladi.iterdir(), 1))
# If file needs downloaded, this will call `progress_callback` to report start, stop and download progress
print(a_file.fspath)
```

I have only worked with S3, but I hope the other services also provide a transfer progress reporting system. Is this something you would like to implement or should I develop my code enough to create a pull request?

A more complex example using the rich library to show download progress of several files downloaded at once.

```py
import concurrent.futures
from functools import partial
from itertools import islice
from cloudpathlib import CloudPath, S3Client
from rich.progress import BarColumn, DownloadColumn, Progress, TransferSpeedColumn, TextColumn

callback_state: dict[CloudPath, dict] = {}

def progress_callback(progress, direction, state, cloud_path, bytes_sent):
if state == "start":
progress.start()
size = cloud_path.stat().st_size
task_id = progress.add_task("Downloading", total=size, filename=cloud_path.name)
callback_state[cloud_path] = dict(task_id=task_id, size=size)
elif state == "stop":
if cloud_path in callback_state:
del callback_state[cloud_path]
else:
info = callback_state[cloud_path]
progress.update(info["task_id"], advance=bytes_sent)

def download_url(cloud_path):
# Trigger download of non-cached files
cloud_path.fspath

def download(urls) -> None:
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
future_to_url = {}
for url in urls:
future_to_url[executor.submit(download_url, url)] = url

for future in concurrent.futures.as_completed(future_to_url):
try:
future.result()
except Exception as exc:
print('%r generated an exception: %s' % (future_to_url[future], exc))
progress.stop()

if __name__ == "__main__":
# Set up the progress bar and attach the callback to the S3Client
progress = Progress(
TextColumn("[blue]{task.fields[filename]}"),
BarColumn(),
DownloadColumn(),
TransferSpeedColumn(),
)
client = S3Client(
transfer_callable=partial(progress_callback, progress),
no_sign_request=True,
)

# Download the first first 5 files from this resource.
ladi = CloudPath("s3://ladi/Images/FEMA_CAP/2020/70349", client=client)
download(islice(ladi.iterdir(), 5))
```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.