diff --git a/datacommons_client/README.md b/datacommons_client/README.md index f4563663..c53fe700 100644 --- a/datacommons_client/README.md +++ b/datacommons_client/README.md @@ -1,6 +1,6 @@ # Data Commons Python API -This is a Python library for accessing data in the Data Commons Graph. +This is a Python library for accessing data in the Data Commons Graph via the V2 REST API (`node`, `observation`, `resolve`) and the SDMX 3.0 REST API (`data`, `availability`). To get started, install this package from pip. @@ -15,10 +15,61 @@ with the optional Pandas dependency. pip install "datacommons-client[Pandas]" ``` -Once the package is installed, import `datacommons_client`. +Once the package is installed, import `datacommons_client` and initialize `DataCommonsClient`: ```python -import datacommons_client as dc +from datacommons_client import DataCommonsClient + +client = DataCommonsClient(api_key="YOUR_API_KEY") + +# V2 Observation query +observations = client.observation.fetch( + variable_dcids="Count_Person", + entity_dcids=["country/USA"], +) + +# SDMX 3.0 Data query (returns SDMX-CSV string, or use fetch_data_as_dataframe for a Pandas DataFrame) +csv_data = client.sdmx.fetch_data( + variable="Count_Person", + constraints={"observationAbout": "country/USA"}, +) +df = client.sdmx.fetch_data_as_dataframe( + variable="Count_Person", + constraints={"observationAbout": "country/USA"}, +) + +# SDMX 3.0 Availability query (returns parsed SDMX-JSON, or use fetch_available_values for {component_id: [values]}) +availability = client.sdmx.fetch_availability( + component_id="provenance", + variable="Count_Person", +) +available_values = client.sdmx.fetch_available_values( + component_id="provenance", + variable="Count_Person", +) +``` + +## Connecting to a Private Data Commons Platform Instance + +To query a private [Data Commons Platform](https://github.com/datacommonsorg/datacommons) instance protected by IAM (such as an authenticated Cloud Run service), pass the instance URL along with your `Authorization` header and set `validate_instance=False`: + +```python +from datacommons_client import DataCommonsClient + +client = DataCommonsClient( + url="https://your-dcp-service-url.a.run.app/core/api/v2", + headers={"Authorization": f"Bearer {id_token}"}, + validate_instance=False, +) + +# Query V2 or SDMX 3.0 endpoints on the private instance +node_data = client.node.fetch(node_dcids="FinancialTrade", expression="->name") +sdmx_csv = client.sdmx.fetch_data( + variable="FinancialTrade", + constraints={"sourceCountry": "country/FRA"}, +) ``` For more detail on getting started with the API, please visit . + + diff --git a/datacommons_client/__init__.py b/datacommons_client/__init__.py index ad654341..ac16ac8c 100644 --- a/datacommons_client/__init__.py +++ b/datacommons_client/__init__.py @@ -10,7 +10,14 @@ from datacommons_client.endpoints.node import NodeEndpoint from datacommons_client.endpoints.observation import ObservationEndpoint from datacommons_client.endpoints.resolve import ResolveEndpoint +from datacommons_client.endpoints.sdmx import ApiLayout +from datacommons_client.endpoints.sdmx import build_query_params +from datacommons_client.endpoints.sdmx import parse_filters +from datacommons_client.endpoints.sdmx import SdmxClient +from datacommons_client.endpoints.sdmx import SdmxEndpoint from datacommons_client.utils.context import use_api_key +from datacommons_client.utils.error_handling import SdmxAPIError +from datacommons_client.utils.error_handling import SdmxClientError __all__ = [ "DataCommonsClient", @@ -18,5 +25,12 @@ "NodeEndpoint", "ObservationEndpoint", "ResolveEndpoint", + "SdmxEndpoint", + "SdmxClient", + "ApiLayout", + "SdmxClientError", + "SdmxAPIError", + "parse_filters", + "build_query_params", "use_api_key", ] diff --git a/datacommons_client/client.py b/datacommons_client/client.py index cb8b90d1..03e506c0 100644 --- a/datacommons_client/client.py +++ b/datacommons_client/client.py @@ -1,9 +1,11 @@ +from collections.abc import Mapping from typing import Literal, Optional from datacommons_client.endpoints.base import API from datacommons_client.endpoints.node import NodeEndpoint from datacommons_client.endpoints.observation import ObservationEndpoint from datacommons_client.endpoints.resolve import ResolveEndpoint +from datacommons_client.endpoints.sdmx import SdmxEndpoint from datacommons_client.models.observation import ObservationDate from datacommons_client.utils.dataframes import add_entity_names_to_observations_dataframe from datacommons_client.utils.dataframes import add_property_constraints_to_observations_dataframe @@ -20,7 +22,8 @@ class DataCommonsClient: """ A client for interacting with the Data Commons API. - This class provides convenient access to the V2 Data Commons API endpoints. + This class provides convenient access to the V2 Data Commons API endpoints + and the SDMX 3.0 Data and Availability APIs. Attributes: api (API): An instance of the API class that handles requests. @@ -29,6 +32,7 @@ class DataCommonsClient: observation (ObservationEndpoint): Handles observation-related queries, allowing retrieval of statistical observations associated with entities, variables, and dates (e.g., GDP of California in 2010). resolve (ResolveEndpoint): Manages resolution queries to find different DCIDs for entities. + sdmx (SdmxEndpoint): Queries the SDMX 3.0 Data and Availability REST APIs. """ @@ -37,7 +41,9 @@ def __init__(self, *, dc_instance: Optional[str] = "datacommons.org", url: Optional[str] = None, - surface_header_value: Optional[str] = None): + surface_header_value: Optional[str] = None, + headers: Optional[Mapping[str, str]] = None, + validate_instance: bool = True): """ Initializes the DataCommonsClient. @@ -46,6 +52,9 @@ def __init__(self, custom DC instances do not currently require an API key. dc_instance (Optional[str]): The Data Commons instance to use. Defaults to "datacommons.org". url (Optional[str]): A custom, fully resolved URL for the Data Commons API. Defaults to None. + surface_header_value (Optional[str]): Optional Data Commons surface header identifier. + headers (Optional[Mapping[str, str]]): Optional additional HTTP headers (e.g. Authorization bearer token). + validate_instance (bool): Whether to validate the target instance URL during initialization. Defaults to True. """ # If a fully resolved URL is provided, and the default dc_instance is used, # ignore that default value @@ -56,12 +65,15 @@ def __init__(self, self.api = API(api_key=api_key, dc_instance=dc_instance, url=url, - surface_header_value=surface_header_value) + surface_header_value=surface_header_value, + headers=headers, + validate_instance=validate_instance) # Create instances of the endpoints self.node = NodeEndpoint(api=self.api) self.observation = ObservationEndpoint(api=self.api) self.resolve = ResolveEndpoint(api=self.api) + self.sdmx = SdmxEndpoint(api=self.api) def _find_filter_facet_ids( self, diff --git a/datacommons_client/endpoints/base.py b/datacommons_client/endpoints/base.py index ff4adfdc..9e81da11 100644 --- a/datacommons_client/endpoints/base.py +++ b/datacommons_client/endpoints/base.py @@ -1,8 +1,10 @@ +from collections.abc import Mapping import re from typing import Any, Dict, Optional from datacommons_client.utils.context import _API_KEY_CONTEXT_VAR from datacommons_client.utils.request_handling import check_instance_is_valid +from datacommons_client.utils.request_handling import CUSTOM_DC_V2 from datacommons_client.utils.request_handling import post_request from datacommons_client.utils.request_handling import resolve_instance_url @@ -21,6 +23,9 @@ def __init__( dc_instance: Optional[str] = None, url: Optional[str] = None, surface_header_value: Optional[str] = None, + *, + headers: Optional[Mapping[str, str]] = None, + validate_instance: bool = True, ): """ Initializes the API instance. @@ -33,7 +38,9 @@ def __init__( of the API is required (for local development, for example). If provided, dc_instance` should not be provided. surface_header_value: indicates which DC surface (MCP server, etc.) makes a call to the python library. - If the call originated internally, this is null and we pass in "clientlib-python" as the surface header + If the call originated internally, this is null and we pass in "clientlib-python" as the surface header + headers: Optional additional HTTP headers (e.g. Authorization bearer tokens) to include in requests. + validate_instance: Whether to probe the instance URL during initialization. Defaults to True. Raises: ValueError: If both `dc_instance` and `url` are provided. @@ -45,14 +52,33 @@ def __init__( dc_instance = "datacommons.org" if url is not None: - # Use the given URL directly (strip trailing slash) - self.base_url = check_instance_is_valid(url.rstrip("/"), api_key=api_key) + clean_url = url.rstrip("/") + if not validate_instance: + self.base_url = clean_url + elif headers: + self.base_url = check_instance_is_valid(clean_url, + api_key=api_key, + headers=dict(headers)) + else: + self.base_url = check_instance_is_valid(clean_url, api_key=api_key) else: - # Resolve from dc_instance - self.base_url = resolve_instance_url(dc_instance) + clean_dc = (dc_instance.replace("https://", "").replace("http://", + "").rstrip("/")) + if not validate_instance: + if clean_dc == "datacommons.org": + self.base_url = resolve_instance_url("datacommons.org") + else: + self.base_url = f"https://{clean_dc}{CUSTOM_DC_V2}" + elif headers: + self.base_url = resolve_instance_url(dc_instance, + api_key=api_key, + headers=dict(headers)) + else: + self.base_url = resolve_instance_url(dc_instance, api_key=api_key) self.headers = self.build_headers(surface_header_value=surface_header_value, - api_key=api_key) + api_key=api_key, + custom_headers=headers) def __repr__(self) -> str: """Returns a readable representation of the API object. @@ -62,7 +88,9 @@ def __repr__(self) -> str: Returns: str: A string representation of the API object. """ - has_auth = " (Authenticated)" if "X-API-Key" in self.headers else "" + has_auth = (" (Authenticated)" if any( + k.lower() in ("x-api-key", "authorization") for k in self.headers) else + "") return f"" def post(self, @@ -108,15 +136,21 @@ def post(self, all_pages=all_pages, next_token=next_token) - def build_headers(self, - surface_header_value: Optional[str], - api_key: Optional[str] = None) -> dict[str, str]: + def build_headers( + self, + surface_header_value: Optional[str], + api_key: Optional[str] = None, + custom_headers: Optional[Mapping[str, str]] = None, + ) -> dict[str, str]: """Build request headers for API requests. Includes JSON content type. If an API key is provided, add it as `X-API-Key`. Args: self: the API, which includes API key and surface header if available + surface_header_value: Optional surface header identifier. + api_key: Optional API key for X-API-Key header. + custom_headers: Optional custom headers to merge into the request headers. Returns: A dictionary of headers for the request. @@ -131,6 +165,9 @@ def build_headers(self, if surface_header_value: headers["x-surface"] = surface_header_value + if custom_headers: + headers.update(custom_headers) + return headers diff --git a/datacommons_client/endpoints/sdmx.py b/datacommons_client/endpoints/sdmx.py new file mode 100644 index 00000000..c821e5c7 --- /dev/null +++ b/datacommons_client/endpoints/sdmx.py @@ -0,0 +1,493 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Endpoint client for the Data Commons SDMX 3.0 Data and Availability REST APIs.""" + +from collections.abc import Mapping, Sequence +from enum import Enum +from http import HTTPStatus +import io +from typing import Any, Dict, Optional + +import requests + +from datacommons_client.endpoints.base import API +from datacommons_client.endpoints.base import Endpoint +from datacommons_client.utils.context import _API_KEY_CONTEXT_VAR +from datacommons_client.utils.decorators import requires_pandas +from datacommons_client.utils.error_handling import SdmxAPIError +from datacommons_client.utils.error_handling import SdmxClientError + +try: + import pandas as pd +except ImportError: + pd = None + +# The SDMX context, agency, resource and version are fixed for Data Commons; +# the key is always the `*` wildcard. +DATAFLOW: str = "DC/DF_OBS/1.0.0/*" + +DEFAULT_TIMEOUT_SECONDS: int = 60 + + +class ApiLayout(Enum): + """How a Data Commons deployment lays out its REST API paths. + + The public Data Commons API serves each API at the root of the host, while + a self-hosted DCP instance serves them beneath a `/core/api` prefix. + """ + + ROOT = "root" + CORE_API = "core_api" + + +# Path beneath which each deployment flavor serves the SDMX API. +_API_ROOTS = { + ApiLayout.ROOT: "sdmx/v3", + ApiLayout.CORE_API: "core/api/sdmx/v3", +} + + +def parse_filters(filters: Sequence[str]) -> dict[str, list[str]]: + """Parses `key=value` filter strings into a constraint mapping. + + Repeating a key accumulates its values, which the API combines with OR. + """ + if isinstance(filters, str): + raise TypeError( + "filters must be a sequence of 'key=value' strings, not a str.") + + constraints: dict[str, list[str]] = {} + for item in filters: + key, separator, value = item.partition("=") + if not separator or not key.strip() or not value.strip(): + raise ValueError(f"Invalid filter '{item}'. Expected key=value " + "(for example: observationAbout=country/FRA).") + constraints.setdefault(key.strip(), []).append(value.strip()) + return constraints + + +def build_query_params( + variable: str, + constraints: Optional[Mapping[str, Any]] = None, +) -> dict[str, str]: + """Builds the `c[]=` query parameters for an SDMX request.""" + if not variable or not variable.strip(): + raise ValueError("variable must not be empty.") + + params: dict[str, str] = {} + for component, value in (constraints or {}).items(): + if value is None: + continue + if isinstance(value, str): + values = [value] + elif isinstance(value, (Sequence, set)): + values = sorted(value) if isinstance(value, set) else list(value) + else: + values = [str(value)] + cleaned = [str(v).strip() for v in values if str(v).strip()] + if cleaned: + params[f"c[{component.strip()}]"] = ",".join(cleaned) + + params["c[variableMeasured]"] = variable.strip() + return params + + +def extract_availability_values( + payload: Mapping[str, Any],) -> dict[str, list[str]]: + """Extracts `{component_id: [values]}` from an SDMX-JSON Availability response. + + Unpacks the nested `data.dataConstraints[*].cubeRegions[*].keyValues[*]` + structure returned by the SDMX 3.0 Availability endpoint into a flat mapping + of component IDs to their available string values. + """ + result: dict[str, list[str]] = {} + data = payload.get("data") if isinstance(payload, Mapping) else None + if not isinstance(data, Mapping): + return result + + for constraint in data.get("dataConstraints") or (): + if not isinstance(constraint, Mapping): + continue + for region in constraint.get("cubeRegions") or (): + if not isinstance(region, Mapping) or not region.get("include", True): + continue + for item in region.get("keyValues") or region.get("components") or (): + if not isinstance(item, Mapping) or not item.get("include", True): + continue + comp_id = item.get("id") + if not isinstance(comp_id, str) or not comp_id: + continue + values_bucket = result.setdefault(comp_id, []) + for val in item.get("values") or (): + v = val.get("value") if isinstance(val, Mapping) else val + if v is not None and str(v) not in values_bucket: + values_bucket.append(str(v)) + return result + + +def _resolve_sdmx_host_and_layout( + raw_base_url: str, + preferred_layout: Optional[ApiLayout] = None, + headers: Optional[Mapping[str, str]] = None, +) -> tuple[str, ApiLayout]: + """Resolves the host origin URL and initial SDMX layout from an API base URL. + + `API.base_url` typically ends with `/v2` (public Data Commons) or + `/core/api/v2` (custom DCP instance), whereas SDMX endpoints are served at + `/sdmx/v3` and `/core/api/sdmx/v3` respectively. + """ + base = raw_base_url.rstrip("/") + if base.endswith("/core/api/sdmx/v3"): + host = base[:-len("/core/api/sdmx/v3")].rstrip("/") + inferred = ApiLayout.CORE_API + elif base.endswith("/sdmx/v3"): + host = base[:-len("/sdmx/v3")].rstrip("/") + inferred = ApiLayout.ROOT + elif base.endswith("/core/api/v2"): + host = base[:-len("/core/api/v2")].rstrip("/") + inferred = ApiLayout.CORE_API + elif base.endswith("/core/api"): + host = base[:-len("/core/api")].rstrip("/") + inferred = ApiLayout.CORE_API + elif base.endswith("/v2"): + host = base[:-len("/v2")].rstrip("/") + inferred = ApiLayout.ROOT + else: + host = base + has_bearer_auth = bool(headers and + any(k.lower() == "authorization" for k in headers)) + inferred = ApiLayout.CORE_API if has_bearer_auth else ApiLayout.ROOT + + return host, (preferred_layout or inferred) + + +class SdmxEndpoint(Endpoint): + """Queries the SDMX 3.0 Data and Availability APIs of a Data Commons endpoint.""" + + def __init__( + self, + api: API, + *, + session: Optional[requests.Session] = None, + timeout: int = DEFAULT_TIMEOUT_SECONDS, + preferred_layout: Optional[ApiLayout] = None, + ) -> None: + """Initializes the SdmxEndpoint instance. + + Args: + api: An API instance providing the environment and authentication configuration. + session: Optional pre-configured `requests.Session` (used in tests). + timeout: Per-request timeout in seconds. + preferred_layout: Optional layout hint (`ApiLayout.ROOT` or `ApiLayout.CORE_API`). + """ + super().__init__(endpoint="sdmx/v3", api=api) + self._session = session or requests.Session() + self._timeout = timeout + self._preferred_layout = preferred_layout + self._api_root: Optional[str] = None + + def __repr__(self) -> str: + """Returns a readable representation of the SdmxEndpoint object.""" + return f"" + + def post(self, + payload: dict[str, Any], + all_pages: bool = True, + next_token: Optional[str] = None) -> Dict[str, Any]: + """SDMX endpoints are queried via GET rather than POST.""" + raise NotImplementedError( + "SDMX endpoints only support GET requests via fetch_data() and " + "fetch_availability().") + + @property + def base_url(self) -> str: + """Returns the host base URL of the endpoint being queried.""" + host, _ = _resolve_sdmx_host_and_layout( + self.api.base_url, + self._preferred_layout, + self.api.headers, + ) + return host + + def fetch_data( + self, + variable: str, + constraints: Optional[Mapping[str, str | Sequence[str]]] = None, + *, + response_format: str = "csv", + log: bool = True, + multi_entity: bool = True, + accept: Optional[str] = None, + ) -> str: + """Fetches statistical observations for a variable via the SDMX 3.0 Data API. + + Args: + variable: The statistical variable measured (e.g. `Count_Person`). + constraints: Dimension and attribute filters to apply. + response_format: SDMX response format; defaults to SDMX-CSV (`"csv"`). + log: Request server-side SDMX execution logs (`X-Log-SDMX` header). + multi_entity: Query across multi-entity schemas (`X-Use-Multi-Entity-Schema` header). + accept: Optional `Accept` header override. + + Returns: + The raw response body, SDMX-CSV by default. + """ + params = build_query_params(variable, constraints) + if response_format: + params["format"] = response_format + + response = self._get( + f"data/dataflow/{DATAFLOW}", + params, + log=log, + multi_entity=multi_entity, + accept=accept, + ) + return response.text + + get_data = fetch_data + + @requires_pandas + def fetch_data_as_dataframe( + self, + variable: str, + constraints: Optional[Mapping[str, str | Sequence[str]]] = None, + *, + log: bool = True, + multi_entity: bool = True, + accept: Optional[str] = None, + ) -> "pd.DataFrame": + """Fetches statistical observations via SDMX-CSV and returns a pandas DataFrame. + + Args: + variable: The statistical variable measured (e.g. `Count_Person`). + constraints: Dimension and attribute filters to apply. + log: Request server-side SDMX execution logs (`X-Log-SDMX` header). + multi_entity: Query across multi-entity schemas (`X-Use-Multi-Entity-Schema` header). + accept: Optional `Accept` header override. + + Returns: + A `pandas.DataFrame` containing the SDMX-CSV rows and columns. + """ + csv_text = self.fetch_data( + variable, + constraints, + response_format="csv", + log=log, + multi_entity=multi_entity, + accept=accept, + ) + if not csv_text or not csv_text.strip(): + return pd.DataFrame() + return pd.read_csv(io.StringIO(csv_text)) + + def fetch_availability( + self, + component_id: str, + variable: str, + constraints: Optional[Mapping[str, str | Sequence[str]]] = None, + *, + log: bool = True, + multi_entity: bool = True, + accept: Optional[str] = None, + ) -> dict[str, Any] | str: + """Queries the values available for a dimension or attribute via the SDMX 3.0 Availability API. + + Args: + component_id: The component to inspect (e.g. `provenance`, `unit`). + variable: The statistical variable measured. + constraints: Dimension and attribute filters to apply. + log: Request server-side SDMX execution logs (`X-Log-SDMX` header). + multi_entity: Query across multi-entity schemas (`X-Use-Multi-Entity-Schema` header). + accept: Optional `Accept` header override. + + Returns: + The parsed SDMX-JSON structure, or the raw body if it is not JSON. + """ + if not component_id or not component_id.strip(): + raise ValueError("component_id must not be empty.") + + response = self._get( + f"availability/dataflow/{DATAFLOW}/{component_id.strip()}", + build_query_params(variable, constraints), + log=log, + multi_entity=multi_entity, + accept=accept, + ) + + if "json" not in response.headers.get("Content-Type", ""): + return response.text + try: + return response.json() + except ValueError: + return response.text + + get_availability = fetch_availability + + def fetch_available_values( + self, + component_id: str, + variable: str, + constraints: Optional[Mapping[str, str | Sequence[str]]] = None, + *, + log: bool = True, + multi_entity: bool = True, + accept: Optional[str] = None, + ) -> dict[str, list[str]]: + """Queries available dimension/attribute values and returns `{component_id: [values]}`. + + Convenience wrapper around `fetch_availability()` that unpacks the nested + SDMX-JSON `dataConstraints[*].cubeRegions[*].keyValues[*]` payload. + + Args: + component_id: The component to inspect (e.g. `provenance`, `unit`, or `*`). + variable: The statistical variable measured. + constraints: Dimension and attribute filters to apply. + log: Request server-side SDMX execution logs (`X-Log-SDMX` header). + multi_entity: Query across multi-entity schemas (`X-Use-Multi-Entity-Schema` header). + accept: Optional `Accept` header override. + + Returns: + A dictionary mapping each returned component ID to its list of available values. + """ + payload = self.fetch_availability( + component_id, + variable, + constraints, + log=log, + multi_entity=multi_entity, + accept=accept, + ) + if not isinstance(payload, Mapping): + return {} + return extract_availability_values(payload) + + def _get( + self, + resource: str, + params: Mapping[str, str], + *, + log: bool, + multi_entity: bool, + accept: Optional[str], + ) -> requests.Response: + """Fetches an SDMX resource, discovering which API root the endpoint uses.""" + base_headers = { + k: v for k, v in self.api.headers.items() if k.lower() != "content-type" + } + ctx_api_key = _API_KEY_CONTEXT_VAR.get() + if ctx_api_key: + base_headers["X-API-Key"] = ctx_api_key + + headers = { + "X-Log-SDMX": str(log).lower(), + "X-Use-Multi-Entity-Schema": str(multi_entity).lower(), + **base_headers, + } + if accept: + headers["Accept"] = accept + + response = None + for api_root in self._candidate_api_roots(): + response = self._send(f"{api_root}/{resource}", params, headers) + if response.status_code != HTTPStatus.NOT_FOUND: + if response.ok: + self._api_root = api_root + break + + return self._checked(response) + + def _candidate_api_roots(self) -> tuple[str, ...]: + """Returns the API roots to try, most likely first. + + A DCP instance serves the SDMX API under `/core/api` while the public + Data Commons API serves it at the root. The endpoint's preferred layout + is only a hint, so the other root is retained as a fallback: a 404 from + the first root transparently retries against the second. A valid SDMX + query never returns 404, so the status unambiguously signals that the + endpoint uses the other layout. + """ + if self._api_root: + return (self._api_root,) + + _, layout = _resolve_sdmx_host_and_layout( + self.api.base_url, + self._preferred_layout, + self.api.headers, + ) + preferred = _API_ROOTS[layout] + fallbacks = (root for root in _API_ROOTS.values() if root != preferred) + return (preferred, *fallbacks) + + def _send( + self, + path: str, + params: Mapping[str, str], + headers: Mapping[str, str], + ) -> requests.Response: + """Performs a single authenticated GET request.""" + url = f"{self.base_url}/{path}" + try: + return self._session.get(url, + params=params, + headers=headers, + timeout=self._timeout) + except requests.RequestException as e: + raise SdmxClientError( + f"Could not reach the Data Commons endpoint at {url}: {e}") from e + + @staticmethod + def _checked(response: requests.Response) -> requests.Response: + """Returns the response, raising `SdmxAPIError` for error statuses.""" + if response.ok: + return response + raise SdmxAPIError(response.status_code, + _error_message(response), + response=response) + + +SdmxClient = SdmxEndpoint + + +def _error_message(response: requests.Response) -> str: + """Extracts the most informative error message from a failed response. + + Falls back to the HTTP reason phrase so that responses with blank or + uninformative bodies still describe the failure. + """ + message = "" + try: + payload = response.json() + except ValueError: + body = response.text.strip() + # HTML bodies, such as a proxy or load balancer error page, are too + # noisy to surface verbatim; leave them to the reason phrase fallback. + if not body.startswith("<"): + message = body + else: + if isinstance(payload, dict): + err_field = payload.get("error") + if isinstance(err_field, dict): + message = str( + payload.get("message") or err_field.get("message") or + err_field.get("status") or "") + else: + message = str(payload.get("message") or err_field or "") + # Surface the raw payload when it has content but no known field. + if not message.strip() and payload: + message = response.text + elif payload: + message = str(payload) + + return message.strip() or response.reason or "Unknown error" diff --git a/datacommons_client/tests/endpoints/test_base.py b/datacommons_client/tests/endpoints/test_base.py index eb81c71a..635313c6 100644 --- a/datacommons_client/tests/endpoints/test_base.py +++ b/datacommons_client/tests/endpoints/test_base.py @@ -23,7 +23,7 @@ def test_api_initialization_default(mock_check_instance, mock_resolve_instance): "Content-Type": "application/json", "x-surface": "clientlib-python" } - mock_resolve_instance.assert_called_once_with("datacommons.org") + mock_resolve_instance.assert_called_once_with("datacommons.org", api_key=None) @patch( @@ -55,7 +55,8 @@ def test_api_initialization_with_dc_instance(mock_resolve_instance_url): "Content-Type": "application/json", "x-surface": "clientlib-python" } - mock_resolve_instance_url.assert_called_once_with("custom-instance") + mock_resolve_instance_url.assert_called_once_with("custom-instance", + api_key=None) @patch( @@ -289,3 +290,34 @@ def test_endpoint_repr(mock_check_instance): expected_repr = ">" assert repr(endpoint) == expected_repr + + +@patch("datacommons_client.endpoints.base.check_instance_is_valid") +def test_api_initialization_validate_instance_false_with_url( + mock_check_instance): + """Tests that validate_instance=False skips check_instance_is_valid and merges custom headers.""" + api = API( + url="https://custom.example.com/core/api/v2/", + headers={"Authorization": "Bearer test-token"}, + validate_instance=False, + ) + assert api.base_url == "https://custom.example.com/core/api/v2" + assert api.headers["Authorization"] == "Bearer test-token" + assert repr(api) == ( + "") + mock_check_instance.assert_not_called() + + +@patch("datacommons_client.endpoints.base.resolve_instance_url") +def test_api_initialization_validate_instance_false_with_dc_instance( + mock_resolve_instance_url): + """Tests that validate_instance=False with custom dc_instance constructs the URL without network validation.""" + api = API( + dc_instance="https://custom.example.com/", + headers={"authorization": "Bearer test-token"}, + validate_instance=False, + ) + assert api.base_url == "https://custom.example.com/core/api/v2" + assert repr(api) == ( + "") + mock_resolve_instance_url.assert_not_called() diff --git a/datacommons_client/tests/endpoints/test_sdmx_endpoint.py b/datacommons_client/tests/endpoints/test_sdmx_endpoint.py new file mode 100644 index 00000000..30ea075e --- /dev/null +++ b/datacommons_client/tests/endpoints/test_sdmx_endpoint.py @@ -0,0 +1,439 @@ +import copy +from http import HTTPStatus +import json +import pickle +from typing import Any +from unittest.mock import MagicMock + +import pytest +import requests + +from datacommons_client import DataCommonsClient +from datacommons_client import use_api_key +from datacommons_client.endpoints.base import API +from datacommons_client.endpoints.sdmx import ApiLayout +from datacommons_client.endpoints.sdmx import build_query_params +from datacommons_client.endpoints.sdmx import parse_filters +from datacommons_client.endpoints.sdmx import SdmxAPIError +from datacommons_client.endpoints.sdmx import SdmxClient +from datacommons_client.endpoints.sdmx import SdmxClientError +from datacommons_client.endpoints.sdmx import SdmxEndpoint +from datacommons_client.utils.error_handling import APIError +from datacommons_client.utils.error_handling import DataCommonsError + +_CSV = "STRUCTURE,OBS_VALUE\ndataflow,100\n" +_AVAILABILITY = {"data": {"dataConstraints": [{"id": "DF_OBS_AVAILABILITY"}]}} + +_ROOT_DATA_URL = "https://api.datacommons.org/sdmx/v3/data/dataflow/DC/DF_OBS/1.0.0/*" +_CORE_DATA_URL = ( + "https://dc-service-xyz.run.app/core/api/sdmx/v3/data/dataflow/DC/DF_OBS/1.0.0/*" +) + + +@pytest.fixture +def public_api() -> API: + return API(api_key="test-key") + + +@pytest.fixture +def instance_api() -> API: + return API( + url="https://dc-service-xyz.run.app/core/api/v2", + headers={"Authorization": "Bearer test-token"}, + validate_instance=False, + ) + + +@pytest.fixture +def make_response(): + """Builds a mock `requests.Response`.""" + + def _make( + status_code: int = 200, + text: str = "", + *, + json_body: Any = None, + content_type: str = "text/csv", + reason: str = "", + ) -> MagicMock: + response = MagicMock(spec=requests.Response) + response.status_code = status_code + response.ok = status_code < HTTPStatus.BAD_REQUEST + response.reason = reason or HTTPStatus(status_code).phrase + response.headers = {"Content-Type": content_type} + if json_body is not None: + response.text = json.dumps(json_body) + response.json.return_value = json_body + else: + response.text = text + response.json.side_effect = ValueError("not JSON") + return response + + return _make + + +@pytest.fixture +def make_session(): + """Builds a mock `requests.Session` returning `responses` in order.""" + + def _make(*responses: MagicMock) -> MagicMock: + session = MagicMock(spec=requests.Session) + session.get.side_effect = list(responses) + return session + + return _make + + +class TestParseFilters: + + def test_parses_key_value_pairs(self): + assert parse_filters(["a=1", "b=2"]) == {"a": ["1"], "b": ["2"]} + + def test_groups_repeated_keys(self): + assert parse_filters(["a=1", "a=2"]) == {"a": ["1", "2"]} + + def test_strips_surrounding_whitespace(self): + assert parse_filters([" a = 1 "]) == {"a": ["1"]} + + def test_keeps_equals_signs_in_values(self): + assert parse_filters(["a=x=y"]) == {"a": ["x=y"]} + + def test_rejects_bare_string_input(self): + with pytest.raises(TypeError, match="must be a sequence"): + parse_filters("a=1") + + @pytest.mark.parametrize("bad", ["novalue", "=1", " =1", "a=", "a= "]) + def test_rejects_malformed_filters(self, bad): + with pytest.raises(ValueError, match="Invalid filter"): + parse_filters([bad]) + + +class TestBuildQueryParams: + + def test_includes_the_variable(self): + assert build_query_params("Count_Person") == { + "c[variableMeasured]": "Count_Person" + } + + def test_rejects_empty_variable(self): + with pytest.raises(ValueError, match="variable must not be empty"): + build_query_params(" ") + + def test_renders_constraints_as_component_params(self): + params = build_query_params("V", {"observationAbout": "country/FRA"}) + assert params["c[observationAbout]"] == "country/FRA" + + def test_joins_multiple_values_with_commas(self): + params = build_query_params("V", {"provenance": ["a", "b"]}) + assert params["c[provenance]"] == "a,b" + + def test_joins_set_values_with_commas(self): + params = build_query_params("V", {"provenance": {"b", "a"}}) + assert params["c[provenance]"] == "a,b" + + def test_ignores_empty_or_none_constraints(self): + params = build_query_params("V", {"unit": [], "scalingFactor": None}) + assert params == {"c[variableMeasured]": "V"} + + +class TestSdmxEndpoint: + + def test_sdmx_client_is_alias_for_sdmx_endpoint(self): + assert SdmxClient is SdmxEndpoint + + def test_errors_inherit_from_api_and_datacommons_error(self): + err = SdmxAPIError(404, "Not found") + assert isinstance(err, SdmxClientError) + assert isinstance(err, APIError) + assert isinstance(err, DataCommonsError) + assert pickle.loads(pickle.dumps(err)).status_code == 404 + assert copy.deepcopy(err).message == "Not found" + + def test_repr_and_post_not_implemented(self, public_api): + endpoint = SdmxEndpoint(public_api) + assert repr(endpoint) == ( + ">") + with pytest.raises(NotImplementedError): + endpoint.post(payload={}) + + def test_fetch_data_returns_the_response_body(self, public_api, make_response, + make_session): + session = make_session(make_response(text=_CSV)) + endpoint = SdmxEndpoint(public_api, session=session) + + assert endpoint.fetch_data("Count_Person") == _CSV + + def test_get_data_alias_returns_the_response_body(self, public_api, + make_response, + make_session): + session = make_session(make_response(text=_CSV)) + endpoint = SdmxEndpoint(public_api, session=session) + + assert endpoint.get_data("Count_Person") == _CSV + + def test_fetch_data_sends_auth_headers_and_params(self, public_api, + make_response, + make_session): + session = make_session(make_response(text=_CSV)) + + SdmxEndpoint(public_api, session=session).fetch_data( + "Count_Person", {"observationAbout": "country/USA"}) + + url, kwargs = session.get.call_args.args[0], session.get.call_args.kwargs + assert url == _ROOT_DATA_URL + assert kwargs["headers"]["X-API-Key"] == "test-key" + assert kwargs["headers"]["x-surface"] == "clientlib-python" + assert "Content-Type" not in kwargs["headers"] + assert kwargs["headers"]["X-Log-SDMX"] == "true" + assert kwargs["headers"]["X-Use-Multi-Entity-Schema"] == "true" + assert kwargs["params"]["c[observationAbout]"] == "country/USA" + assert kwargs["params"]["format"] == "csv" + + def test_use_api_key_context_override(self, public_api, make_response, + make_session): + session = make_session(make_response(text=_CSV)) + endpoint = SdmxEndpoint(public_api, session=session) + + with use_api_key("override-key"): + endpoint.fetch_data("Count_Person") + + headers = session.get.call_args.kwargs["headers"] + assert headers["X-API-Key"] == "override-key" + + def test_header_flags_can_be_disabled(self, public_api, make_response, + make_session): + session = make_session(make_response(text=_CSV)) + + SdmxEndpoint(public_api, + session=session).fetch_data("V", + log=False, + multi_entity=False, + accept="application/json") + + headers = session.get.call_args.kwargs["headers"] + assert headers["X-Log-SDMX"] == "false" + assert headers["X-Use-Multi-Entity-Schema"] == "false" + assert headers["Accept"] == "application/json" + + def test_fetch_availability_parses_json(self, public_api, make_response, + make_session): + session = make_session( + make_response(json_body=_AVAILABILITY, content_type="application/json")) + endpoint = SdmxEndpoint(public_api, session=session) + + assert endpoint.fetch_availability("provenance", + "Count_Person") == _AVAILABILITY + + def test_fetch_availability_rejects_empty_component(self, public_api): + endpoint = SdmxEndpoint(public_api) + with pytest.raises(ValueError, match="component_id must not be empty"): + endpoint.fetch_availability(" ", "Count_Person") + + def test_fetch_availability_falls_back_to_raw_text(self, public_api, + make_response, + make_session): + session = make_session( + make_response(text="not json", content_type="text/plain")) + endpoint = SdmxEndpoint(public_api, session=session) + + assert endpoint.get_availability("provenance", "V") == "not json" + + def test_fetch_available_values_unpacks_cube_regions(self, public_api, + make_response, + make_session): + availability_payload = { + "data": { + "dataConstraints": [{ + "id": + "DF_OBS_AVAILABILITY", + "cubeRegions": [{ + "include": + True, + "keyValues": [ + { + "id": "provenance", + "include": True, + "values": ["dc/base/CensusPEP", "dc/base/WHO"], + }, + { + "id": "TIME_PERIOD", + "include": True, + "values": ["2020", "2021"], + }, + ], + }], + }] + } + } + session = make_session( + make_response(json_body=availability_payload, + content_type="application/json")) + endpoint = SdmxEndpoint(public_api, session=session) + + values = endpoint.fetch_available_values("*", "Count_Person") + assert values == { + "provenance": ["dc/base/CensusPEP", "dc/base/WHO"], + "TIME_PERIOD": ["2020", "2021"], + } + + def test_fetch_data_as_dataframe(self, public_api, make_response, + make_session): + session = make_session(make_response(text=_CSV)) + endpoint = SdmxEndpoint(public_api, session=session) + + df = endpoint.fetch_data_as_dataframe("Count_Person") + assert list(df.columns) == ["STRUCTURE", "OBS_VALUE"] + assert len(df) == 1 + assert df.iloc[0]["OBS_VALUE"] == 100 + + def test_instance_api_prefers_the_core_api_root(self, instance_api, + make_response, make_session): + session = make_session(make_response(text=_CSV)) + + SdmxEndpoint(instance_api, session=session).fetch_data("V") + + assert session.get.call_args.args[0] == _CORE_DATA_URL + assert (session.get.call_args.kwargs["headers"]["Authorization"] == + "Bearer test-token") + + def test_bare_host_with_authorization_header_prefers_core_api( + self, make_response, make_session): + api = API( + url="https://dc-service-xyz.run.app", + headers={"Authorization": "Bearer test-token"}, + validate_instance=False, + ) + session = make_session(make_response(text=_CSV)) + + SdmxEndpoint(api, session=session).fetch_data("V") + + assert session.get.call_args.args[0] == _CORE_DATA_URL + + def test_bare_host_with_preferred_layout_core_api(self, make_response, + make_session): + api = API(url="https://dc-service-xyz.run.app", validate_instance=False) + session = make_session(make_response(text=_CSV)) + + SdmxEndpoint(api, session=session, + preferred_layout=ApiLayout.CORE_API).fetch_data("V") + + assert session.get.call_args.args[0] == _CORE_DATA_URL + + def test_a_404_retries_against_the_other_api_root(self, instance_api, + make_response, + make_session): + session = make_session(make_response(404), make_response(text=_CSV)) + endpoint = SdmxEndpoint(instance_api, session=session) + + assert endpoint.fetch_data("V") == _CSV + + attempted = [call.args[0] for call in session.get.call_args_list] + assert attempted == [ + _CORE_DATA_URL, + "https://dc-service-xyz.run.app/sdmx/v3/data/dataflow/DC/DF_OBS/1.0.0/*", + ] + + def test_the_discovered_api_root_is_reused(self, instance_api, make_response, + make_session): + session = make_session( + make_response(404), + make_response(text=_CSV), + make_response(text=_CSV), + ) + endpoint = SdmxEndpoint(instance_api, session=session) + + endpoint.fetch_data("V") + endpoint.fetch_data("V") + + # Only the first query pays for the discovery attempt. + assert session.get.call_count == 3 + + def test_transient_500_does_not_cache_api_root(self, instance_api, + make_response, make_session): + session = make_session( + make_response(500, reason="Server Error"), + make_response(404), + make_response(text=_CSV), + ) + endpoint = SdmxEndpoint(instance_api, session=session) + + with pytest.raises(SdmxAPIError): + endpoint.fetch_data("V") + + assert endpoint._api_root is None + assert endpoint.fetch_data("V") == _CSV + assert endpoint._api_root == "sdmx/v3" + + def test_a_404_from_every_root_is_reported(self, public_api, make_response, + make_session): + session = make_session(make_response(404, reason="Not Found"), + make_response(404)) + endpoint = SdmxEndpoint(public_api, session=session) + + with pytest.raises(SdmxAPIError) as excinfo: + endpoint.fetch_data("V") + + assert excinfo.value.status_code == 404 + + def test_api_errors_surface_the_server_message(self, public_api, + make_response, make_session): + session = make_session( + make_response(401, json_body={"message": "API key not valid"})) + endpoint = SdmxEndpoint(public_api, session=session) + + with pytest.raises(SdmxAPIError, match="API key not valid") as excinfo: + endpoint.fetch_data("V") + + assert excinfo.value.status_code == 401 + + def test_api_errors_extract_nested_error_message(self, public_api, + make_response, make_session): + session = make_session( + make_response( + 401, + json_body={ + "error": { + "code": 401, + "message": "Nested auth failure", + } + }, + )) + endpoint = SdmxEndpoint(public_api, session=session) + + with pytest.raises(SdmxAPIError, match="Nested auth failure") as excinfo: + endpoint.fetch_data("V") + + assert excinfo.value.message == "Nested auth failure" + + def test_api_errors_fall_back_to_the_reason_phrase(self, public_api, + make_response, + make_session): + session = make_session(make_response(500, text="", reason="Server Error")) + endpoint = SdmxEndpoint(public_api, session=session) + + with pytest.raises(SdmxAPIError, match="Server Error"): + endpoint.fetch_data("V") + + def test_html_error_bodies_are_not_echoed(self, public_api, make_response, + make_session): + session = make_session( + make_response(502, text="gateway", reason="Bad Gateway")) + endpoint = SdmxEndpoint(public_api, session=session) + + with pytest.raises(SdmxAPIError, match="Bad Gateway"): + endpoint.fetch_data("V") + + def test_network_failures_are_wrapped(self, public_api, make_session): + session = make_session() + session.get.side_effect = requests.ConnectionError("refused") + endpoint = SdmxEndpoint(public_api, session=session) + + with pytest.raises(SdmxClientError, match="Could not reach"): + endpoint.fetch_data("V") + + def test_datacommons_client_exposes_sdmx_endpoint(self): + client = DataCommonsClient(api_key="test-key") + assert isinstance(client.sdmx, SdmxEndpoint) + assert client.sdmx.api is client.api + assert client.sdmx.base_url == "https://api.datacommons.org" diff --git a/datacommons_client/tests/test_client.py b/datacommons_client/tests/test_client.py index 221befff..d93ad924 100644 --- a/datacommons_client/tests/test_client.py +++ b/datacommons_client/tests/test_client.py @@ -471,3 +471,15 @@ def test_use_api_key_with_node_fetch_place_ancestors(mock_post_request): client.node.fetch_place_ancestors(place_dcids=["geoId/07"]) _, kwargs = mock_post_request.call_args assert kwargs["headers"]["X-API-Key"] == "context-key" + + +def test_datacommons_client_with_custom_headers_and_validate_instance_false(): + """Tests DataCommonsClient initialization with custom headers and validate_instance=False.""" + client = DataCommonsClient( + url="https://dc-service-xyz.run.app/core/api/v2", + headers={"Authorization": "Bearer tok"}, + validate_instance=False, + ) + assert client.api.base_url == "https://dc-service-xyz.run.app/core/api/v2" + assert client.api.headers["Authorization"] == "Bearer tok" + assert client.sdmx.base_url == "https://dc-service-xyz.run.app" diff --git a/datacommons_client/utils/error_handling.py b/datacommons_client/utils/error_handling.py index 4a7f89ce..cf844255 100644 --- a/datacommons_client/utils/error_handling.py +++ b/datacommons_client/utils/error_handling.py @@ -87,3 +87,41 @@ class NoDataForPropertyError(DataCommonsError): """Raised when there is no data that meets the specified property filters.""" default_message = "No available data for the specified property filters." + + +class SdmxClientError(APIError): + """Base exception for SDMX client operations.""" + + default_message = "An error occurred while querying the SDMX API." + + def __init__( + self, + message: Optional[str] = None, + response: Optional[Response] = None, + ) -> None: + resolved_message = message or self.default_message + super().__init__(response=response, message=resolved_message) + self.message = resolved_message + + def __str__(self) -> str: + return str(self.args[0]) + + +class SdmxAPIError(SdmxClientError): + """Raised when an SDMX endpoint returns an HTTP error status.""" + + def __init__( + self, + status_code: int, + message: str, + response: Optional[Response] = None, + ) -> None: + super().__init__( + message=f"SDMX API returned HTTP {status_code}: {message}", + response=response, + ) + self.status_code = status_code + self.message = message + + def __reduce__(self): + return (self.__class__, (self.status_code, self.message, self.response)) diff --git a/datacommons_client/utils/request_handling.py b/datacommons_client/utils/request_handling.py index 79f2cb80..07a92d19 100644 --- a/datacommons_client/utils/request_handling.py +++ b/datacommons_client/utils/request_handling.py @@ -14,8 +14,11 @@ CUSTOM_DC_V2: str = "/core/api/v2" -def check_instance_is_valid(instance_url: str, - api_key: str | None = None) -> str: +def check_instance_is_valid( + instance_url: str, + api_key: str | None = None, + headers: Optional[Dict[str, str]] = None, +) -> str: """Check that the given instance URL points to a valid Data Commons instance. This function attempts a GET request against a known node in Data Commons to @@ -27,6 +30,7 @@ def check_instance_is_valid(instance_url: str, Args: instance_url: The Data Commons instance URL to validate. api_key: Optional API key for authentication. + headers: Optional additional HTTP headers to include in the validation request. Returns: The validated instance URL. @@ -38,12 +42,12 @@ def check_instance_is_valid(instance_url: str, # Test URL for a known node in Data Commons test_url = f"{instance_url}/node?nodes=country%2FGTM&property=->name" - headers = {} + request_headers = dict(headers) if headers else {} if api_key: - headers["X-API-Key"] = api_key + request_headers["X-API-Key"] = api_key try: - response = requests.get(test_url, headers=headers) + response = requests.get(test_url, headers=request_headers) response.raise_for_status() except requests.exceptions.RequestException as exc: raise InvalidDCInstanceError(exc.response) from exc @@ -56,7 +60,11 @@ def check_instance_is_valid(instance_url: str, return instance_url -def resolve_instance_url(dc_instance: str) -> str: +def resolve_instance_url( + dc_instance: str, + api_key: str | None = None, + headers: Optional[Dict[str, str]] = None, +) -> str: """Resolve the base API URL for a given Data Commons instance. If the instance is `datacommons.org`, the default URL is returned. Otherwise, @@ -64,12 +72,15 @@ def resolve_instance_url(dc_instance: str) -> str: Args: dc_instance: The identifier or domain of the Data Commons instance. + api_key: Optional API key for authentication during instance validation. + headers: Optional additional HTTP headers for instance validation. Returns: The resolved base API URL. """ # if https or http included in the string, remove it - dc_instance = dc_instance.replace("https://", "").replace("http://", "") + dc_instance = (dc_instance.replace("https://", "").replace("http://", + "").rstrip("/")) # If the instance is the default, return the base URL if dc_instance == "datacommons.org": @@ -77,6 +88,10 @@ def resolve_instance_url(dc_instance: str) -> str: # Otherwise, validate the custom instance URL url = f"https://{dc_instance}{CUSTOM_DC_V2}" + if headers: + return check_instance_is_valid(url, api_key=api_key, headers=headers) + if api_key is not None: + return check_instance_is_valid(url, api_key=api_key) return check_instance_is_valid(url)