diff --git a/documentation/index.md b/documentation/index.md index 94646af..4ccaeee 100644 --- a/documentation/index.md +++ b/documentation/index.md @@ -205,7 +205,15 @@ Other useful functions: # if no folder supplied # path = Download.get_path_for_url(url, folder) -For more detail and additional functions, check the API documentation. +### Resumable downloads + +Pass `resume=True` and a fixed `path` to continue an interrupted download using +HTTP Range requests. If the server returns 200 instead of 206, the file is +downloaded from scratch. Add `retries` to automatically retry mid-stream +failures; each retry continues from the current end of the partial file. +`retries` has no effect without `resume=True`. + + path = downloader.download_file(url, path=mypath, resume=True, retries=3) ## Retrieving files @@ -261,6 +269,8 @@ Examples: retriever = Retrieve(downloader, fallback_dir, saved_dir, temp_dir, save=False, use_saved=True) data = retriever.download_json(url, filename, logstr="test json", fallback=False, log_level=logging.DEBUG) +`Retrieve.download_file` also accepts `resume=True` and `retries=N` — see [Resumable downloads](#resumable-downloads). + ## Date utilities There are utilities to parse dates. By default, *no timezone information will be parsed diff --git a/pyproject.toml b/pyproject.toml index 394aaa8..1f06988 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,6 +70,7 @@ dev = [ "pytest", "pytest-cov", "pytest-loguru", + "pytest-mock", "pre-commit", "ruff==0.14.13", ] diff --git a/src/hdx/utilities/downloader.py b/src/hdx/utilities/downloader.py index 54224ca..1cfbe84 100755 --- a/src/hdx/utilities/downloader.py +++ b/src/hdx/utilities/downloader.py @@ -250,13 +250,16 @@ def hash_stream(self, url: Path | str) -> str: f"Download of {url} failed in retrieval of stream!" % url ) - def stream_path(self, path: Path | str, errormsg: str) -> Path: + def stream_path( + self, path: Path | str, errormsg: str, append: bool = False + ) -> Path: """Stream file from url and store in provided path. Must call setup method first. Args: path: Path for downloaded file errormsg: Error message to display if there is a problem + append: Whether to append to an existing file. Defaults to False. Returns: Path of downloaded file @@ -264,7 +267,7 @@ def stream_path(self, path: Path | str, errormsg: str) -> Path: f = None try: path = Path(path) - f = path.open("wb") + f = path.open("ab" if append else "wb") for chunk in self.response.iter_content(chunk_size=10240): if chunk: # filter out keep-alive new chunks f.write(chunk) @@ -325,6 +328,8 @@ def download_file( path (str): Full path to use for downloaded file instead of folder and filename. overwrite (bool): Whether to overwrite existing file. Defaults to False. keep (bool): Whether to keep already downloaded file. Defaults to False. + resume (bool): Whether to resume a partial download using Range requests where the server supports it. Defaults to False. + retries (int): Number of times to retry a mid-stream failure. Only effective when resume=True so that each retry can continue from the partial file. Defaults to 0. post (bool): Whether to use POST instead of GET. Defaults to False. parameters (dict): Parameters to pass. Defaults to None. timeout (float): Timeout for connecting to URL. Defaults to None (no timeout). @@ -340,25 +345,84 @@ def download_file( path = kwargs.get("path") overwrite = kwargs.get("overwrite", False) keep = kwargs.get("keep", False) + resume = kwargs.get("resume", False) + retries = kwargs.get("retries", 0) try: - path = get_path_for_url(url, folder, filename, path, overwrite, keep) + # When resuming, skip uniqueness renaming so we get back the exact target path + path = get_path_for_url( + url, folder, filename, path, overwrite, keep or resume + ) except ValueError as ex: raise DownloadError(ex) from ex - if keep and exists(path): + if keep and not resume and exists(path): return path - self.setup( - url, - stream=True, - post=kwargs.get("post", False), - parameters=kwargs.get("parameters"), - timeout=kwargs.get("timeout"), - headers=kwargs.get("headers"), - encoding=kwargs.get("encoding"), - json_string=kwargs.get("json_string", False), - ) - return self.stream_path( - path, f"Download of {url} failed in retrieval of stream!" - ) + errormsg = f"Download of {url} failed in retrieval of stream!" + setup_headers = dict(kwargs.get("headers") or {}) + append = False + if resume and path.exists() and path.stat().st_size > 0: + setup_headers["Range"] = f"bytes={path.stat().st_size}-" + # Range bytes must come from the raw stream: partial gzip streams + # lack the header and cannot be decoded, so request uncompressed. + setup_headers.setdefault("Accept-Encoding", "identity") + try: + self.setup( + url, + stream=True, + post=kwargs.get("post", False), + parameters=kwargs.get("parameters"), + timeout=kwargs.get("timeout"), + headers=setup_headers, + encoding=kwargs.get("encoding"), + json_string=kwargs.get("json_string", False), + ) + except DownloadError: + if self.response is not None and self.response.status_code == 416: + return path + raise + append = self.response.status_code == 206 + else: + self.setup( + url, + stream=True, + post=kwargs.get("post", False), + parameters=kwargs.get("parameters"), + timeout=kwargs.get("timeout"), + headers=setup_headers or None, + encoding=kwargs.get("encoding"), + json_string=kwargs.get("json_string", False), + ) + attempts_remaining = retries + while True: + try: + return self.stream_path(path, errormsg, append=append) + except DownloadError: + if not resume or attempts_remaining == 0: + raise + attempts_remaining -= 1 + logger.warning( + "Download of %s interrupted, retrying (attempt %d of %d)...", + url, + retries - attempts_remaining, + retries, + ) + setup_headers["Range"] = f"bytes={path.stat().st_size}-" + setup_headers.setdefault("Accept-Encoding", "identity") + try: + self.setup( + url, + stream=True, + post=kwargs.get("post", False), + parameters=kwargs.get("parameters"), + timeout=kwargs.get("timeout"), + headers=setup_headers, + encoding=kwargs.get("encoding"), + json_string=kwargs.get("json_string", False), + ) + except DownloadError: + if self.response is not None and self.response.status_code == 416: + return path + raise + append = self.response.status_code == 206 def download(self, url: Path | str, **kwargs: Any) -> requests.Response: """Download url. diff --git a/src/hdx/utilities/retriever.py b/src/hdx/utilities/retriever.py index 7189ceb..3773a3e 100644 --- a/src/hdx/utilities/retriever.py +++ b/src/hdx/utilities/retriever.py @@ -184,6 +184,8 @@ def download_file( filename: str | None = None, logstr: str | None = None, fallback: bool = False, + resume: bool = False, + retries: int = 0, log_level: int = None, **kwargs: Any, ) -> Path: @@ -194,6 +196,8 @@ def download_file( filename: Filename of saved file. Defaults to getting from url. logstr: Text to use in log string to describe download. Defaults to filename. fallback: Whether to use static fallback if download fails. Defaults to False. + resume: Whether to resume a partial download using Range requests where the server supports it. Defaults to False. + retries: Number of times to retry a mid-stream failure. Only effective when resume=True. Defaults to 0. log_level: Level at which to log messages. Overrides level from constructor. **kwargs: Parameters to pass to download_file call @@ -219,7 +223,9 @@ def download_file( log_level, f"Downloading {logstr} from {self.get_url_logstr(url)} into {output_path}", ) - return self.downloader.download_file(url, path=output_path, **kwargs) + return self.downloader.download_file( + url, path=output_path, resume=resume, retries=retries, **kwargs + ) except DownloadError: if not fallback: raise diff --git a/tests/hdx/utilities/test_downloader.py b/tests/hdx/utilities/test_downloader.py index 20766bb..78315f3 100755 --- a/tests/hdx/utilities/test_downloader.py +++ b/tests/hdx/utilities/test_downloader.py @@ -1314,3 +1314,159 @@ def test_generate_downloaders(self, downloaders): downloader1 = Download.get_downloader() downloader2 = Download.get_downloader("test") assert_downloaders(downloader1, downloader2, downloaders) + + +class TestResumableDownload: + url = "https://example.com/testfile.bin" + + def _make_response(self, mocker, status_code, content=b""): + import requests + + mock_resp = mocker.MagicMock() + mock_resp.status_code = status_code + if status_code >= 400: + mock_resp.raise_for_status.side_effect = requests.exceptions.HTTPError( + response=mock_resp + ) + else: + mock_resp.raise_for_status.return_value = None + mock_resp.iter_content.return_value = iter([content] if content else []) + return mock_resp + + def test_resume_206_appends(self, tmp_path, mocker): + partial = tmp_path / "file.bin" + partial.write_bytes(b"hello ") + mock_resp = self._make_response(mocker, 206, b"world") + with Download(user_agent="test") as downloader: + mocker.patch.object(downloader.session, "get", return_value=mock_resp) + result = downloader.download_file(self.url, path=partial, resume=True) + assert result == partial + assert partial.read_bytes() == b"hello world" + sent_headers = downloader.session.get.call_args.kwargs["headers"] + assert sent_headers["Range"] == "bytes=6-" + + def test_resume_200_overwrites(self, tmp_path, mocker): + partial = tmp_path / "file.bin" + partial.write_bytes(b"partial") + mock_resp = self._make_response(mocker, 200, b"fullcontent") + with Download(user_agent="test") as downloader: + mocker.patch.object(downloader.session, "get", return_value=mock_resp) + downloader.download_file(self.url, path=partial, resume=True) + assert partial.read_bytes() == b"fullcontent" + + def test_resume_416_returns_existing(self, tmp_path, mocker): + partial = tmp_path / "file.bin" + partial.write_bytes(b"complete") + mock_resp = self._make_response(mocker, 416) + with Download(user_agent="test") as downloader: + mocker.patch.object(downloader.session, "get", return_value=mock_resp) + result = downloader.download_file(self.url, path=partial, resume=True) + assert result == partial + assert partial.read_bytes() == b"complete" + + def test_resume_no_partial_file_does_full_download(self, tmp_path, mocker): + path = tmp_path / "newfile.bin" + mock_resp = self._make_response(mocker, 200, b"fullcontent") + with Download(user_agent="test") as downloader: + mocker.patch.object(downloader.session, "get", return_value=mock_resp) + downloader.download_file(self.url, path=path, resume=True) + assert path.read_bytes() == b"fullcontent" + sent_headers = downloader.session.get.call_args.kwargs.get("headers") + assert sent_headers is None or "Range" not in sent_headers + + def test_no_resume_overwrites_partial(self, tmp_path, mocker): + partial = tmp_path / "file.bin" + partial.write_bytes(b"staledata") + mock_resp = self._make_response(mocker, 200, b"freshcontent") + with Download(user_agent="test") as downloader: + mocker.patch.object(downloader.session, "get", return_value=mock_resp) + downloader.download_file(self.url, path=partial, overwrite=True) + assert partial.read_bytes() == b"freshcontent" + sent_headers = downloader.session.get.call_args.kwargs.get("headers") + assert sent_headers is None or "Range" not in sent_headers + + def test_resume_real_file(self, tmp_path, fixturesfolder): + fixture = fixturesfolder / "downloader" / "resume_test.dat" + expected = fixture.read_bytes() + split_at = 40 + partial = tmp_path / "resume_test.dat" + partial.write_bytes(expected[:split_at]) + url = "https://raw.githubusercontent.com/OCHA-DAP/hdx-python-utilities/main/tests/fixtures/downloader/resume_test.dat" + with Download(user_agent="test") as downloader: + result = downloader.download_file(url, path=partial, resume=True) + assert result == partial + assert partial.read_bytes() == expected + + def test_retry_succeeds_on_second_attempt(self, tmp_path, mocker): + partial = tmp_path / "file.bin" + partial.write_bytes(b"hello ") + first_resp = self._make_response(mocker, 206, b"wo") + first_resp.iter_content.side_effect = Exception("connection reset") + second_resp = self._make_response(mocker, 206, b"world") + with Download(user_agent="test") as downloader: + mocker.patch.object( + downloader.session, "get", side_effect=[first_resp, second_resp] + ) + result = downloader.download_file( + self.url, path=partial, resume=True, retries=1 + ) + assert result == partial + assert partial.read_bytes() == b"hello world" + second_call_headers = downloader.session.get.call_args_list[1].kwargs["headers"] + assert second_call_headers["Range"] == "bytes=6-" + + def test_retry_exhausted_raises(self, tmp_path, mocker): + partial = tmp_path / "file.bin" + partial.write_bytes(b"hello ") + failing_resp = self._make_response(mocker, 206, b"world") + failing_resp.iter_content.side_effect = Exception("connection reset") + with Download(user_agent="test") as downloader: + mocker.patch.object( + downloader.session, + "get", + return_value=failing_resp, + ) + with pytest.raises(DownloadError): + downloader.download_file(self.url, path=partial, resume=True, retries=2) + assert downloader.session.get.call_count == 3 + + def test_retry_not_attempted_without_resume(self, tmp_path, mocker): + path = tmp_path / "file.bin" + failing_resp = self._make_response(mocker, 200, b"content") + failing_resp.iter_content.side_effect = Exception("connection reset") + with Download(user_agent="test") as downloader: + mocker.patch.object(downloader.session, "get", return_value=failing_resp) + with pytest.raises(DownloadError): + downloader.download_file(self.url, path=path, retries=3) + assert downloader.session.get.call_count == 1 + + def test_retry_updates_range_from_partial_bytes(self, tmp_path, mocker): + partial = tmp_path / "file.bin" + partial.write_bytes(b"hello ") + + def make_partial_resp(content, fail_after): + resp = self._make_response(mocker, 206, b"") + chunks = [content[i : i + 1] for i in range(len(content))] + + def iter_content_side_effect(chunk_size=1): + for i, chunk in enumerate(chunks): + if i == fail_after: + raise Exception("connection reset") + yield chunk + + resp.iter_content.side_effect = iter_content_side_effect + return resp + + first_resp = make_partial_resp(b"wor", fail_after=2) + second_resp = self._make_response(mocker, 206, b"rld") + with Download(user_agent="test") as downloader: + mocker.patch.object( + downloader.session, "get", side_effect=[first_resp, second_resp] + ) + result = downloader.download_file( + self.url, path=partial, resume=True, retries=1 + ) + assert result == partial + assert partial.read_bytes() == b"hello world" + second_call_headers = downloader.session.get.call_args_list[1].kwargs["headers"] + assert second_call_headers["Range"] == "bytes=8-" diff --git a/tests/hdx/utilities/test_retriever.py b/tests/hdx/utilities/test_retriever.py index 87cdace..8a499de 100755 --- a/tests/hdx/utilities/test_retriever.py +++ b/tests/hdx/utilities/test_retriever.py @@ -462,6 +462,27 @@ def test_download_usesaved(self, dirs, retrieverfolder, fallback_dir): # Uses saved file assert headers == ["header1", "header2", "header3", "header4"] + def test_resume_forwarded_to_downloader(self, dirs, fallback_dir, mocker): + saved_dir, temp_dir = dirs + downloader = mocker.MagicMock(spec=Download) + filename = "test.txt" + expected_path = temp_dir / filename + downloader.download_file.return_value = expected_path + with Retrieve( + downloader, + fallback_dir, + saved_dir, + temp_dir, + save=False, + use_saved=False, + ) as retriever: + url = "http://example.com/test.txt" + path = retriever.download_file(url, filename, resume=True) + assert path == expected_path + downloader.download_file.assert_called_once_with( + url, path=expected_path, resume=True, retries=0 + ) + def test_generate_retrievers(self, downloaders, dirs, fallback_dir): saved_dir, temp_dir = dirs Retrieve.generate_retrievers( diff --git a/uv.lock b/uv.lock index 70ab34a..5c5b0f8 100644 --- a/uv.lock +++ b/uv.lock @@ -565,6 +565,7 @@ dev = [ { name = "pytest" }, { name = "pytest-cov" }, { name = "pytest-loguru" }, + { name = "pytest-mock" }, { name = "ruff" }, ] @@ -599,6 +600,7 @@ dev = [ { name = "pytest" }, { name = "pytest-cov" }, { name = "pytest-loguru" }, + { name = "pytest-mock" }, { name = "ruff", specifier = "==0.14.13" }, ] @@ -1335,6 +1337,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/33/ef/b0c2e96e3508bca8d1874e39789d541cd7f4731b38bcf9c7098f0b882001/pytest_loguru-0.4.0-py3-none-any.whl", hash = "sha256:3cc7b9c6b22cb158209ccbabf0d678dacd3f3c7497d6f46f1c338c13bee1ac77", size = 3886, upload-time = "2024-03-20T00:52:12.72Z" }, ] +[[package]] +name = "pytest-mock" +version = "3.15.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/68/14/eb014d26be205d38ad5ad20d9a80f7d201472e08167f0bb4361e251084a9/pytest_mock-3.15.1.tar.gz", hash = "sha256:1849a238f6f396da19762269de72cb1814ab44416fa73a8686deac10b0d87a0f", size = 34036, upload-time = "2025-09-16T16:37:27.081Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/5a/cc/06253936f4a7fa2e0f48dfe6d851d9c56df896a9ab09ac019d70b760619c/pytest_mock-3.15.1-py3-none-any.whl", hash = "sha256:0a25e2eb88fe5168d535041d09a4529a188176ae608a6d249ee65abc0949630d", size = 10095, upload-time = "2025-09-16T16:37:25.734Z" }, +] + [[package]] name = "python-dateutil" version = "2.9.0.post0"