Cum pentru a imprima "puncte" (sau alt fel de feedback-ul) în timp ce scrierea unui fișier în python?

requests.get în mod implicit popularitate totalitatea resursa solicitată înainte de a ajunge înapoi la tine. Cu toate acestea, ea are un argument opțional stream, care vă permite să invoce .iter_content sau .iter_lines pe Response obiect. Acest lucru vă permite să ia măsuri în fiecare N bytes (sau ca în fiecare bucată de date ajunge), sau la fiecare linie, respectiv. Ceva de genul asta:

chunks = []
chunk_size = 16384     # 16Kb chunks
# alternately
# chunk_size = None    # whenever a chunk arrives
res = requests.get(url_to_file, stream=True)
for chunk in res.iter_content(chunk_size):
    chunks.append(chunk)
    print(".", end="")
data = b''.join(chunks)

Acest lucru încă blocuri, deși, astfel încât nimic altceva nu se va întâmpla. Daca vrei mai mult de JavaScript stil, pe Grismar comentariu, tu ar trebui să ruleze sub Python asincron în buclă. În acest caz, am sugerăm să utilizați aiohttp mai degrabă decât requests, așa cum este creată cu async stil în minte.

Aici e o versiune care va descărca fișiere într-un bytearray într-un thread separat.

Așa cum am menționat în alte răspunsuri și comentarii, există și alte alternativs, care sunt dezvoltate cu async operațiuni în minte, așa că nu citesc prea mult în decizia de a merge cu threadinge doar pentru a demonstra conceptul (și din cauza de comoditate, deoarece acesta vine cu python).

În codul de mai jos, dacă dimensiunea fișierului este cunoscut, fiecare . va corespunde la 1%. Ca un bonus, descărcat și numărul total de octeți vor fi tipărite la începutul linie de genul (1234 B / 1234567 B). Dacă dimensiunea nu este cunoscut, soluție de rezervă este de a avea fiecare . reprezintă o bucată.

import requests
import threading


def download_file(url: str):
    headers = {"<some_key>": "<some_value>"}
    data = bytearray()
    with requests.get(url, headers=headers, stream=True) as request:
        if file_size := request.headers.get("Content-Length"):
            file_size = int(file_size)
        else:
            file_size = None
        received = 0
        for chunk in request.iter_content(chunk_size=2**15):
            received += len(chunk)
            data += chunk
            try:
                num_dots = int(received * 100 / file_size)
                print(
                    f"({received} B/{file_size} B) "
                    + "." * num_dots, end="\r"
                )
            except TypeError:
                print(".", end="")
        print("\nDone!")

url = "<some_url>"
thread = threading.Thread(target=download_file, args=(url,))
thread.start()
# Do something in the meantime
thread.join()

Nu păstrați în minte că am plecat de blocare pentru a proteja împotriva accesul simultan la stdout pentru a reduce zgomotul. De asemenea, am lăsat de scris bytarray la dosar, la sfârșitul (sau scris pe bucăți de fișiere în care acestea sunt primite dacă fișierul este mare), dar păstrează în minte că, poate doriți să utilizați un sistem de blocare pentru că la fel de bine, dacă ai citit și/sau scrie la același fișier în orice altă parte a script-ul.