Reputation: 427
I need to upload a file and show its stats to the user (as it shows on the console).
Didn't find any library for this, the function can be as simple as showing the uploaded percentage (uploaded/filesize*100
) and showing the uploaded size (20/50MB) with the upload speed and ETA.
There's a nice library named alive-progress
on printing out the progress bar.
but I just don't have any idea on how to do this, like the simpler version of what is done on youtube-dl
.
Upvotes: 0
Views: 2806
Reputation: 11
The tqdm lib can do this.
file_size = os.path.getsize(file_path)
with open(file_path, 'rb') as f:
with tqdm(desc=f"[INFO] Uploading", total=file_size, unit="B", unit_scale=True, unit_divisor=1024) as t:
reader_wrapper = CallbackIOWrapper(t.update, f, "read")
resp = httpx.post(url, files={'apk_file': reader_wrapper})
it looks like:
[INFO] Uploading: 100%|██████████| 9.60M/9.60M [00:02<00:00, 3.85MB/s]
Upvotes: 1
Reputation: 632
The following code should run on Python 3.7
or later.
Just edit SRC_FILE
and DEST_URL
before copy and paste.
import aiohttp
import asyncio
import os
import aiofiles
import math
from tqdm import tqdm
async def file_sender(file_name=None, chunk_size=65536):
file_size = os.path.getsize(file_name)
chunks = max(1, int(math.ceil(file_size / chunk_size)))
progress = tqdm(desc=f"Uploading", total=file_size, unit="B", unit_scale=True, unit_divisor=1024)
async with aiofiles.open(file_name, 'rb') as f:
for _ in range(chunks):
chunk = await f.read(chunk_size)
progress.update(len(chunk))
yield chunk
async def async_http_upload_from_file(src, dst):
async with aiohttp.ClientSession() as session:
await session.post(dst, data=file_sender(file_name=src))
SRC_FILE = 'path/to/your/file'
DEST_URL = 'path/to/your/url'
asyncio.run(async_http_upload_from_file(SRC_FILE, DEST_URL))
I think you can try using this URL for testing:
DEST_URL = 'http://httpbin.org/post'
Upvotes: 0
Reputation: 407
There is another way. Combining \r
with print(text, end='')
.
I don't know how you're getting your uploaded_size
so there is a sample code that should work anyway.
import time
#Progress bar
def updateProgressBar(size_uploaded, size_file, size_bar=50):
perc_uploaded = round(size_uploaded / size_file * 100)
progress = round(perc_uploaded / 100 * size_bar)
status_bar = f"-{'▒' * progress}{' ' * (size_bar - progress)}-"
status_count = f"[{size_uploaded}/{size_file}MB]"
#add your status_eta
#add your status_speed
print(f"\r{status_bar} | {status_count} | {perc_uploaded}%", end='')
#using the carriage-return (\r) to "overwrite" the previous line
#For the demo
file_size = 2**16 #fake file size
uploaded_size = 0
while uploaded_size < file_size:
uploaded_size += 1
updateProgressBar(uploaded_size, file_size)
print("\nDone!")
time.sleep(10)
I suggest that each time you're getting an update about your uploaded_size/ETA/upload_speed you call the updateProgressBar
method.
Upvotes: 0
Reputation: 407
You could use the sys
lib.
Using stdout
& flush
(more details here).
import sys, time
lenght_bar = 50
sys.stdout.write("Loading : |%s|" % (" " * lenght_bar))
sys.stdout.write("\b" * (lenght_bar+1)) #use backspace
for i in range(lenght_bar):
sys.stdout.write("▒")
sys.stdout.flush()
time.sleep(0.1) #process
sys.stdout.write("| 100%")
sys.stdout.write("\nDone")
time.sleep(10)
Upvotes: 1