upload_progress.py

import dataclasses
import functools
import io
import pathlib
from datetime import timedelta, datetime
from io import FileIO
from typing import Callable

import httpx


class ProgressIO(io.BufferedReader):
    @dataclasses.dataclass
    class Progress:
        total: int
        processed: int

        def percent(self) -> float:
            return round(100 * self.processed / self.total, 2)

    def __init__(self, file: pathlib.Path, on_progress: Callable[[Progress], None]):
        super().__init__(raw=FileIO(str(file)))
        self._size = file.stat().st_size
        self._on_progress_throttled = self.throttle(1)(on_progress)
        self._on_progress = on_progress
        self._processed = 0
        self._iter = None

    def __next__(self) -> bytes:
        chunk = super().__next__()
        self._processed += len(chunk)
        progress = ProgressIO.Progress(self._size, self._processed)
        if self._processed == self._size:
            self._on_progress(progress)
        else:
            self._on_progress_throttled(progress)

        return chunk

    @staticmethod
    def throttle(seconds: float = 0):
        period = timedelta(seconds=seconds)

        def decorator(fn):
            last_called = datetime.min

            @functools.wraps(fn)
            def wrapper(*args, **kwargs):
                now = datetime.now()
                nonlocal last_called
                if now - last_called > period:
                    last_called = now
                    return fn(*args, **kwargs)

            return wrapper

        return decorator


if __name__ == "__main__":
    f = pathlib.Path(r"/path/to/file")
    httpx.post("http://httpbin.org/post", files={"file": ProgressIO(f, on_progress=lambda p: print(p.percent()))})