Function _download_with_progress

ci/compiler/platformio_cache.py:169–210

Download HTTP/HTTPS file with cancellation support.

(
    url: str, temp_path: Path, cancel_event: threading.Event
)

def _download_with_progress(
    url: str, temp_path: Path, cancel_event: threading.Event
) -> DownloadResult:
    """Download HTTP/HTTPS file with cancellation support."""
    try:
        # HTTP/HTTPS download using httpx with streaming
        with httpx.Client(follow_redirects=True, timeout=30.0) as client:
            with client.stream(
                "GET", url, headers={"User-Agent": "PlatformIO-Cache/1.0"}
            ) as response:
                response.raise_for_status()

                with open(temp_path, "wb") as f:
                    for chunk in response.iter_bytes(chunk_size=8192):
                        # Check for cancellation periodically
                        if cancel_event.is_set():
                            cancelled_error = RuntimeError(
                                f"Download cancelled for {url}"
                            )
                            logger.warning(
                                f"Download cancelled for {url}",
                                exc_info=cancelled_error,
                            )
                            return DownloadResult(url, temp_path, cancelled_error)
                        f.write(chunk)

        return DownloadResult(url, temp_path)  # Success

    except KeyboardInterrupt as e:
        # Set cancel event and interrupt main thread
        cancel_event.set()
        _thread.interrupt_main()
        logger.warning(
            f"Download interrupted by KeyboardInterrupt for {url}", exc_info=e
        )
        return DownloadResult(url, temp_path, e)
    except Exception as e:
        if not cancel_event.is_set():
            logger.error(f"Download failed for {url}: {e}", exc_info=e)
        else:
            logger.warning(f"Download failed for {url} (cancelled): {e}", exc_info=e)
        return DownloadResult(url, temp_path, e)


def _copy_file_with_progress(