From 92b31717f077ebbeddf9dfdcfb0854b23c0dad1e Mon Sep 17 00:00:00 2001 From: gguerrini Date: Wed, 22 Jul 2026 09:18:12 +0200 Subject: [PATCH 1/5] Fixed quota message --- hda/utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hda/utils.py b/hda/utils.py index 6d53e09..fdf9579 100644 --- a/hda/utils.py +++ b/hda/utils.py @@ -32,7 +32,7 @@ def build_quota_hit_message(response: requests.Response) -> str: msg = ( f"{remaining} requests remaining out of {limit}. " - f"Please wait until {datetime.fromtimestamp(int(reset))} " + f"Please wait until {datetime.fromtimestamp(int(reset/1000))} " f"to submit a new request." ) return msg From 398f4634cccb93246b6bb96218552714558a9c68 Mon Sep 17 00:00:00 2001 From: gguerrini Date: Wed, 22 Jul 2026 09:18:47 +0200 Subject: [PATCH 2/5] Removed resume as not supported. Added STAC api. --- docs/source/api.rst | 5 + docs/source/usage.rst | 4 +- hda/api.py | 244 +++++++++++------------------------------- hda/stac.py | 117 ++++++++++++++++++++ 4 files changed, 189 insertions(+), 181 deletions(-) create mode 100644 hda/stac.py diff --git a/docs/source/api.rst b/docs/source/api.rst index 70e4bd4..02862e0 100644 --- a/docs/source/api.rst +++ b/docs/source/api.rst @@ -28,3 +28,8 @@ API .. autoclass:: Client :members: + +.. automodule:: hda.api + +.. autoclass:: StacMixin + :members: \ No newline at end of file diff --git a/docs/source/usage.rst b/docs/source/usage.rst index acb62ed..699d92c 100644 --- a/docs/source/usage.rst +++ b/docs/source/usage.rst @@ -32,7 +32,7 @@ Just by exporting environment variables, the user and the password can be set to from hda import Client, Configuration - conf = Configuration() # By default, values are retrived from the environment + config = Configuration() # By default, values are retrived from the environment client = Client(config=config) A last way of specifying the credentials is by providing the path for an alternative configuration file that has @@ -42,7 +42,7 @@ the same *.hdarc* format: from hda import Client, Configuration - conf = Configuration(path="/custom/config") + config = Configuration(path="/custom/config") client = Client(config=config) While it is not recommended, nothing prohibit to mix those methods. In that case, the precedence rules are: diff --git a/hda/api.py b/hda/api.py index 5121ec0..7051077 100644 --- a/hda/api.py +++ b/hda/api.py @@ -43,6 +43,7 @@ from tqdm import tqdm from hda.utils import build_quota_hit_message, bytes_to_string, convert +from hda.stac import StacMixin BROKER_URL = "https://gateway.prod.wekeo2.eu/hda-broker/" ITEMS_PER_PAGE = 100 @@ -351,7 +352,6 @@ def _download( self, result, download_dir: str = ".", - force=False, to_s3=False, s3_bucket=None, s3_key_prefix="", @@ -360,38 +360,15 @@ def _download( s3_secret_access_key=None, s3_verify_ssl=True, ): - if ( - not to_s3 - and "properties" in result - and "location" in result["properties"] - and "size" in result["properties"] - ): - filename = os.path.basename(result["properties"]["location"]) - size = result["properties"]["size"] - outfile = os.path.join(download_dir, filename) - if os.path.exists(outfile): - outfile_size = os.stat(outfile).st_size - - if size == outfile_size: - logger.debug( - "File {} already exists and has the expected size {}".format( - outfile, size - ) - ) - if force: - logger.debug("Downloading anyway because force keyword is set") - else: - logger.debug("Skipping download, use force=True to download anyway") - return self.client.accept_tac(self.dataset) download_id = self._get_download_id(result) + expected_size = result.get("properties", {}).get("size", 0) self.stream( download_id, - result["properties"]["size"], + expected_size, download_dir, - force=force, to_s3=to_s3, s3_bucket=s3_bucket, s3_key_prefix=s3_key_prefix, @@ -430,7 +407,6 @@ def build_url(result): def download( self, download_dir: str = ".", - force=False, *, to_s3=False, s3_bucket=None, @@ -440,7 +416,7 @@ def download( s3_secret_access_key=None, s3_verify_ssl=True, ): - """Downloads the results into the given download directory or S3 bucket. + """Download the results into the given download directory or S3 bucket. The process is executed concurrently using :py:attr:`hda.api.Client.max_workers` threads. """ @@ -453,7 +429,6 @@ def download( self._download, result, download_dir, - force, to_s3, s3_bucket, s3_key_prefix, @@ -469,6 +444,9 @@ def download( result = future.result() logger.info(f"Successfully downloaded: {result}") except Exception as exc: + print( + f"Download task failed: {exc}, {type(exc)}, {future.__dict__}" + ) logger.error(f"Download task failed: {exc}") @@ -563,6 +541,7 @@ def __init__( self.retry_max = retry_max self.progress = progress self.max_workers = max_workers + self.stac = StacMixin(self) self._session = None self._access_token = None @@ -617,7 +596,7 @@ def is_token_expired(): if is_token_expired(): logger.debug("====== Token expired, renewing") payload = self._get_token() - if 'error' in payload: + if "error" in payload: logger.debug("Token payload: %s", shorten(payload)) raise ConfigurationError(payload["error_description"]) self._access_token = payload["access_token"] @@ -697,6 +676,7 @@ def robust(self, call): def wrapped(*args, **kwargs): tries = 0 + sleep_delay = 10.0 while tries < self.retry_max: try: r = call(*args, **kwargs) @@ -744,7 +724,10 @@ def wrapped(*args, **kwargs): tries += 1 logger.warning("Retrying in %s seconds", self.sleep_max) - time.sleep(self.sleep_max) + time.sleep(sleep_delay) + sleep_delay *= 1.5 + if sleep_delay > self.sleep_max: + sleep_delay = self.sleep_max return r @@ -888,8 +871,6 @@ def _stream_to_local_file( self, response: requests.Response, outfile: str, - mode: str, - current_total: int, content_size: Optional[int], ) -> int: """Streams the response content to a local file.""" @@ -903,9 +884,8 @@ def _stream_to_local_file( disable=not self.progress, leave=False, position=next(self._tqdm_position), - initial=current_total, ) as pbar: - with open(outfile, mode) as f: + with open(outfile, "wb") as f: for chunk in response.iter_content(chunk_size=1024): if chunk: f.write(chunk) @@ -1010,40 +990,6 @@ def _stream_to_s3( return downloaded_in_session - def _handle_resume_logic( - self, - total_downloaded: int, - content_size: Optional[int], - outfile: str, - sleep_delay: float, - ) -> Tuple[int, str, float, Dict[str, str]]: - """Handles logging, sleeping, and setting headers for download resumption. - Return: - - the downloaded size - - the file opening mode - - the new sleep_delay - - the HTTP headers - """ - - logger.error( - f"Download incomplete, downloaded {total_downloaded} byte(s) out of {content_size}" - ) - - logger.warning(f"Sleeping {sleep_delay} seconds") - time.sleep(sleep_delay) - - # Update state for next attempt - mode = "ab" # Append mode - total_downloaded = os.path.getsize(outfile) - sleep_delay *= 1.5 - if sleep_delay > self.sleep_max: - sleep_delay = self.sleep_max - - headers = {"Range": "bytes=%d-" % total_downloaded} - logger.warning("Resuming download at byte %s" % (total_downloaded,)) - - return total_downloaded, mode, sleep_delay, headers - def _finalize_download( self, total_downloaded: int, content_size: Optional[int], start_time: float ) -> None: @@ -1064,9 +1010,8 @@ def _finalize_download( def stream( self, download_id: str, - size: int, + expected_size: int, download_dir: str = ".", - force: bool = False, *, to_s3: bool = False, s3_bucket: Optional[str] = None, @@ -1086,8 +1031,6 @@ def stream( :type size: int :param download_dir: The directory into which the resource must be downloaded. :type download_dir: str, optional - :param force: Whether to override the product if a local file already exists. - :type force: bool, optional :param to_s3: Whether to download the product directly to S3 (needs optional dependencies). :type s3: bool, optional :param s3_bucket: The S3 bucket to stream the product to. @@ -1101,125 +1044,68 @@ def stream( :param s3_verify_ssl: Whether to verify the SSL Certificate. :type s3_verify_ssl: bool """ - # Set loop variables full_url = self.full_url(*[f"dataaccess/download/{download_id}"]) - start_time = time.time() - mode = "wb" - total_downloaded = 0 - sleep_delay = 10.0 - tries = 0 - headers = None - # S3 Setup - s3_client = None - if to_s3: - s3_client = init_s3_client( - s3_bucket, - s3_endpoint, - s3_access_key_id, - s3_secret_access_key, - s3_verify_ssl, - ) - s3_key = None # Will be set after first request - else: - download_dir = os.path.expanduser(download_dir) - os.makedirs(download_dir, exist_ok=True) + response = self.session.head(full_url, verify=self.config.verify) + response.raise_for_status() + + filename = get_filename(response, download_id) + expected_size = get_content_size(response, expected_size) - logger.info(f"Downloading {full_url} ({bytes_to_string(size)})") + start_time = time.time() + total_downloaded = 0 - while tries < self.retry_max: + try: response = self.robust(self.session.get)( full_url, stream=True, verify=self.config.verify, - headers=headers, timeout=self.timeout, ) - try: - response.raise_for_status() - - logger.debug("Headers: %s", response.headers) - filename = get_filename(response, download_id) - content_size = get_content_size(response, size) - - # Local file precheck - if not to_s3: - outfile = os.path.join(download_dir, filename) - - # XXX EXtract this block - if content_size is not None and os.path.exists(outfile): - outfile_size = os.stat(outfile).st_size - if content_size == outfile_size: - logger.debug( - f"File {outfile_size} already exists and has the expected size {outfile_size}" - ) - if force: - logger.debug( - "Downloading anyway because force keyword is set" - ) - else: - logger.debug( - "Skipping download, use force=True to download anyway" - ) - return filename - - # S3 key finalized - if to_s3 and s3_key is None: - s3_key = os.path.join(s3_key_prefix, filename).lstrip("/") - - # Finally, streaming - downloaded_in_session = 0 - if to_s3: - downloaded_in_session = self._stream_to_s3( - response, s3_client, s3_bucket, s3_key, content_size - ) - else: - downloaded_in_session = self._stream_to_local_file( - response, outfile, mode, total_downloaded, content_size - ) + response.raise_for_status() - total_downloaded += downloaded_in_session - - if content_size is None or total_downloaded >= content_size: - size = content_size # Use the accurate size for final checks - break - - except ( - requests.exceptions.RequestException, - RuntimeError, - DownloadSizeError, - ) as e: - logger.error("Download interrupted: %s" % (e,)) - print("Download interrupted: %s" % (e,)) - if tries >= self.retry_max: - # If this was the last attempt, re-raise the error. - raise - - # For connection failures (not partial download), we sleep and retry. - logger.warning("Sleeping %s seconds before retry" % (sleep_delay,)) - time.sleep(sleep_delay) - sleep_delay *= 1.5 - if sleep_delay > self.sleep_max: - sleep_delay = self.sleep_max - continue - except S3InitializeError as e: - # This is not recovable, exit right away - logger.error("Download interrupted: %s" % (e,)) - print("Download interrupted: %s" % (e,)) - break - finally: - response.close() - - if not to_s3: - # Only need resume logic for local files - total_downloaded, mode, sleep_delay, headers = ( - self._handle_resume_logic( - total_downloaded, content_size, outfile, sleep_delay - ) + logger.debug("Headers: %s", response.headers) + + if to_s3: + s3_client = init_s3_client( + s3_bucket, + s3_endpoint, + s3_access_key_id, + s3_secret_access_key, + s3_verify_ssl, + ) + s3_key = os.path.join(s3_key_prefix, filename).lstrip("/") + total_downloaded = self._stream_to_s3( + response, s3_client, s3_bucket, s3_key, expected_size ) else: - # S3 stream is non-resumable at the moment - logger.warning("S3 download was incomplete. Retrying from start.") + download_dir = os.path.expanduser(download_dir) + os.makedirs(download_dir, exist_ok=True) + outfile = os.path.join(download_dir, filename) + total_downloaded = self._stream_to_local_file( + response, outfile, expected_size + ) + + logger.info(f"Downloading {full_url} ({bytes_to_string(expected_size)})") + print(f"Downloading {full_url} ({bytes_to_string(expected_size)})") + + if expected_size is None or total_downloaded >= expected_size: + size = expected_size # Use the accurate size for final checks + + except ( + RuntimeError, + DownloadSizeError, + ) as e: + logger.error("Download interrupted: %s" % (e,)) + print("Download interrupted: %s" % (e,)) + except S3InitializeError as e: + # This is not recovable, exit right away + logger.error("Download interrupted: %s" % (e,)) + print("Download interrupted: %s" % (e,)) + except Exception as e: + print(f"{type(e)} - {e}") + finally: + response.close() self._finalize_download(total_downloaded, size, start_time) diff --git a/hda/stac.py b/hda/stac.py new file mode 100644 index 0000000..2a5ef04 --- /dev/null +++ b/hda/stac.py @@ -0,0 +1,117 @@ +import math +import re +from typing import Iterator +from urllib.parse import parse_qs, urlparse, quote + +ISO_PATTERN = r"\d{4}-\d{2}-\d{2}(?:T\d{2}:\d{2}:\d{2})?Z" +INTERVAL_PATTERN = rf"^{ISO_PATTERN}(?:/{ISO_PATTERN})?$" + +INTERVAL_REGEX = re.compile(INTERVAL_PATTERN) + + +def validate_interval(interval: str) -> bool: + return bool(INTERVAL_REGEX.match(interval)) + + +class Page: + def __init__(self, response, client, items_key): + self.items = response.get(items_key, []) + self.total_available = response.get("numberMatched") + self.number_returned = response.get("numberReturned") + self._links = response.get("links", []) + self._client = client + self._items_key = items_key + + def __str__(self) -> str: + return f"Page {self.current_page} of {self.total_pages}, {self.number_returned} items" + + def __repr__(self) -> str: + return f"" + + @property + def current_page(self) -> int: + """Extracts the page number from the 'self' link.""" + self_link = next((link["href"] for link in self._links if link["rel"] == "self"), "") + query_params = parse_qs(urlparse(self_link).query) + # Default to page 1 if the parameter isn't found + return int(query_params.get("page", [1])[0]) + + @property + def total_pages(self) -> int: + """Calculates total pages based on the fixed limit of 20.""" + if self.total_available == 0 or self.total_available is None: + return 0 + return math.ceil(self.total_available / 20) + + @property + def has_next(self) -> bool: + return any(link["rel"] == "next" for link in self._links) + + def next_page(self) -> "Page": + next_url = next(link["href"] for link in self._links if link["rel"] == "next") + response = self._client.get(next_url) + return Page(response, self._client, self._items_key) + + +class StacMixin: + def __init__(self, client): + self._client = client + + def get_info(self) -> dict: + """Returns the Landing Page (root) metadata.""" + return self._client.get("stac/") + + def get_conformance(self) -> list[str]: + """Returns the list of supported OGC/STAC features.""" + return self._client.get("stac/conformance/") + + def get_collections_page(self, page: int = 1) -> Page: + """Iterates through all available collections (paginated).""" + response = self._client.get(f"stac/collections/?page={page}") + return Page(response, self._client, "collections") + + def get_collection(self, collection_id: str) -> dict: + """Retrieves metadata for a specific collection.""" + return self._client.get("stac/collections/", collection_id) + + def get_items_page(self, collection_id: str, limit: int = 20, page: int = 1) -> Page: + """Iterates through items within a specific collection.""" + response = self._client.get(f"stac/collections/{quote(collection_id)}/items?page={page}&limit={limit}") + return Page(response, self._client, "items") + + def get_item(self, collection_id: str, item_id: str) -> dict: + """Retrieves a single item from a collection.""" + return self._client.get(f"stac/collections/{quote(collection_id)}/items/{quote(item_id)}") + + def search( + self, + *, + collections: list[str] = None, + ids: list[str] = None, + bbox: tuple[float, float, float, float] = None, + interval: str = None, + limit: int = 1, + **kwargs, + ) -> Iterator[dict]: + """ + Cross-collection search. Returns a generator that handles + pagination internally. + + """ + payload = {} + keys = { + "collections": collections, + "ids": ids, + "bbox": bbox, + "datetime": interval, + "limit": limit, + "token": self._client.token, + } + # if not validate_interval(interval): + # raise ValueError("Bad interval format") + + for key, param in keys.items(): + if param: + payload[key] = param + + return self._client.post(payload, "stac/search") From ca6e564997caaa62dcb6ca0ff687c828cc0a165e Mon Sep 17 00:00:00 2001 From: gguerrini Date: Wed, 22 Jul 2026 17:42:47 +0200 Subject: [PATCH 3/5] Added pagination for CDSE adapter --- hda/api.py | 40 ++++++++++++++++++++++++++++------------ 1 file changed, 28 insertions(+), 12 deletions(-) diff --git a/hda/api.py b/hda/api.py index 7051077..11d3f61 100644 --- a/hda/api.py +++ b/hda/api.py @@ -209,7 +209,7 @@ def make_request(self, query): elif self.request_type == RequestType.POST: return self.request(query, self.action) - def run(self, *, query=None, limit=None, items_per_page=100): + def run(self, *, query=None, limit=None, items_per_page=ITEMS_PER_PAGE): if query is None: query = {} @@ -222,18 +222,34 @@ def run(self, *, query=None, limit=None, items_per_page=100): yield from self.yield_result(page, limit) prop = page["properties"] - while prop["startIndex"] < prop["totalResults"]: - if self.returned >= prop["totalResults"]: - return - if limit is not None and self.returned > limit: - return - - params["startIndex"] = prop["startIndex"] + items_per_page - query.update(params) - page = self.make_request(query) - prop = page["properties"] - yield from self.yield_result(page, limit) + if "totalResults" not in prop: + # CDSE Adapter: the total counter is not available, we follow the + # next index to paginate through the results + while prop.get("nextIndex"): + if limit is not None and self.returned > limit: + return + + params["startIndex"] = prop["nextIndex"] + query.update(params) + page = self.make_request(query) + prop = page["properties"] + print(f"{prop=}") + yield from self.yield_result(page, limit) + else: + # Use the regular pagination mechanism + while prop["startIndex"] < prop["totalResults"]: + if self.returned >= prop["totalResults"]: + return + + if limit is not None and self.returned > limit: + return + + params["startIndex"] = prop["startIndex"] + items_per_page + query.update(params) + page = self.make_request(query) + prop = page["properties"] + yield from self.yield_result(page, limit) class SearchPaginator(Paginator): From bc0e9bfeed6c33973ef654e6edd4f162a096d626 Mon Sep 17 00:00:00 2001 From: gguerrini Date: Wed, 29 Jul 2026 12:55:39 +0200 Subject: [PATCH 4/5] Added federated authentication, CDSE adapter and STAC endpoints --- docs/source/changelog.rst | 6 +++++ docs/source/conf.py | 6 ++--- docs/source/index.rst | 12 ++------- docs/source/installation.rst | 5 ++-- docs/source/quickstart.rst | 8 +++--- docs/source/usage.rst | 17 +++++++++--- hda/api.py | 51 ++++++++++++++++++++++-------------- hda/stac.py | 27 +++++++++++-------- tests/test_hda.py | 2 +- 9 files changed, 80 insertions(+), 54 deletions(-) diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst index 56398dd..cf1d8d7 100644 --- a/docs/source/changelog.rst +++ b/docs/source/changelog.rst @@ -1,6 +1,12 @@ Changelog ========= +Version 2.40 +------------ +* Added support for STAC endpoints +* Added federated authentication +* Updated authentication endpoint + Version 2.38 ------------ * Added optional S3 support diff --git a/docs/source/conf.py b/docs/source/conf.py index 957a6f7..b684059 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -12,11 +12,11 @@ # -- Project information project = "WEkEO HDA API Client" -copyright = "2025, ECMWF" +copyright = "2026, ECMWF" author = "ECMWF" -release = "2.30" -version = "2.30" +release = "2.40" +version = "2.40" # -- General configuration diff --git a/docs/source/index.rst b/docs/source/index.rst index 3b2300a..3794665 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -5,19 +5,11 @@ This package provides a fully compliant Python 3 client that can be used to sear HDA is a RESTful interface allowing users to search and download WEkEO datasets. -Documentation about its usage can be found at `WEkEO website `_. +Documentation about its usage can be found at `WEkEO website `_. Check out the :doc:`usage ` section for further information, including how to :doc:`install ` the project. -.. warning:: - Starting from version 2.0, the client supports HDA v2 only. If you need to interact with HDA v1, please use client version 1.14, which is the last release compatible with that API. - - Note that HDA v1 was decommissioned in Q2 2024. - - Although HDA v2 is a complete overhaul of the original API, the client interface remains unchanged. The primary difference is a significantly simplified query format. - When possible, the client will still accept queries written in the legacy format and automatically convert them into the new structure. - Requirements ------------ - Python 3 @@ -33,7 +25,7 @@ Pull requests are welcome. For major changes, please open an issue first to disc Please make sure to update this documentation as appropriate - changes on the interface, version etc. -Licence +License ------- Please refer to LICENSE.txt diff --git a/docs/source/installation.rst b/docs/source/installation.rst index ef7d06f..133ec32 100644 --- a/docs/source/installation.rst +++ b/docs/source/installation.rst @@ -4,9 +4,8 @@ Installation Get your credentials -------------------- -1. If you don't have a WEkEO account, please self register through the WEkEO `registration form `_, then proceed to the step below. - -2. Copy the code below in the file `$HOME/.hdarc` in your Unix/Linux environment. Adapt the following template with the credentials of your WEkEO account: +1. If you don't have a WEkEO account, please self register through the WEkEO `registration form `_, then proceed to the step below. Alternatively, you can use your EUMETSAT or CMEMS credentials. +2. Copy the code below in the file `$HOME/.hdarc` in your Unix/Linux environment. Adapt the following template with your credentials: .. code-block:: ini diff --git a/docs/source/quickstart.rst b/docs/source/quickstart.rst index 291f2a1..f0ce22d 100644 --- a/docs/source/quickstart.rst +++ b/docs/source/quickstart.rst @@ -1,7 +1,7 @@ Quick start =========== -Once the WEkEO HDA API client is installed, it can be used to request data from the datasets listed in the `WEkEO catalogue `_. +Once the WEkEO HDA API client is installed, it can be used to request data from the datasets listed in the `WEkEO catalogue `_. On the WEkEO portal, under **DATA**, each dataset search has a *Show API request* button, it displays the json request to be used. The request can be formatted using the interactive form. The API call must follow the syntax. @@ -18,8 +18,8 @@ The client can be used directly into another python script as in the following e query = { 'dataset_id': 'EO:EUM:DAT:SENTINEL-3:OL_1_EFR___', - 'dtstart': '2023-07-03T13:59:00.000Z', - 'dtend': '2023-07-03T14:03:00.000Z', + 'dtstart': '2026-05-03T13:59:00.000Z', + 'dtend': '2026-05-03T14:03:00.000Z', } matches = c.search(query) print(matches) @@ -30,7 +30,7 @@ The client can be used directly into another python script as in the following e Please refer to the official documentation of the HDA for instructions on how to get the list of the available parameters. .. warning:: - The query format has been streamlined in version 2, but the client still accepts most of the old queries and automatically + The query format has been streamlined since version 2, but the client still accepts most of the old queries and automatically convert them into the new format under the hood, before they are submitted to the API. You might still want to explicitly change the queries to reflect the updated structure. diff --git a/docs/source/usage.rst b/docs/source/usage.rst index 699d92c..5fe565c 100644 --- a/docs/source/usage.rst +++ b/docs/source/usage.rst @@ -52,6 +52,8 @@ While it is not recommended, nothing prohibit to mix those methods. In that case 3. The custom configuration file if no environment is set 4. Finally, the *$HOME/.hdarc* file, which is the default one +Please note that, since version 3.40, the usage of the API is no longer limited to WEkEO users. EUMETSAT and CMEMS credentials are equally accepted. + Advanced client usage --------------------- @@ -89,7 +91,16 @@ Depending on the number of downloads, this can speed up the process, especially This number can be easily changed by specifying a different `max_workers` value for the :class:`hda.api.Client` class. -Keep in mind though the following: +Keep in mind that each WEkEO account has usage quotas. While the numbers are pretty high, it is not recommended to hammer the API for just a small potential speed gain. + +STAC catalogue +~~~~~~~~~~~~~~ + +Besides the search and download endpoints, a whole STAC compatible branch of API is available: + +.. code-block:: python + + c = Client() + c.stac.get_info() -1. As a general rule of thumb, the number of threads should be equal to the number of CPU core -2. Each WEkEO account has usage quotas. While the numbers are pretty high, it is not recommended to hammer the API for just a small potential speed gain +Please refer to the :doc:`api ` section for a description of the provided methods. diff --git a/hda/api.py b/hda/api.py index 11d3f61..606cda3 100644 --- a/hda/api.py +++ b/hda/api.py @@ -26,7 +26,7 @@ import time from enum import Enum from itertools import cycle -from typing import Any, Dict, Optional, Tuple +from typing import Any, Optional from urllib.parse import urljoin try: @@ -42,8 +42,8 @@ import requests from tqdm import tqdm -from hda.utils import build_quota_hit_message, bytes_to_string, convert from hda.stac import StacMixin +from hda.utils import build_quota_hit_message, bytes_to_string, convert BROKER_URL = "https://gateway.prod.wekeo2.eu/hda-broker/" ITEMS_PER_PAGE = 100 @@ -234,7 +234,6 @@ def run(self, *, query=None, limit=None, items_per_page=ITEMS_PER_PAGE): query.update(params) page = self.make_request(query) prop = page["properties"] - print(f"{prop=}") yield from self.yield_result(page, limit) else: # Use the regular pagination mechanism @@ -632,15 +631,29 @@ def _get_token(self): """ def get_new_token(): - data = { + # The gettoken endpoint changed on July 2026 + url = urljoin(self.config.url, "gettoken").replace("hda-broker", "identity") + + base_payload = { "username": self.config.user, "password": self.config.password, } - return requests.post( - urljoin(self.config.url, "gettoken"), - json=data, - verify=self.config.verify, - ) + + # WEkEO authentication includes multiple federated accounts + origins = [None, "eumetsat", "cmems"] + + for origin in origins: + payload = {**base_payload, "origin": origin} if origin else base_payload + response = requests.post( + url, + json=payload, + verify=self.config.verify, + ) + + if response.status_code == 200: + return response + + response.raise_for_status() def refresh_token(): return requests.post( @@ -1026,7 +1039,7 @@ def _finalize_download( def stream( self, download_id: str, - expected_size: int, + size: int, download_dir: str = ".", *, to_s3: bool = False, @@ -1066,7 +1079,7 @@ def stream( response.raise_for_status() filename = get_filename(response, download_id) - expected_size = get_content_size(response, expected_size) + content_size = get_content_size(response, size) start_time = time.time() total_downloaded = 0 @@ -1092,21 +1105,21 @@ def stream( ) s3_key = os.path.join(s3_key_prefix, filename).lstrip("/") total_downloaded = self._stream_to_s3( - response, s3_client, s3_bucket, s3_key, expected_size + response, s3_client, s3_bucket, s3_key, content_size ) else: download_dir = os.path.expanduser(download_dir) os.makedirs(download_dir, exist_ok=True) outfile = os.path.join(download_dir, filename) total_downloaded = self._stream_to_local_file( - response, outfile, expected_size + response, outfile, content_size ) - logger.info(f"Downloading {full_url} ({bytes_to_string(expected_size)})") - print(f"Downloading {full_url} ({bytes_to_string(expected_size)})") + logger.info(f"Downloading {full_url} ({bytes_to_string(content_size)})") + print(f"Downloading {full_url} ({bytes_to_string(content_size)})") - if expected_size is None or total_downloaded >= expected_size: - size = expected_size # Use the accurate size for final checks + if content_size is None or total_downloaded >= content_size: + size = content_size # Use the accurate size for final checks except ( RuntimeError, @@ -1123,9 +1136,9 @@ def stream( finally: response.close() - self._finalize_download(total_downloaded, size, start_time) + self._finalize_download(total_downloaded, content_size, start_time) - if total_downloaded < size: + if total_downloaded < content_size: # Final check failure, should only happen if retry_max was hit raise DownloadSizeError( f"Download failed: {total_downloaded} byte(s) out of {size} (missing {size - total_downloaded})" diff --git a/hda/stac.py b/hda/stac.py index 2a5ef04..cee6074 100644 --- a/hda/stac.py +++ b/hda/stac.py @@ -1,7 +1,7 @@ import math import re from typing import Iterator -from urllib.parse import parse_qs, urlparse, quote +from urllib.parse import parse_qs, quote, urlparse ISO_PATTERN = r"\d{4}-\d{2}-\d{2}(?:T\d{2}:\d{2}:\d{2})?Z" INTERVAL_PATTERN = rf"^{ISO_PATTERN}(?:/{ISO_PATTERN})?$" @@ -31,7 +31,9 @@ def __repr__(self) -> str: @property def current_page(self) -> int: """Extracts the page number from the 'self' link.""" - self_link = next((link["href"] for link in self._links if link["rel"] == "self"), "") + self_link = next( + (link["href"] for link in self._links if link["rel"] == "self"), "" + ) query_params = parse_qs(urlparse(self_link).query) # Default to page 1 if the parameter isn't found return int(query_params.get("page", [1])[0]) @@ -74,14 +76,20 @@ def get_collection(self, collection_id: str) -> dict: """Retrieves metadata for a specific collection.""" return self._client.get("stac/collections/", collection_id) - def get_items_page(self, collection_id: str, limit: int = 20, page: int = 1) -> Page: + def get_items_page( + self, collection_id: str, limit: int = 20, page: int = 1 + ) -> Page: """Iterates through items within a specific collection.""" - response = self._client.get(f"stac/collections/{quote(collection_id)}/items?page={page}&limit={limit}") + response = self._client.get( + f"stac/collections/{quote(collection_id)}/items?page={page}&limit={limit}" + ) return Page(response, self._client, "items") def get_item(self, collection_id: str, item_id: str) -> dict: """Retrieves a single item from a collection.""" - return self._client.get(f"stac/collections/{quote(collection_id)}/items/{quote(item_id)}") + return self._client.get( + f"stac/collections/{quote(collection_id)}/items/{quote(item_id)}" + ) def search( self, @@ -94,9 +102,7 @@ def search( **kwargs, ) -> Iterator[dict]: """ - Cross-collection search. Returns a generator that handles - pagination internally. - + Cross-collection search. """ payload = {} keys = { @@ -105,10 +111,9 @@ def search( "bbox": bbox, "datetime": interval, "limit": limit, - "token": self._client.token, } - # if not validate_interval(interval): - # raise ValueError("Bad interval format") + if not validate_interval(interval): + raise ValueError("Bad interval format") for key, param in keys.items(): if param: diff --git a/tests/test_hda.py b/tests/test_hda.py index 9cb503f..d0dae58 100644 --- a/tests/test_hda.py +++ b/tests/test_hda.py @@ -151,7 +151,7 @@ def test_download_s3_raises_importerror_when_missing_s3(monkeypatch, fresh_hda_a with pytest.raises(ImportError): client.stream( download_id="abc", - size=100, + expected_size=100, download_dir=".", to_s3=True, s3_bucket="my-bucket", From 239ac83a56eab42fa971f976f913eb129b7c9168 Mon Sep 17 00:00:00 2001 From: gguerrini Date: Wed, 29 Jul 2026 14:58:18 +0200 Subject: [PATCH 5/5] Fixed tests --- hda/api.py | 19 ++++++++++--------- hda/utils.py | 2 +- tests/test_hda.py | 3 +-- 3 files changed, 12 insertions(+), 12 deletions(-) diff --git a/hda/api.py b/hda/api.py index 606cda3..9ea6192 100644 --- a/hda/api.py +++ b/hda/api.py @@ -1073,6 +1073,16 @@ def stream( :param s3_verify_ssl: Whether to verify the SSL Certificate. :type s3_verify_ssl: bool """ + s3_client = None + if to_s3: + s3_client = init_s3_client( + s3_bucket, + s3_endpoint, + s3_access_key_id, + s3_secret_access_key, + s3_verify_ssl, + ) + full_url = self.full_url(*[f"dataaccess/download/{download_id}"]) response = self.session.head(full_url, verify=self.config.verify) @@ -1096,13 +1106,6 @@ def stream( logger.debug("Headers: %s", response.headers) if to_s3: - s3_client = init_s3_client( - s3_bucket, - s3_endpoint, - s3_access_key_id, - s3_secret_access_key, - s3_verify_ssl, - ) s3_key = os.path.join(s3_key_prefix, filename).lstrip("/") total_downloaded = self._stream_to_s3( response, s3_client, s3_bucket, s3_key, content_size @@ -1131,8 +1134,6 @@ def stream( # This is not recovable, exit right away logger.error("Download interrupted: %s" % (e,)) print("Download interrupted: %s" % (e,)) - except Exception as e: - print(f"{type(e)} - {e}") finally: response.close() diff --git a/hda/utils.py b/hda/utils.py index fdf9579..66e6b7d 100644 --- a/hda/utils.py +++ b/hda/utils.py @@ -32,7 +32,7 @@ def build_quota_hit_message(response: requests.Response) -> str: msg = ( f"{remaining} requests remaining out of {limit}. " - f"Please wait until {datetime.fromtimestamp(int(reset/1000))} " + f"Please wait until {datetime.fromtimestamp(int(reset)/1000)} " f"to submit a new request." ) return msg diff --git a/tests/test_hda.py b/tests/test_hda.py index d0dae58..e5b555c 100644 --- a/tests/test_hda.py +++ b/tests/test_hda.py @@ -110,7 +110,6 @@ def test_hda_e2e(): } matches = c.search(r, limit=10) - print(matches) assert len(matches.results) == 10, matches @@ -151,7 +150,7 @@ def test_download_s3_raises_importerror_when_missing_s3(monkeypatch, fresh_hda_a with pytest.raises(ImportError): client.stream( download_id="abc", - expected_size=100, + size=100, download_dir=".", to_s3=True, s3_bucket="my-bucket",