diff --git a/databusclient/api/delete.py b/databusclient/api/delete.py index 199e5a4..dfb6b04 100644 --- a/databusclient/api/delete.py +++ b/databusclient/api/delete.py @@ -14,6 +14,7 @@ from databusclient.api.utils import ( fetch_databus_jsonld, get_databus_id_parts_from_file_url, + get_http_session, ) @@ -23,16 +24,26 @@ class DeleteQueue: Allows adding multiple databus URIs to a queue and executing their deletion in batch. """ - def __init__(self, databus_key: str, manifest_context=None): + def __init__( + self, + databus_key: str, + manifest_context=None, + session: requests.Session | None = None, + timeout: int = 30, + ): """Create a DeleteQueue bound to a given Databus API key. Args: databus_key: API key used to authenticate deletion requests. manifest_context: Optional ManifestContext to record deletion outcomes into. Passed through to _delete_list on execute(). + session: Optional HTTP session with retry strategy. + timeout: Request timeout in seconds. """ self.databus_key = databus_key self.manifest_context = manifest_context + self.session = session + self.timeout = timeout self.queue: set[str] = set() def add_uri(self, databusURI: str): @@ -79,6 +90,8 @@ def execute(self): self.databus_key, force=True, manifest_context=self.manifest_context, + session=self.session, + timeout=self.timeout, ) @@ -122,6 +135,8 @@ def _delete_resource( force: bool = False, queue: DeleteQueue = None, manifest_context=None, + session: requests.Session | None = None, + timeout: int = 30, ): """Delete a single Databus resource (version, artifact, group). @@ -134,6 +149,8 @@ def _delete_resource( dry_run: If True, do not perform the deletion but only print what would be deleted. force: If True, skip confirmation prompt and proceed with deletion. queue: If queue is provided, add the URI to the queue instead of deleting immediately. + session: Optional HTTP session to use for requests. + timeout: Request timeout in seconds. """ # Confirm the deletion request, skip the request or cancel deletion process @@ -158,9 +175,12 @@ def _delete_resource( queue.add_uri(databusURI) return + if session is None: + session = get_http_session() + print(f"[DELETE] {databusURI}") headers = {"accept": "*/*", "X-API-KEY": databus_key} - response = requests.delete(databusURI, headers=headers, timeout=30) + response = session.delete(databusURI, headers=headers, timeout=timeout) if response.status_code in (200, 204): print(f"Successfully deleted: {databusURI}") @@ -179,6 +199,8 @@ def _delete_list( force: bool = False, queue: DeleteQueue = None, manifest_context=None, + session: requests.Session | None = None, + timeout: int = 30, ): """Delete a list of Databus resources. @@ -188,10 +210,19 @@ def _delete_list( dry_run: If True, do not perform the deletion but only print what would be deleted. force: If True, skip confirmation prompt and proceed with deletion. queue: If queue is provided, add the URIs to the queue instead of deleting immediately. + session: Optional HTTP session with retry strategy. + timeout: Request timeout in seconds. """ for databusURI in databusURIs: _delete_resource( - databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + databusURI, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + session=session, + timeout=timeout, ) @@ -202,6 +233,8 @@ def _delete_artifact( force: bool = False, queue: DeleteQueue = None, manifest_context=None, + session: requests.Session | None = None, + timeout: int = 30, ): """Delete an artifact and all its versions. @@ -214,8 +247,10 @@ def _delete_artifact( dry_run: If True, do not perform the deletion but only print what would be deleted. force: If True, skip confirmation prompt and proceed with deletion. queue: If queue is provided, add the URI to the queue instead of deleting immediately. + session: Optional HTTP session with retry strategy. + timeout: Request timeout in seconds. """ - artifact_body = fetch_databus_jsonld(databusURI, databus_key) + artifact_body = fetch_databus_jsonld(databusURI, databus_key, session=session, timeout=timeout) json_dict = json.loads(artifact_body) versions = json_dict.get("databus:hasVersion") @@ -235,11 +270,27 @@ def _delete_artifact( else: # Delete all versions _delete_list( - version_uris, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + version_uris, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + session=session, + timeout=timeout, ) # Finally, delete the artifact itself - _delete_resource(databusURI, databus_key, dry_run=dry_run, force=force, queue=queue,manifest_context=manifest_context) + _delete_resource( + databusURI, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + session=session, + timeout=timeout, + ) def _delete_group( @@ -249,6 +300,8 @@ def _delete_group( force: bool = False, queue: DeleteQueue = None, manifest_context=None, + session: requests.Session | None = None, + timeout: int = 30, ): """Delete a group and all its artifacts and versions. @@ -261,8 +314,10 @@ def _delete_group( dry_run: If True, do not perform the deletion but only print what would be deleted. force: If True, skip confirmation prompt and proceed with deletion. queue: If queue is provided, add the URI to the queue instead of deleting immediately. + session: Optional HTTP session with retry strategy. + timeout: Request timeout in seconds. """ - group_body = fetch_databus_jsonld(databusURI, databus_key) + group_body = fetch_databus_jsonld(databusURI, databus_key, session=session, timeout=timeout) json_dict = json.loads(group_body) artifacts = json_dict.get("databus:hasArtifact", []) @@ -279,14 +334,38 @@ def _delete_group( # Delete all artifacts (which deletes their versions) for artifact_uri in artifact_uris: _delete_artifact( - artifact_uri, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + artifact_uri, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + session=session, + timeout=timeout, ) # Finally, delete the group itself - _delete_resource(databusURI, databus_key, dry_run=dry_run, force=force, queue=queue,manifest_context=manifest_context) + _delete_resource( + databusURI, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + session=session, + timeout=timeout, + ) -def delete(databusURIs: List[str], databus_key: str, dry_run: bool, force: bool, manifest_context=None): +def delete( + databusURIs: List[str], + databus_key: str, + dry_run: bool = False, + force: bool = False, + manifest_context=None, + session: requests.Session | None = None, + timeout: int = 30, +): """Delete a dataset from the databus. Delete a group, artifact, or version identified by the given databus URI. @@ -297,9 +376,15 @@ def delete(databusURIs: List[str], databus_key: str, dry_run: bool, force: bool, databus_key: Databus API key to authenticate the deletion requests. dry_run: If True, will only print what would be deleted without performing actual deletions. force: If True, skip confirmation prompt and proceed with deletion. + session: Optional HTTP session with retry strategy. + timeout: Request timeout in seconds. """ + if session is None: + session = get_http_session() - queue = DeleteQueue(databus_key, manifest_context=manifest_context) + queue = DeleteQueue( + databus_key, manifest_context=manifest_context, session=session, timeout=timeout + ) for databusURI in databusURIs: _host, _account, group, artifact, version, file = ( @@ -309,24 +394,52 @@ def delete(databusURIs: List[str], databus_key: str, dry_run: bool, force: bool, if group == "collections" and artifact is not None: print(f"Deleting collection: {databusURI}") _delete_resource( - databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + databusURI, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + session=session, + timeout=timeout, ) elif file is not None: print(f"Deleting file is not supported via API: {databusURI}") elif version is not None: print(f"Deleting version: {databusURI}") _delete_resource( - databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + databusURI, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + session=session, + timeout=timeout, ) elif artifact is not None: print(f"Deleting artifact and all its versions: {databusURI}") _delete_artifact( - databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + databusURI, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + session=session, + timeout=timeout, ) elif group is not None and group != "collections": print(f"Deleting group and all its artifacts and versions: {databusURI}") _delete_group( - databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + databusURI, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + session=session, + timeout=timeout, ) else: print(f"Deleting {databusURI} is not supported.") diff --git a/databusclient/api/deploy.py b/databusclient/api/deploy.py index 2c8cd08..3ec7a28 100644 --- a/databusclient/api/deploy.py +++ b/databusclient/api/deploy.py @@ -11,6 +11,7 @@ from typing import Dict, List, Optional, Tuple, Union import requests +from databusclient.api.utils import get_http_session _debug = False @@ -176,14 +177,18 @@ def _get_file_stats(distribution_str: str) -> Tuple[Optional[str], Optional[int] return sha256sum, content_length -def _load_file_stats(url: str) -> Tuple[str, int]: +def _load_file_stats( + url: str, session: requests.Session | None = None, timeout: int = 30 +) -> Tuple[str, int]: """Download the file at ``url`` and compute its SHA-256 and length. This is used as a fallback when the caller did not supply checksum/size information in the CLI or metadata file. """ + if session is None: + session = get_http_session() - resp = requests.get(url, timeout=30) + resp = session.get(url, timeout=timeout) if resp.status_code >= 400: raise requests.exceptions.RequestException(response=resp) @@ -463,6 +468,8 @@ def deploy( verify_parts: bool = False, log_level: DeployLogLevel = DeployLogLevel.debug, debug: bool = False, + session: requests.Session | None = None, + timeout: int = 30, ) -> None: """Deploys a dataset to the databus. The endpoint is inferred from the DataID identifier. Parameters @@ -487,11 +494,14 @@ def deploy( except (KeyError, IndexError, TypeError) as e: raise DeployError(f"Invalid dataid structure: {e}") + if session is None: + session = get_http_session() + api_uri = ( base + f"/api/publish?verify-parts={str(verify_parts).lower()}&log-level={log_level.name}" ) - resp = requests.post(api_uri, data=data, headers=headers, timeout=30) + resp = session.post(api_uri, data=data, headers=headers, timeout=timeout) if debug or _debug: try: @@ -517,6 +527,8 @@ def deploy_from_metadata( artifact_version_description: str, license_url: str, apikey: str, + session: requests.Session | None = None, + timeout: int = 30, ) -> None: """ Deploy a dataset from metadata entries. @@ -550,7 +562,7 @@ def deploy_from_metadata( ) print(f"Deploying dataset version: {version_id}") - deploy(dataset, apikey) + deploy(dataset, apikey, session=session, timeout=timeout) print(f"Successfully deployed to {version_id}") print(f"Deployed {len(metadata)} file(s):") diff --git a/databusclient/api/utils.py b/databusclient/api/utils.py index df4f49d..fb13462 100644 --- a/databusclient/api/utils.py +++ b/databusclient/api/utils.py @@ -7,6 +7,50 @@ from typing import Optional, Tuple import hashlib import requests +from requests.adapters import HTTPAdapter +from urllib3.util import Retry + + +def get_http_session( + retries: int = 3, + backoff_factor: float = 0.5, + status_forcelist: Tuple[int, ...] = (429, 500, 502, 503, 504), + allowed_methods: Optional[frozenset] = None, + backoff_jitter: float = 0.1, +) -> requests.Session: + """Create and configure a requests Session with HTTP retry strategy and exponential backoff. + + Args: + retries: Total number of retries to allow. + backoff_factor: Backoff factor to apply between attempts. + status_forcelist: Set of HTTP status codes to force retry on. + allowed_methods: Set of HTTP methods allowed for retry. + backoff_jitter: Jitter to add to backoff delay. + + Returns: + Configured requests.Session object. + """ + session = requests.Session() + kwargs = { + "total": retries, + "backoff_factor": backoff_factor, + "status_forcelist": status_forcelist, + "raise_on_status": False, + } + if allowed_methods is not None: + kwargs["allowed_methods"] = allowed_methods + + try: + kwargs["backoff_jitter"] = backoff_jitter + retry_strategy = Retry(**kwargs) + except TypeError: + kwargs.pop("backoff_jitter", None) + retry_strategy = Retry(**kwargs) + + adapter = HTTPAdapter(max_retries=retry_strategy) + session.mount("http://", adapter) + session.mount("https://", adapter) + return session def get_databus_id_parts_from_file_url( @@ -28,33 +72,36 @@ def get_databus_id_parts_from_file_url( A tuple containing (host, accountId, groupId, artifactId, versionId, fileId). Each element is a string or None if not present. """ - """Split a Databus URI into its six parts. - - The returned tuple is (host, accountId, groupId, artifactId, versionId, fileId). - Missing parts are returned as ``None``. - """ - uri = uri.removeprefix("https://").removeprefix("http://") parts = uri.strip("/").split("/") parts += [None] * (6 - len(parts)) # pad with None if less than 6 parts return tuple(parts[:6]) # return only the first 6 parts -def fetch_databus_jsonld(uri: str, databus_key: str | None = None) -> str: +def fetch_databus_jsonld( + uri: str, + databus_key: str | None = None, + session: requests.Session | None = None, + timeout: int = 30, +) -> str: """Fetch the JSON-LD representation of a Databus resource. Args: uri: Full Databus resource URI. databus_key: Optional API key for protected resources. + session: Optional HTTP session to use for requests. + timeout: Request timeout in seconds. Returns: The response body as a string containing JSON-LD. """ + if session is None: + session = get_http_session() headers = {"Accept": "application/ld+json"} if databus_key is not None: headers["X-API-KEY"] = databus_key - response = requests.get(uri, headers=headers, timeout=30) + response = session.get(uri, headers=headers, timeout=timeout) response.raise_for_status() return response.text diff --git a/databusclient/cli.py b/databusclient/cli.py index 4b12e47..ebbb834 100644 --- a/databusclient/cli.py +++ b/databusclient/cli.py @@ -9,6 +9,7 @@ import databusclient.api.deploy as api_deploy from databusclient.api.delete import delete as api_delete from databusclient.api.download import DownloadAuthError, download as api_download, format_download_connection_error +from databusclient.api.utils import get_http_session from databusclient.manifest.context import ManifestContext from databusclient.manifest.writer import ManifestWriter from databusclient.manifest.replay import ManifestReplayError, replay_manifest, load_manifest @@ -75,6 +76,22 @@ def app(): ) @click.option("--remote", help="rclone remote name (e.g., 'nextcloud')") @click.option("--path", help="Remote path on Nextcloud (e.g., 'datasets/mydataset')") +@click.option( + "--max-retries", + "--retries", + "retries", + default=3, + show_default=True, + type=int, + help="Maximum number of HTTP retries for network requests", +) +@click.option( + "--request-timeout", + default=30, + show_default=True, + type=int, + help="Timeout in seconds for HTTP network requests", +) @click.argument("distributions", nargs=-1) def deploy( version_id, @@ -87,6 +104,8 @@ def deploy( webdav_url, remote, path, + retries: int, + request_timeout: int, distributions: List[str], manifest_path, ): @@ -97,6 +116,8 @@ def deploy( - Upload & deploy via Nextcloud (--webdav-url, --remote, --path) """ + session = api_deploy.get_http_session(retries=retries) + # Sanity checks for conflicting options if metadata_file and any([distributions, webdav_url, remote, path]): raise click.UsageError( @@ -147,7 +168,7 @@ def _write_manifest(): if manifest_context: manifest_context.replay_params["deploy_mode"] = "classic" manifest_context.replay_params["resolved_distributions"] = dataid["@graph"][-1].get("distribution", []) - api_deploy.deploy(dataid=dataid, api_key=apikey) + api_deploy.deploy(dataid=dataid, api_key=apikey, session=session, timeout=request_timeout) if manifest_context: for dist in distributions: url = str(dist).split("|")[0] @@ -170,7 +191,7 @@ def _write_manifest(): manifest_context.replay_params["deploy_mode"] = "metadata" manifest_context.replay_params["resolved_metadata"] = metadata api_deploy.deploy_from_metadata( - metadata, version_id, title, abstract, description, license_url, apikey + metadata, version_id, title, abstract, description, license_url, apikey, session=session, timeout=request_timeout ) if manifest_context: for entry in metadata: @@ -206,7 +227,7 @@ def _write_manifest(): try: metadata = webdav.upload_to_webdav(distributions, remote, path, webdav_url) api_deploy.deploy_from_metadata( - metadata, version_id, title, abstract, description, license_url, apikey + metadata, version_id, title, abstract, description, license_url, apikey, session=session, timeout=request_timeout ) if manifest_context: for entry in metadata: @@ -440,7 +461,31 @@ def download( default=None, help="Write a JSON-LD manifest of this operation to PATH.", ) -def delete(databusuris: List[str], databus_key: str, dry_run: bool, force: bool, manifest_path: str): +@click.option( + "--max-retries", + "--retries", + "retries", + default=3, + show_default=True, + type=int, + help="Maximum number of HTTP retries for network requests", +) +@click.option( + "--request-timeout", + default=30, + show_default=True, + type=int, + help="Timeout in seconds for HTTP network requests", +) +def delete( + databusuris: List[str], + databus_key: str, + dry_run: bool, + force: bool, + manifest_path: str, + retries: int, + request_timeout: int, +): """ Delete a dataset from the databus. @@ -456,6 +501,7 @@ def delete(databusuris: List[str], databus_key: str, dry_run: bool, force: bool, "dry_run": dry_run, }) + session = get_http_session(retries=retries) try: api_delete( databusURIs=databusuris, @@ -463,6 +509,8 @@ def delete(databusuris: List[str], databus_key: str, dry_run: bool, force: bool, dry_run=dry_run, force=force, manifest_context=manifest_context, + session=session, + timeout=request_timeout, ) except Exception as exc: if manifest_context: diff --git a/tests/test_retry.py b/tests/test_retry.py new file mode 100644 index 0000000..5bdabf1 --- /dev/null +++ b/tests/test_retry.py @@ -0,0 +1,117 @@ +"""Tests for HTTP retry session helper and CLI options.""" + +import requests +from click.testing import CliRunner +from requests.adapters import HTTPAdapter + +from databusclient.api.utils import get_http_session, fetch_databus_jsonld +from databusclient.cli import app + + +def test_get_http_session_defaults(): + session = get_http_session() + assert isinstance(session, requests.Session) + + http_adapter = session.adapters.get("http://") + https_adapter = session.adapters.get("https://") + + assert isinstance(http_adapter, HTTPAdapter) + assert isinstance(https_adapter, HTTPAdapter) + assert http_adapter.max_retries.total == 3 + + +def test_get_http_session_custom_retries(): + session = get_http_session(retries=5, backoff_factor=1.0) + http_adapter = session.adapters.get("https://") + + assert http_adapter.max_retries.total == 5 + assert http_adapter.max_retries.backoff_factor == 1.0 + + +def test_get_http_session_with_jitter(): + session = get_http_session(retries=4, backoff_factor=0.5, backoff_jitter=0.2) + http_adapter = session.adapters.get("https://") + assert http_adapter.max_retries.total == 4 + if hasattr(http_adapter.max_retries, "backoff_jitter"): + assert http_adapter.max_retries.backoff_jitter == 0.2 + + +def test_fetch_databus_jsonld_with_custom_session(): + class MockResponse: + status_code = 200 + text = '{"@context": "https://databus.dbpedia.org/context.jsonld"}' + + def raise_for_status(self): + pass + + class MockSession(requests.Session): + def get(self, url, headers=None, timeout=None): + assert timeout == 15 + assert headers["Accept"] == "application/ld+json" + assert headers["X-API-KEY"] == "test_key" + return MockResponse() + + session = MockSession() + result = fetch_databus_jsonld( + "https://databus.dbpedia.org/test", + databus_key="test_key", + session=session, + timeout=15, + ) + assert result == '{"@context": "https://databus.dbpedia.org/context.jsonld"}' + + +def test_cli_delete_force_and_retry_options(monkeypatch): + captured = {} + + def fake_api_delete(**kwargs): + captured.update(kwargs) + + monkeypatch.setattr("databusclient.cli.api_delete", fake_api_delete) + + runner = CliRunner() + result = runner.invoke( + app, + [ + "delete", + "https://databus.dbpedia.org/user/group/artifact/version", + "--databus-key", + "test-key", + "--force", + "--retries", + "5", + "--request-timeout", + "45", + ], + ) + + assert result.exit_code == 0 + assert captured["force"] is True + assert captured["timeout"] == 45 + assert captured["session"].adapters["https://"].max_retries.total == 5 + + +def test_cli_delete_max_retries_alias(monkeypatch): + captured = {} + + def fake_api_delete(**kwargs): + captured.update(kwargs) + + monkeypatch.setattr("databusclient.cli.api_delete", fake_api_delete) + + runner = CliRunner() + result = runner.invoke( + app, + [ + "delete", + "https://databus.dbpedia.org/user/group/artifact/version", + "--databus-key", + "test-key", + "--force", + "--max-retries", + "7", + ], + ) + + assert result.exit_code == 0 + assert captured["session"].adapters["https://"].max_retries.total == 7