Requests get / put with progress bar

requests_get_progress.py

import requests
from tqdm import tqdm
import json


CHUNK_SIZE = 32 * 1024


def get(url, token):
    headers = {"Authorization": token}
    response = requests.get(url, stream=True, headers=headers)
    data = bytearray()
    total_size = int(response.headers.get('content-length', 0))
    with tqdm(desc=f"Downloading", total=total_size, unit="B", unit_scale=True, unit_divisor=1024) as t:
        for chunk in response.iter_content(CHUNK_SIZE):
            if chunk:
                data.extend(chunk)
                t.update(len(chunk))
    response.raise_for_status()
    return json.loads(data)
  

requests_put_progress.py

import requests
from tqdm import tqdm
from io import BytesIO
import json


class ReaderWrapper(object):
    def __init__(self, callback: Callable[[int], object], stream, length):
        self.callback = callback
        self.stream = stream
        self.length = length

    def read(self, __len: int = -1) -> bytes:
        data = self.stream.read(__len)
        self.callback(len(data))
        return data

    def __len__(self):
        return self.length


def put(url, token, data):
    headers = {"Content-type": "application/json", "Authorization": token}
    data = json.dumps(data).encode("utf-8")
    with tqdm(desc=f"Uploading", total=len(data), unit="B", unit_scale=True, unit_divisor=1024) as t:
        reader_wrapper = ReaderWrapper(t.update, BytesIO(data), len(data))
        response = requests.put(url, headers=headers, data=reader_wrapper)
        response.raise_for_status()
        return (response.status_code, response.text)