From d5611b9b55c9009c467804a5907343e157058fbf Mon Sep 17 00:00:00 2001 From: Ayush Raj Date: Mon, 16 Nov 2026 03:14:33 +0530 Subject: [PATCH 1/2] feat(download): add --resume support for interrupted downloads via HTTP Range headers # Conflicts: # databusclient/api/download.py # databusclient/cli.py --- databusclient/api/download.py | 246 +++++++++++++++++++++++----------- databusclient/cli.py | 10 ++ tests/test_download.py | 171 +++++++++++++++++++++++ 3 files changed, 346 insertions(+), 81 deletions(-) diff --git a/databusclient/api/download.py b/databusclient/api/download.py index e2c0755..aa9e43d 100644 --- a/databusclient/api/download.py +++ b/databusclient/api/download.py @@ -443,6 +443,7 @@ def _download_file( expected_checksum: str | None = None, manifest_context=None, user_agent: str | None = None, + resume: bool = False, ) -> None: """Download a file from the internet with a progress bar using tqdm. @@ -463,6 +464,8 @@ def _download_file( user_agent: HTTP User-Agent for file HEAD and GET requests. None uses the versioned databus-python-client default. expected_checksum: The expected checksum of the file. + manifest_context: ManifestContext instance to record download metadata. + resume: Whether to resume partially downloaded files via HTTP Range headers. """ _validate_graph_mode(graph_mode) source_url = url @@ -478,6 +481,11 @@ def _download_file( if dirpath: os.makedirs(dirpath, exist_ok=True) # Create the necessary directories + # Check for existing partial file if resume requested + existing_bytes = 0 + if resume and os.path.isfile(filename): + existing_bytes = os.path.getsize(filename) + # --- 1. Get redirect URL by requesting HEAD --- headers = {"User-Agent": DEFAULT_USER_AGENT if user_agent is None else user_agent} @@ -521,97 +529,149 @@ def _download_file( headers["X-API-KEY"] = databus_key response = requests.head(url, headers=headers, timeout=30) - # --- 2. Try direct GET to redirected URL --- - headers["Accept-Encoding"] = ( - "identity" # disable gzip to get correct content-length - ) - response = requests.get( - url, headers=headers, stream=True, allow_redirects=True, timeout=30 - ) - www = response.headers.get( - "WWW-Authenticate", "" - ) # Check if authentication is required - - # --- 3. Handle authentication responses --- - # 3a. Server requests Bearer auth. Only attempt token exchange for hosts - # we explicitly consider Vault-protected (VAULT_REQUIRED_HOSTS). This avoids - # sending tokens to unrelated hosts and makes auth behavior predictable. - if response.status_code == 401 and "bearer" in www.lower(): - # If host is not configured for Vault, do not attempt token exchange. - if host not in VAULT_REQUIRED_HOSTS: - raise DownloadAuthError( - "Server requests Bearer authentication but this host is not configured for Vault token exchange." - " Try providing a databus API key with --databus-key or contact your administrator." - ) + file_already_complete = False + if ( + existing_bytes > 0 + and response.status_code == 200 + and response.headers.get("content-length") + ): + try: + remote_total = int(response.headers.get("content-length")) + if remote_total > 0 and existing_bytes == remote_total: + print( + f"File {filename} is already completely downloaded ({existing_bytes} bytes)." + ) + file_already_complete = True + total_size_in_bytes = existing_bytes + elif remote_total > 0 and existing_bytes > remote_total: + print( + f"Existing file {filename} ({existing_bytes} bytes) is larger than remote size ({remote_total} bytes). Restarting download." + ) + existing_bytes = 0 + except (ValueError, TypeError): + pass - # Host requires Vault; ensure token file provided. - if not vault_token_file: - raise DownloadAuthError( - f"Vault token required for host '{host}', but no token was provided. Please use --vault-token." + if not file_already_complete: + if existing_bytes > 0: + headers["Range"] = f"bytes={existing_bytes}-" + + # --- 2. Try direct GET to redirected URL --- + headers["Accept-Encoding"] = ( + "identity" # disable gzip to get correct content-length + ) + response = requests.get( + url, headers=headers, stream=True, allow_redirects=True, timeout=30 + ) + www = response.headers.get( + "WWW-Authenticate", "" + ) # Check if authentication is required + + # --- 3. Handle authentication responses --- + # 3a. Server requests Bearer auth. Only attempt token exchange for hosts + # we explicitly consider Vault-protected (VAULT_REQUIRED_HOSTS). This avoids + # sending tokens to unrelated hosts and makes auth behavior predictable. + if response.status_code == 401 and "bearer" in www.lower(): + # If host is not configured for Vault, do not attempt token exchange. + if host not in VAULT_REQUIRED_HOSTS: + raise DownloadAuthError( + "Server requests Bearer authentication but this host is not configured for Vault token exchange." + " Try providing a databus API key with --databus-key or contact your administrator." + ) + + # Host requires Vault; ensure token file provided. + if not vault_token_file: + raise DownloadAuthError( + f"Vault token required for host '{host}', but no token was provided. Please use --vault-token." + ) + + # --- 3b. Fetch Vault token and retry --- + # Token exchange is potentially sensitive and should only be performed + # for known hosts. __get_vault_access__ handles reading the refresh + # token and exchanging it; errors are translated to DownloadAuthError + # for user-friendly CLI output. + vault_token = __get_vault_access__( + url, vault_token_file, auth_url, client_id ) + headers["Authorization"] = f"Bearer {vault_token}" - # --- 3b. Fetch Vault token and retry --- - # Token exchange is potentially sensitive and should only be performed - # for known hosts. __get_vault_access__ handles reading the refresh - # token and exchanging it; errors are translated to DownloadAuthError - # for user-friendly CLI output. - vault_token = __get_vault_access__(url, vault_token_file, auth_url, client_id) - headers["Authorization"] = f"Bearer {vault_token}" + # Retry with token + response = requests.get(url, headers=headers, stream=True, timeout=30) - # Retry with token - response = requests.get(url, headers=headers, stream=True, timeout=30) + # Map common auth failures to friendly messages + if response.status_code == 401: + raise DownloadAuthError( + "Vault token is invalid or expired. Please generate a new token." + ) + if response.status_code == 403: + raise DownloadAuthError( + "Vault token is valid but has insufficient permissions to access this file." + ) - # Map common auth failures to friendly messages - if response.status_code == 401: + # 3c. Generic forbidden without Bearer challenge + if response.status_code == 403: raise DownloadAuthError( - "Vault token is invalid or expired. Please generate a new token." + "Access forbidden: your token or API key does not have permission to download this file." ) - if response.status_code == 403: + + # 3d. Generic unauthorized without Bearer + if response.status_code == 401: raise DownloadAuthError( - "Vault token is valid but has insufficient permissions to access this file." + "Unauthorized: access denied. Check your --databus-key or --vault-token settings." ) - # 3c. Generic forbidden without Bearer challenge - if response.status_code == 403: - raise DownloadAuthError( - "Access forbidden: your token or API key does not have permission to download this file." - ) + try: + if response.status_code == 416 and existing_bytes > 0: + print( + f"Server returned 416 Range Not Satisfiable for {url}. File {filename} may already be completely downloaded." + ) + total_size_in_bytes = existing_bytes + file_already_complete = True + else: + response.raise_for_status() # Raise if still failing + except requests.exceptions.HTTPError as e: + if response.status_code == 404: + print(f"WARNING: Skipping file {url} because it was not found (404).") + if manifest_context is not None: + manifest_context.record_file( + url=url, + status="failed", + error_message="404 Not Found", + ) + return + else: + raise e - # 3d. Generic unauthorized without Bearer - if response.status_code == 401: - raise DownloadAuthError( - "Unauthorized: access denied. Check your --databus-key or --vault-token settings." - ) + if not file_already_complete: + # --- 4. Download with progress bar --- + is_partial = response.status_code == 206 + file_mode = "ab" if (is_partial and existing_bytes > 0) else "wb" + content_len = int(response.headers.get("content-length", 0)) - try: - response.raise_for_status() # Raise if still failing - except requests.exceptions.HTTPError as e: - if response.status_code == 404: - print(f"WARNING: Skipping file {url} because it was not found (404).") - if manifest_context is not None: - manifest_context.record_file( - url=url, - status="failed", - error_message="404 Not Found", - ) - return + if is_partial and existing_bytes > 0: + total_size_in_bytes = existing_bytes + content_len + initial_bytes = existing_bytes else: - raise e + total_size_in_bytes = content_len + initial_bytes = 0 - # --- 4. Download with progress bar --- - total_size_in_bytes = int(response.headers.get("content-length", 0)) - block_size = 1024 # 1 KiB + block_size = 1024 # 1 KiB - progress_bar = tqdm(total=total_size_in_bytes, unit="iB", unit_scale=True) - with open(filename, "wb") as f: - for data in response.iter_content(block_size): - progress_bar.update(len(data)) - f.write(data) - progress_bar.close() + progress_bar = tqdm( + total=total_size_in_bytes, + initial=initial_bytes, + unit="iB", + unit_scale=True, + ) + with open(filename, file_mode) as f: + for data in response.iter_content(block_size): + progress_bar.update(len(data)) + f.write(data) + progress_bar.close() - # --- 5. Verify download size --- - if total_size_in_bytes != 0 and progress_bar.n != total_size_in_bytes: - raise IOError("Downloaded size does not match Content-Length header") + # --- 5. Verify download size --- + final_size = os.path.getsize(filename) + if total_size_in_bytes != 0 and final_size != total_size_in_bytes: + raise IOError("Downloaded size does not match Content-Length header") # --- 6. Validate checksum on original downloaded file (BEFORE conversion) --- actual_checksum = None @@ -892,6 +952,7 @@ def _download_files( validate_checksum: bool = False, checksums: dict | None = None, user_agent: str | None = None, + resume: bool = False, ) -> None: """Download multiple files from the databus. @@ -911,6 +972,7 @@ def _download_files( user_agent: HTTP User-Agent for file HEAD and GET requests. None uses the versioned databus-python-client default. checksums: Dictionary mapping URLs to their expected checksums. + resume: Whether to resume partially downloaded files via HTTP Range headers. """ for url in urls: expected = None @@ -932,6 +994,7 @@ def _download_files( expected_checksum=expected, manifest_context=manifest_context, user_agent=user_agent, + resume=resume, ) @@ -1084,6 +1147,7 @@ def _download_collection( manifest_context=None, validate_checksum: bool = False, user_agent: str | None = None, + resume: bool = False, ) -> None: """Download all files in a databus collection. @@ -1103,6 +1167,7 @@ def _download_collection( validate_checksum: Whether to validate checksums after downloading. user_agent: HTTP User-Agent for file HEAD and GET requests. None uses the versioned databus-python-client default. + resume: Whether to resume partially downloaded files via HTTP Range headers. """ query = _get_sparql_query_of_collection(uri, databus_key=databus_key) file_urls = _get_file_download_urls_from_sparql_query( @@ -1130,6 +1195,7 @@ def _download_collection( validate_checksum=validate_checksum, checksums=checksums if checksums else None, user_agent=user_agent, + resume=resume, ) @@ -1148,6 +1214,7 @@ def _download_version( manifest_context=None, validate_checksum: bool = False, user_agent: str | None = None, + resume: bool = False, ) -> None: """Download all files in a databus artifact version. @@ -1166,6 +1233,7 @@ def _download_version( validate_checksum: Whether to validate checksums after downloading. user_agent: HTTP User-Agent for file HEAD and GET requests. None uses the versioned databus-python-client default. + resume: Whether to resume partially downloaded files via HTTP Range headers. """ json_str = fetch_databus_jsonld(uri, databus_key=databus_key) file_urls = _get_file_download_urls_from_artifact_jsonld(json_str) @@ -1192,6 +1260,7 @@ def _download_version( validate_checksum=validate_checksum, checksums=checksums, user_agent=user_agent, + resume=resume, ) @@ -1211,6 +1280,7 @@ def _download_artifact( manifest_context=None, validate_checksum: bool = False, user_agent: str | None = None, + resume: bool = False, ) -> None: """Download files in a databus artifact. @@ -1230,6 +1300,7 @@ def _download_artifact( validate_checksum: Whether to validate checksums after downloading. user_agent: HTTP User-Agent for file HEAD and GET requests. None uses the versioned databus-python-client default. + resume: Whether to resume partially downloaded files via HTTP Range headers. """ json_str = fetch_databus_jsonld(uri, databus_key=databus_key) versions = _get_databus_versions_of_artifact(json_str, all_versions=all_versions) @@ -1262,6 +1333,7 @@ def _download_artifact( validate_checksum=validate_checksum, checksums=checksums, user_agent=user_agent, + resume=resume, ) @@ -1342,6 +1414,7 @@ def _download_group( manifest_context=None, validate_checksum: bool = False, user_agent: str | None = None, + resume: bool = False, ) -> None: """Download files in a databus group. @@ -1361,6 +1434,7 @@ def _download_group( validate_checksum: Whether to validate checksums after downloading. user_agent: HTTP User-Agent for file HEAD and GET requests. None uses the versioned databus-python-client default. + resume: Whether to resume partially downloaded files via HTTP Range headers. """ json_str = fetch_databus_jsonld(uri, databus_key=databus_key) artifacts = _get_databus_artifacts_of_group(json_str) @@ -1382,6 +1456,7 @@ def _download_group( manifest_context=manifest_context, validate_checksum=validate_checksum, user_agent=user_agent, + resume=resume, ) @@ -1436,6 +1511,7 @@ def download( validate_checksum: bool = False, manifest_context=None, user_agent: str | None = None, + resume: bool = False, ) -> None: """Download datasets from databus. @@ -1458,6 +1534,8 @@ def download( validate_checksum: Whether to validate checksums after downloading. user_agent: HTTP User-Agent for file HEAD and GET requests. None uses the versioned databus-python-client default. + manifest_context: ManifestContext instance to record download metadata. + resume: Whether to resume partially downloaded files via HTTP Range headers. """ _validate_graph_mode(graph_mode) for databusURI in databusURIs: @@ -1490,7 +1568,8 @@ def download( graph_mode=graph_mode, manifest_context=manifest_context, validate_checksum=validate_checksum, - user_agent=user_agent, + user_agent=user_agent, + resume=resume, ) elif file is not None: print(f"Downloading file: {databusURI}") @@ -1517,7 +1596,8 @@ def download( manifest_context=manifest_context, validate_checksum=validate_checksum, expected_checksum=expected, - user_agent=user_agent, + user_agent=user_agent, + resume=resume, ) elif version is not None: print(f"Downloading version: {databusURI}") @@ -1535,7 +1615,8 @@ def download( graph_mode=graph_mode, manifest_context=manifest_context, validate_checksum=validate_checksum, - user_agent=user_agent, + user_agent=user_agent, + resume=resume, ) elif artifact is not None: print( @@ -1556,7 +1637,8 @@ def download( graph_mode=graph_mode, manifest_context=manifest_context, validate_checksum=validate_checksum, - user_agent=user_agent, + user_agent=user_agent, + resume=resume, ) elif group is not None and group != "collections": print( @@ -1577,7 +1659,8 @@ def download( graph_mode=graph_mode, manifest_context=manifest_context, validate_checksum=validate_checksum, - user_agent=user_agent, + user_agent=user_agent, + resume=resume, ) elif account is not None: print("accountId not supported yet") # TODO @@ -1621,5 +1704,6 @@ def download( manifest_context=manifest_context, validate_checksum=validate_checksum, checksums=checksums if checksums else None, - user_agent=user_agent, + user_agent=user_agent, + resume=resume, ) diff --git a/databusclient/cli.py b/databusclient/cli.py index 4b12e47..b959d16 100644 --- a/databusclient/cli.py +++ b/databusclient/cli.py @@ -326,6 +326,13 @@ def _write_manifest(): "--user-agent", help="Override the HTTP User-Agent header for file download requests.", ) +@click.option( + "--resume", + "-c", + is_flag=True, + default=False, + help="Resume partially downloaded files using HTTP Range requests.", +) def download( databusuris: List[str], localdir, @@ -343,6 +350,7 @@ def download( validate_checksum, manifest_path, user_agent, + resume, ): """ Download datasets from databus, optionally using vault access if vault options are provided. @@ -375,6 +383,7 @@ def download( "user_agent": user_agent, "authurl": authurl, "clientid": clientid, + "resume": resume, }) try: api_download( @@ -394,6 +403,7 @@ def download( validate_checksum=validate_checksum, user_agent=user_agent, manifest_context=manifest_context, + resume=resume, ) except DownloadAuthError as e: if manifest_context: diff --git a/tests/test_download.py b/tests/test_download.py index b1746d3..58c09c1 100644 --- a/tests/test_download.py +++ b/tests/test_download.py @@ -254,3 +254,174 @@ def test_converted_output_gets_graph_sidecar_at_final_path(monkeypatch, tmp_path assert not (tmp_path / "data.ttl.bz2").exists() assert not (tmp_path / "data.ttl.bz2.graph").exists() assert (tmp_path / "data.ttl.graph").read_text(encoding="utf-8") == url + + +def test_resume_partial_download_appends_bytes(monkeypatch, tmp_path): + import databusclient.api.download as dl + + url = "https://example.org/file.bin" + target_file = tmp_path / "file.bin" + target_file.write_bytes(b"hello ") + + recorded_headers = {} + + class MockHeadResp: + status_code = 200 + headers = {"content-length": "12"} + + class MockPartialGetResp: + status_code = 206 + + def __init__(self): + self.content = b"world!" + self.headers = {"content-length": str(len(self.content))} + + def iter_content(self, block_size): + yield self.content + + def raise_for_status(self): + pass + + def mock_head(req_url, *args, **kwargs): + return MockHeadResp() + + def mock_get(req_url, *args, **kwargs): + if "headers" in kwargs: + recorded_headers.update(kwargs["headers"]) + return MockPartialGetResp() + + monkeypatch.setattr("requests.head", mock_head) + monkeypatch.setattr("requests.get", mock_get) + + dl._download_file(url, localDir=str(tmp_path), resume=True) + + assert recorded_headers.get("Range") == "bytes=6-" + assert target_file.read_bytes() == b"hello world!" + + +def test_resume_server_falls_back_to_200(monkeypatch, tmp_path): + import databusclient.api.download as dl + + url = "https://example.org/file.bin" + target_file = tmp_path / "file.bin" + target_file.write_bytes(b"old_partial_data") + + class MockHeadResp: + status_code = 200 + headers = {"content-length": "18"} + + class MockFullGetResp: + status_code = 200 + + def __init__(self): + self.content = b"fresh_full_content" + self.headers = {"content-length": str(len(self.content))} + + def iter_content(self, block_size): + yield self.content + + def raise_for_status(self): + pass + + monkeypatch.setattr("requests.head", lambda *a, **k: MockHeadResp()) + monkeypatch.setattr("requests.get", lambda *a, **k: MockFullGetResp()) + + dl._download_file(url, localDir=str(tmp_path), resume=True) + + assert target_file.read_bytes() == b"fresh_full_content" + + +def test_resume_already_complete_file_416(monkeypatch, tmp_path): + import requests + import databusclient.api.download as dl + + url = "https://example.org/file.bin" + target_file = tmp_path / "file.bin" + content = b"already completed content" + target_file.write_bytes(content) + + class MockHeadResp: + status_code = 200 + headers = {} # HEAD does not include content-length, forcing GET + + class Mock416GetResp: + status_code = 416 + headers = {} + + def raise_for_status(self): + raise requests.exceptions.HTTPError("416 Range Not Satisfiable", response=self) + + monkeypatch.setattr("requests.head", lambda *a, **k: MockHeadResp()) + monkeypatch.setattr("requests.get", lambda *a, **k: Mock416GetResp()) + + dl._download_file(url, localDir=str(tmp_path), resume=True) + + assert target_file.read_bytes() == content + + +def test_resume_already_complete_file_via_head(monkeypatch, tmp_path): + import databusclient.api.download as dl + + url = "https://example.org/file.bin" + target_file = tmp_path / "file.bin" + content = b"already completed content" + target_file.write_bytes(content) + + get_called = False + + class MockHeadResp: + status_code = 200 + headers = {"content-length": str(len(content))} + + def mock_get(*args, **kwargs): + nonlocal get_called + get_called = True + raise AssertionError("GET should not be called when HEAD confirms file is complete") + + monkeypatch.setattr("requests.head", lambda *a, **k: MockHeadResp()) + monkeypatch.setattr("requests.get", mock_get) + + dl._download_file(url, localDir=str(tmp_path), resume=True) + + assert not get_called + assert target_file.read_bytes() == content + + +def test_cli_resume_flag(monkeypatch, tmp_path): + import json + + captured = {} + + def fake_download(**kwargs): + captured.update(kwargs) + + monkeypatch.setattr("databusclient.cli.api_download", fake_download) + manifest_path = tmp_path / "download.jsonld" + + result = CliRunner().invoke( + app, + [ + "download", + "https://example.org/data.ttl", + "--resume", + "--manifest", + str(manifest_path), + ], + ) + + assert result.exit_code == 0 + assert captured["resume"] is True + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + assert manifest["dbus:replayParams"]["resume"] is True + + # Also test short alias -c + result_short = CliRunner().invoke( + app, + [ + "download", + "https://example.org/data.ttl", + "-c", + ], + ) + assert result_short.exit_code == 0 + assert captured["resume"] is True From 8079cdeae8a6f03ea3b354a7c58e8b80eab7dbaa Mon Sep 17 00:00:00 2001 From: Ayush Raj Date: Mon, 16 Nov 2026 03:18:23 +0530 Subject: [PATCH 2/2] fix(download): validate Content-Range headers, add If-Range, and forward resume in replay --- databusclient/api/download.py | 66 +++++++++++++-- databusclient/manifest/replay.py | 1 + databusclient/workflow/steps.py | 1 + tests/test_download.py | 135 +++++++++++++++++++++++++++++++ 4 files changed, 195 insertions(+), 8 deletions(-) diff --git a/databusclient/api/download.py b/databusclient/api/download.py index aa9e43d..2f17d58 100644 --- a/databusclient/api/download.py +++ b/databusclient/api/download.py @@ -554,6 +554,12 @@ def _download_file( if not file_already_complete: if existing_bytes > 0: headers["Range"] = f"bytes={existing_bytes}-" + etag = response.headers.get("ETag") + last_mod = response.headers.get("Last-Modified") + if etag: + headers["If-Range"] = etag + elif last_mod: + headers["If-Range"] = last_mod # --- 2. Try direct GET to redirected URL --- headers["Accept-Encoding"] = ( @@ -621,11 +627,36 @@ def _download_file( try: if response.status_code == 416 and existing_bytes > 0: - print( - f"Server returned 416 Range Not Satisfiable for {url}. File {filename} may already be completely downloaded." - ) - total_size_in_bytes = existing_bytes - file_already_complete = True + # Content-Range format for 416: bytes */ + content_range = response.headers.get("Content-Range", "") + cr_match = re.search(r"bytes\s+\*/(\d+)", content_range) + if cr_match: + if int(cr_match.group(1)) == existing_bytes: + print( + f"Server returned 416 Range Not Satisfiable for {url}. File {filename} is already completely downloaded." + ) + total_size_in_bytes = existing_bytes + file_already_complete = True + else: + print( + f"Server returned 416 Range Not Satisfiable with mismatched size for {url}. Restarting download from scratch." + ) + existing_bytes = 0 + if os.path.exists(filename): + os.remove(filename) + headers.pop("Range", None) + headers.pop("If-Range", None) + response = requests.get( + url, headers=headers, stream=True, allow_redirects=True, timeout=30 + ) + response.raise_for_status() + else: + # No Content-Range header provided with 416; accept existing file + print( + f"Server returned 416 Range Not Satisfiable for {url}. File {filename} may already be completely downloaded." + ) + total_size_in_bytes = existing_bytes + file_already_complete = True else: response.raise_for_status() # Raise if still failing except requests.exceptions.HTTPError as e: @@ -644,13 +675,32 @@ def _download_file( if not file_already_complete: # --- 4. Download with progress bar --- is_partial = response.status_code == 206 - file_mode = "ab" if (is_partial and existing_bytes > 0) else "wb" content_len = int(response.headers.get("content-length", 0)) if is_partial and existing_bytes > 0: - total_size_in_bytes = existing_bytes + content_len - initial_bytes = existing_bytes + content_range = response.headers.get("Content-Range", "") + cr_match = re.search(r"bytes\s+(\d+)-(\d+)/((\d+)|\*)", content_range) + if cr_match: + range_start = int(cr_match.group(1)) + if range_start == existing_bytes: + total_str = cr_match.group(4) + total_size_in_bytes = int(total_str) if total_str else (existing_bytes + content_len) + file_mode = "ab" + initial_bytes = existing_bytes + else: + # Server sent partial content starting at wrong offset; restart from scratch + print( + f"Content-Range start {range_start} does not match local size {existing_bytes}. Restarting download." + ) + file_mode = "wb" + total_size_in_bytes = content_len + initial_bytes = 0 + else: + total_size_in_bytes = existing_bytes + content_len + file_mode = "ab" + initial_bytes = existing_bytes else: + file_mode = "wb" total_size_in_bytes = content_len initial_bytes = 0 diff --git a/databusclient/manifest/replay.py b/databusclient/manifest/replay.py index 4def7bc..718956e 100644 --- a/databusclient/manifest/replay.py +++ b/databusclient/manifest/replay.py @@ -82,6 +82,7 @@ def _build_download_kwargs( "graph_mode": replay_params.get("graph_mode"), "user_agent": replay_params.get("user_agent"), "validate_checksum": bool(replay_params.get("validate_checksum", False)), + "resume": bool(replay_params.get("resume", False)), "manifest_context": None, } diff --git a/databusclient/workflow/steps.py b/databusclient/workflow/steps.py index 320e616..d01b88e 100644 --- a/databusclient/workflow/steps.py +++ b/databusclient/workflow/steps.py @@ -82,6 +82,7 @@ def run(self, step_config: Dict[str, Any], context: StepContext) -> None: base_uri=resolved.get("base_uri"), graph_mode=resolved.get("graph_mode"), user_agent=resolved.get("user_agent"), + resume=resolved.get("resume", False), validate_checksum=resolved.get("validate_checksum", False), manifest_context=capture_context, ) diff --git a/tests/test_download.py b/tests/test_download.py index 58c09c1..6f7d12a 100644 --- a/tests/test_download.py +++ b/tests/test_download.py @@ -425,3 +425,138 @@ def fake_download(**kwargs): ) assert result_short.exit_code == 0 assert captured["resume"] is True + + +def test_resume_content_range_mismatch_restarts(monkeypatch, tmp_path): + import databusclient.api.download as dl + + url = "https://example.org/file.bin" + target_file = tmp_path / "file.bin" + target_file.write_bytes(b"corrupt") + + class MockHeadResp: + status_code = 200 + headers = {"content-length": "10"} + + class MockMismatch206GetResp: + status_code = 206 + + def __init__(self): + self.content = b"0123456789" + # Server returns range starting at 0 instead of 7 + self.headers = { + "content-length": "10", + "Content-Range": "bytes 0-9/10", + } + + def iter_content(self, block_size): + yield self.content + + def raise_for_status(self): + pass + + monkeypatch.setattr("requests.head", lambda *a, **k: MockHeadResp()) + monkeypatch.setattr("requests.get", lambda *a, **k: MockMismatch206GetResp()) + + dl._download_file(url, localDir=str(tmp_path), resume=True) + + # File should have been restarted and overwritten with the full payload + assert target_file.read_bytes() == b"0123456789" + + +def test_resume_416_mismatch_restarts(monkeypatch, tmp_path): + import requests + import databusclient.api.download as dl + + url = "https://example.org/file.bin" + target_file = tmp_path / "file.bin" + target_file.write_bytes(b"wrong_length_content_12345") + + calls = [] + + class MockHeadResp: + status_code = 200 + headers = {} + + class Mock416GetResp: + status_code = 416 + headers = {"Content-Range": "bytes */10"} + + def raise_for_status(self): + raise requests.exceptions.HTTPError("416", response=self) + + class MockFullGetResp: + status_code = 200 + headers = {"content-length": "10"} + + def iter_content(self, block_size): + yield b"0123456789" + + def raise_for_status(self): + pass + + def mock_get(req_url, *args, **kwargs): + if not calls: + calls.append("416") + return Mock416GetResp() + calls.append("200") + return MockFullGetResp() + + monkeypatch.setattr("requests.head", lambda *a, **k: MockHeadResp()) + monkeypatch.setattr("requests.get", mock_get) + + dl._download_file(url, localDir=str(tmp_path), resume=True) + + assert calls == ["416", "200"] + assert target_file.read_bytes() == b"0123456789" + + +def test_resume_if_range_header_sent(monkeypatch, tmp_path): + import databusclient.api.download as dl + + url = "https://example.org/file.bin" + target_file = tmp_path / "file.bin" + target_file.write_bytes(b"hello ") + + recorded_headers = {} + + class MockHeadResp: + status_code = 200 + headers = {"content-length": "12", "ETag": '"etag-val-123"'} + + class MockPartialGetResp: + status_code = 206 + headers = {"content-length": "6", "Content-Range": "bytes 6-11/12"} + + def iter_content(self, block_size): + yield b"world!" + + def raise_for_status(self): + pass + + monkeypatch.setattr("requests.head", lambda *a, **k: MockHeadResp()) + + def mock_get(req_url, *args, **kwargs): + if "headers" in kwargs: + recorded_headers.update(kwargs["headers"]) + return MockPartialGetResp() + + monkeypatch.setattr("requests.get", mock_get) + + dl._download_file(url, localDir=str(tmp_path), resume=True) + + assert recorded_headers.get("Range") == "bytes=6-" + assert recorded_headers.get("If-Range") == '"etag-val-123"' + assert target_file.read_bytes() == b"hello world!" + + +def test_manifest_replay_forwards_resume(): + from databusclient.manifest.replay import _build_download_kwargs + + manifest = {"@id": "https://example.org/manifest"} + replay_params = { + "databusURIs": ["https://databus.dbpedia.org/user/group/artifact/version"], + "resume": True, + } + kwargs = _build_download_kwargs(manifest, replay_params, overrides={}) + assert kwargs.get("resume") is True