Source code for httk.serve.optimade.client

"""Synchronous, read-only discovery for remote OPTIMADE services.

This module establishes lossless schema snapshots and strict definition-IRI
recognition. Query construction and paginated execution live in
``remote_query`` and are imported lazily by :meth:`OptimadeStore.searcher`.
"""

import json
import re
from collections.abc import Mapping
from dataclasses import dataclass
from decimal import Decimal
from threading import RLock
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Protocol, Self, TypeGuard, cast
from urllib.parse import parse_qsl, quote, urlsplit

from httk.core import (
    OptimadeDocument,
    OptimadeEntryBinding,
    OptimadeResource,
    OptimadeSchemaSnapshot,
    known_optimade_entry_bindings,
    load_entry_type_schema,
    optimade_entry_binding,
    redact_optimade_url,
)

if TYPE_CHECKING:
    from .remote_query import RemoteSearcher


class _HTTPClient(Protocol):
    def get(self, url: str) -> object: ...


@dataclass(frozen=True, slots=True)
class _AllAdvertised:
    """Identity sentinel selecting every field advertised by a remote schema."""

    def __repr__(self) -> str:
        return "ALL_ADVERTISED"


[docs] ALL_ADVERTISED = _AllAdvertised()
_URL_TOKEN = re.compile(r"(?:https?://|/|\?)[^\s'\"<>]+") _SENSITIVE_QUERY_KEYS = frozenset({"access_token", "api_key", "apikey", "token", "key"}) _VERSION_COMPONENT = r"(?:0|[1-9][0-9]*)" _EXPLICIT_VERSION = re.compile(rf"^v({_VERSION_COMPONENT})(?:\.({_VERSION_COMPONENT})(?:\.({_VERSION_COMPONENT}))?)?$") _VERSION_LIKE = re.compile(r"^v[0-9]") _VERSION_MAJOR = re.compile(r"(?:0|[1-9][0-9]*)$") _API_VERSION = re.compile( rf"^({_VERSION_COMPONENT})\.({_VERSION_COMPONENT})\.({_VERSION_COMPONENT})" r"(?:-[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?$" ) _SUPPORTED_API_MAJORS = frozenset({1}) def _safe_detail(value: str) -> str: """Redact complete absolute or slash-relative URL tokens in diagnostics.""" def redact(match: re.Match[str]) -> str: token = match.group() if token.startswith("?"): keys = {name.casefold() for name, _value in parse_qsl(token[1:], keep_blank_values=True)} if keys.isdisjoint(_SENSITIVE_QUERY_KEYS): return token return redact_optimade_url(token) return _URL_TOKEN.sub(redact, value)
[docs] class OptimadeClientError(RuntimeError): """Base class for safe, client-side OPTIMADE failures."""
[docs] class OptimadeTransportError(OptimadeClientError): """The HTTP client could not complete a request.""" def __init__(self, source_url: str, detail: str) -> None:
[docs] self.source_url = redact_optimade_url(source_url)
[docs] self.detail = _safe_detail(detail)
super().__init__(f"OPTIMADE transport request failed for {self.source_url}: {self.detail}")
[docs] class OptimadeHTTPError(OptimadeClientError): """A remote endpoint returned a non-success HTTP status.""" def __init__(self, source_url: str, status_code: int, detail: str | None = None) -> None:
[docs] self.source_url = redact_optimade_url(source_url)
[docs] self.status_code = status_code
[docs] self.detail = None if detail is None else _safe_detail(detail)
message = f"OPTIMADE request to {self.source_url} returned HTTP {status_code}" if self.detail: message += f": {self.detail}" super().__init__(message)
[docs] class OptimadeErrorDocumentError(OptimadeHTTPError): """A non-success response supplied a parseable OPTIMADE ``errors`` document."""
[docs] class OptimadeDiscoveryError(OptimadeClientError): """An ``/info`` document was malformed or inconsistent for discovery.""" def __init__(self, source_url: str, detail: str) -> None:
[docs] self.source_url = redact_optimade_url(source_url)
[docs] self.detail = _safe_detail(detail)
super().__init__(f"Malformed OPTIMADE discovery response from {self.source_url}: {self.detail}")
[docs] class OptimadeVersionNegotiationError(OptimadeClientError): """An OPTIMADE base URL could not be negotiated to a supported API version.""" def __init__(self, source_url: str, detail: str) -> None:
[docs] self.source_url = redact_optimade_url(source_url)
[docs] self.detail = _safe_detail(detail)
super().__init__(f"OPTIMADE API version negotiation failed for {self.source_url}: {self.detail}")
def _frozen_mapping(values: Mapping[str, str]) -> Mapping[str, str]: return MappingProxyType(dict(values)) @dataclass(frozen=True, slots=True)
[docs] class RemoteEntryType: """One immutable remote entry endpoint discovered from ``/info``. ``name`` is solely the service's transport endpoint name. Semantic recognition is intentionally represented by ``binding`` and is derived exclusively from definition IRIs. """
[docs] name: str
[docs] definition_id: str | None
[docs] schema: OptimadeSchemaSnapshot
[docs] property_iris: Mapping[str, str]
[docs] property_names: Mapping[str, str]
[docs] property_types: Mapping[str, tuple[str, str | None]]
[docs] advertised_properties: tuple[str, ...]
[docs] default_response_properties: tuple[str, ...]
[docs] sortable_properties: tuple[str, ...]
[docs] binding: OptimadeEntryBinding | None
[docs] backend: type
def __post_init__(self) -> None: object.__setattr__(self, "property_iris", _frozen_mapping(self.property_iris)) object.__setattr__(self, "property_names", _frozen_mapping(self.property_names)) object.__setattr__(self, "property_types", MappingProxyType(dict(self.property_types))) object.__setattr__(self, "advertised_properties", tuple(self.advertised_properties)) object.__setattr__(self, "default_response_properties", tuple(self.default_response_properties)) object.__setattr__(self, "sortable_properties", tuple(self.sortable_properties))
def _parse_json(document: OptimadeDocument, *, label: str) -> Mapping[str, Any]: try: decoded = json.loads(document.text, parse_float=Decimal, parse_int=int) except json.JSONDecodeError as exc: raise OptimadeDiscoveryError(document.source_url, f"{label} is not valid JSON: {exc.msg}") from exc if not isinstance(decoded, dict): raise OptimadeDiscoveryError(document.source_url, f"{label} root must be a JSON object") return decoded def _mapping(value: object, *, source_url: str, label: str) -> Mapping[str, Any]: if not isinstance(value, dict): raise OptimadeDiscoveryError(source_url, f"{label} must be a JSON object") return value def _nonempty_string(value: object, *, source_url: str, label: str) -> str: if not isinstance(value, str) or not value.strip(): raise OptimadeDiscoveryError(source_url, f"{label} must be a nonempty string") return value def _is_definition_iri(value: object) -> TypeGuard[str]: """Return whether *value* is a minimally well-formed absolute IRI.""" if not isinstance(value, str) or not value or value != value.strip(): return False try: return bool(urlsplit(value).scheme) except ValueError: return False def _error_detail(text: str) -> str | None: """Extract one useful but non-secret detail from an OPTIMADE error body.""" try: root = json.loads(text) except (TypeError, json.JSONDecodeError): return None if not isinstance(root, dict): return None errors = root.get("errors") if not isinstance(errors, list) or not errors: return None first = errors[0] if not isinstance(first, dict): return None for key in ("detail", "title"): value = first.get(key) if isinstance(value, str) and value: return _safe_detail(value) return None
[docs] class OptimadeStore: """A synchronous read-only OPTIMADE service connection with eager discovery.""" def __init__( self, base_url: str, *, client: object | None = None, page_limit: int = 100, max_pages: int = 10_000, allow_cross_origin_pagination: bool = False, response_fields: object | None = None, ) -> None: self._requested_transport_base_url = self._normalise_base_url(base_url)
[docs] self.requested_base_url = redact_optimade_url(self._requested_transport_base_url)
explicit_major = self._explicit_version_major(self._requested_transport_base_url) if explicit_major is not None and explicit_major not in _SUPPORTED_API_MAJORS: raise OptimadeVersionNegotiationError( self._requested_transport_base_url, f"explicit API major version {explicit_major} is unsupported; supported major is 1", ) self._transport_base_url = self._requested_transport_base_url
[docs] self.base_url = redact_optimade_url(self._transport_base_url)
[docs] self.page_limit = self._positive_int(page_limit, "page_limit")
[docs] self.max_pages = self._positive_int(max_pages, "max_pages")
if not isinstance(allow_cross_origin_pagination, bool): raise TypeError("allow_cross_origin_pagination must be a bool")
[docs] self.allow_cross_origin_pagination = allow_cross_origin_pagination
[docs] self.response_fields = response_fields
self._lock = RLock() self._closed = False self._owned_client = client is None if client is None: import httpx client = httpx.Client() self._client = client self._entry_types: tuple[RemoteEntryType, ...] = () self._entry_types_by_name: Mapping[str, RemoteEntryType] = MappingProxyType({})
[docs] self.api_version: str | None = None
try: if explicit_major is None: self._transport_base_url = self._negotiate_base_url() self.base_url = redact_optimade_url(self._transport_base_url) api_version, entry_types, by_name = self._discover() except Exception: if self._owned_client: self._close_owned_client_after_failed_construction() raise self.api_version = api_version self._entry_types = entry_types self._entry_types_by_name = by_name @staticmethod def _positive_int(value: int, name: str) -> int: if not isinstance(value, int) or isinstance(value, bool) or value <= 0: raise ValueError(f"{name} must be a positive integer") return value @staticmethod def _normalise_base_url(base_url: str) -> str: if not isinstance(base_url, str) or not base_url.strip(): raise ValueError("base_url must be a nonempty absolute HTTP(S) URL") candidate = base_url.strip() try: split = urlsplit(candidate) except ValueError as exc: raise ValueError("base_url must be a valid absolute HTTP(S) URL") from exc if split.scheme not in ("http", "https") or not split.netloc or split.query or split.fragment: raise ValueError("base_url must be an absolute HTTP(S) URL without query or fragment") return candidate.rstrip("/") @staticmethod def _explicit_version_major(base_url: str) -> int | None: """Return the explicitly requested API major, if the final path segment is versioned.""" final_segment = urlsplit(base_url).path.rsplit("/", 1)[-1] match = _EXPLICIT_VERSION.fullmatch(final_segment) if match is not None: return int(match.group(1)) if _VERSION_LIKE.match(final_segment): raise OptimadeVersionNegotiationError( base_url, "final path segment is a malformed explicit API version; version suffixes are not supported", ) return None def _negotiate_base_url(self) -> str: """Select httk's first supported major from an unversioned ``/versions`` response.""" versions_url = self._requested_transport_base_url + "/versions" advertised_majors = self._parse_versions(self._get(versions_url), versions_url) for major in advertised_majors: if major in _SUPPORTED_API_MAJORS: return self._requested_transport_base_url + f"/v{major}" raise OptimadeVersionNegotiationError( versions_url, "server does not advertise a supported API major version (supported major is 1)", ) @staticmethod def _parse_versions(text: str, source_url: str) -> tuple[int, ...]: """Parse the restricted, preference-ordered CSV specified for ``/versions``.""" if '"' in text: raise OptimadeVersionNegotiationError(source_url, "/versions restricted CSV must not contain quotes") if "\r" in text.replace("\r\n", ""): raise OptimadeVersionNegotiationError(source_url, "/versions restricted CSV has an invalid line ending") normalized = text.replace("\r\n", "\n") if normalized.endswith("\n"): body = normalized[:-1] else: body = normalized rows = body.split("\n") if not rows or rows[0].split(",", 1)[0] != "version": raise OptimadeVersionNegotiationError( source_url, "/versions restricted CSV header must begin with 'version'" ) if len(rows) == 1: raise OptimadeVersionNegotiationError(source_url, "/versions restricted CSV has no advertised versions") majors: list[int] = [] seen: set[int] = set() for index, row in enumerate(rows[1:], start=2): if not row: raise OptimadeVersionNegotiationError( source_url, f"/versions restricted CSV has a blank or malformed row at line {index}" ) version = row.split(",", 1)[0] if _VERSION_MAJOR.fullmatch(version) is None: raise OptimadeVersionNegotiationError( source_url, f"/versions restricted CSV has an invalid major version at line {index}" ) major = int(version) if major in seen: raise OptimadeVersionNegotiationError( source_url, f"/versions restricted CSV advertises duplicate major version {major}" ) seen.add(major) majors.append(major) return tuple(majors) def _close_owned_client_after_failed_construction(self) -> None: close = getattr(self._client, "close", None) if callable(close): try: close() except Exception: # Preserve the original discovery exception. A best-effort # cleanup failure must not replace the useful schema error. self._closed = True return self._closed = True def _require_open(self) -> None: if self._closed: raise OptimadeClientError("OPTIMADE store is closed") def _get(self, url: str) -> str: self._require_open() try: response = cast(_HTTPClient, self._client).get(url) except Exception as exc: raise OptimadeTransportError(url, str(exc)) from exc status_code = getattr(response, "status_code", None) text = getattr(response, "text", None) if not isinstance(status_code, int) or not isinstance(text, str): raise OptimadeTransportError(url, "HTTP client returned an invalid response object") if not 200 <= status_code < 300: detail = _error_detail(text) if detail is not None: raise OptimadeErrorDocumentError(url, status_code, detail) raise OptimadeHTTPError(url, status_code) return text @property
[docs] def entry_types(self) -> tuple[RemoteEntryType, ...]: """Discovered entry endpoints in the service-advertised order.""" return self._entry_types
@property
[docs] def entry_types_by_name(self) -> Mapping[str, RemoteEntryType]: """An immutable transport-name lookup for :attr:`entry_types`.""" return self._entry_types_by_name
[docs] def entry_type(self, name: str) -> RemoteEntryType: """Return one discovered endpoint by its transport name.""" try: return self._entry_types_by_name[name] except KeyError as exc: raise KeyError(f"No discovered OPTIMADE entry endpoint named {name!r}") from exc
def _discover(self) -> tuple[str | None, tuple[RemoteEntryType, ...], Mapping[str, RemoteEntryType]]: info_url = self._transport_base_url + "/info" info_document = OptimadeDocument.create(self._get(info_url), info_url) info_root = _parse_json(info_document, label="/info response") data = _mapping(info_root.get("data"), source_url=info_document.source_url, label="/info data") if data.get("type") != "info": raise OptimadeDiscoveryError(info_document.source_url, "/info data.type must be 'info'") attributes = _mapping( data.get("attributes"), source_url=info_document.source_url, label="/info data.attributes" ) api_version, entry_info_resource_identity = self._entry_info_format( attributes.get("api_version"), info_document.source_url ) advertised = attributes.get("available_endpoints") if not isinstance(advertised, list): raise OptimadeDiscoveryError(info_document.source_url, "/info available_endpoints must be a JSON array") endpoint_names: list[str] = [] seen_names: set[str] = set() for endpoint in advertised: if not isinstance(endpoint, str): raise OptimadeDiscoveryError(info_document.source_url, "/info available_endpoints must contain strings") if endpoint in ("", "/", "info", "links") or endpoint.startswith("info/"): continue if endpoint not in seen_names: seen_names.add(endpoint) endpoint_names.append(endpoint) descriptors: list[RemoteEntryType] = [] by_name: dict[str, RemoteEntryType] = {} for name in endpoint_names: endpoint_url = self._transport_base_url + "/info/" + quote(name, safe="") document = OptimadeDocument.create(self._get(endpoint_url), endpoint_url) descriptor = self._entry_descriptor( name, document, require_resource_identity=entry_info_resource_identity, ) descriptors.append(descriptor) by_name[name] = descriptor return api_version, tuple(descriptors), MappingProxyType(by_name) @staticmethod def _entry_info_format(api_version: object, source_url: str) -> tuple[str | None, bool]: """Select the entry-info grammar from the version declared by ``/info``. OPTIMADE 1.0 and 1.1 describe ``data`` as the entry-info object itself. Version 1.2 changed it into a resource object requiring ``type`` and ``id``. Missing version metadata retains the pre-existing strict resource-identity check rather than guessing an older grammar. """ if api_version is None: return None, True if not isinstance(api_version, str): raise OptimadeDiscoveryError(source_url, "/info api_version must be a semantic-version string") match = _API_VERSION.fullmatch(api_version) if match is None: raise OptimadeDiscoveryError(source_url, "/info api_version must be a semantic-version string") major = int(match.group(1)) minor = int(match.group(2)) if major not in _SUPPORTED_API_MAJORS: raise OptimadeDiscoveryError( source_url, f"/info declares unsupported API major version {major}; supported major is 1", ) return api_version, minor >= 2 def _entry_descriptor( self, name: str, document: OptimadeDocument, *, require_resource_identity: bool, ) -> RemoteEntryType: root = _parse_json(document, label=f"/info/{name} response") data = _mapping(root.get("data"), source_url=document.source_url, label=f"/info/{name} data") if require_resource_identity and data.get("type") != "info": raise OptimadeDiscoveryError(document.source_url, f"/info/{name} data.type must be 'info'") properties = _mapping( data.get("properties"), source_url=document.source_url, label=f"/info/{name} data.properties" ) property_iris: dict[str, str] = {} property_names: dict[str, str] = {} property_types: dict[str, tuple[str, str | None]] = {} default_response: list[str] = [] sortable: list[str] = [] for property_name, definition in properties.items(): if not isinstance(property_name, str) or not property_name: raise OptimadeDiscoveryError(document.source_url, "property names must be nonempty strings") definition_mapping = _mapping( definition, source_url=document.source_url, label=f"property definition {property_name!r}" ) definition_id = definition_mapping.get("$id") if _is_definition_iri(definition_id): previous = property_names.get(definition_id) if previous is not None: raise OptimadeDiscoveryError( document.source_url, f"properties {previous!r} and {property_name!r} advertise the same definition IRI {definition_id!r}", ) property_iris[property_name] = definition_id property_names[definition_id] = property_name optimade_type = definition_mapping.get("x-optimade-type") if not isinstance(optimade_type, str): legacy_type = definition_mapping.get("type") optimade_type = legacy_type if isinstance(legacy_type, str) else "unknown" items = definition_mapping.get("items") item_type = ( cast(str, items.get("x-optimade-type")) if isinstance(items, Mapping) and isinstance(items.get("x-optimade-type"), str) else None ) property_types[property_name] = (optimade_type, item_type) implementation = definition_mapping.get("x-optimade-implementation") if implementation is not None: implementation_mapping = _mapping( implementation, source_url=document.source_url, label=f"property {property_name!r} x-optimade-implementation", ) if "sortable" in implementation_mapping: sortable_value = implementation_mapping["sortable"] if not isinstance(sortable_value, bool): raise OptimadeDiscoveryError( document.source_url, f"property {property_name!r} sortable metadata must be a bool" ) if sortable_value: sortable.append(property_name) if "response-default" in implementation_mapping: default_value = implementation_mapping["response-default"] if not isinstance(default_value, bool): raise OptimadeDiscoveryError( document.source_url, f"property {property_name!r} response-default metadata must be a bool", ) if default_value: default_response.append(property_name) describedby: str | None = None links = root.get("links") if links is not None: links_mapping = _mapping(links, source_url=document.source_url, label=f"/info/{name} links") if "describedby" in links_mapping: describedby = _nonempty_string( links_mapping["describedby"], source_url=document.source_url, label=f"/info/{name} links.describedby", ) binding = self._recognise_binding(describedby, frozenset(property_names)) try: backend = OptimadeResource if binding is None else binding.resolve_backend() except Exception as exc: binding_id = describedby if describedby is not None else binding.definition_id if binding else None raise OptimadeDiscoveryError( document.source_url, f"could not resolve OPTIMADE binding backend for {binding_id!r}: {exc}" ) from exc return RemoteEntryType( name=name, definition_id=describedby, schema=OptimadeSchemaSnapshot(name, document), property_iris=property_iris, property_names=property_names, property_types=property_types, advertised_properties=tuple(properties), default_response_properties=tuple(default_response), sortable_properties=tuple(sortable), binding=binding, backend=backend, ) @staticmethod def _recognise_binding( describedby: str | None, remote_property_iris: frozenset[str] ) -> OptimadeEntryBinding | None: if describedby is not None: return optimade_entry_binding(describedby) binding_ids = known_optimade_entry_bindings() if not binding_ids: return None bindings: dict[str, OptimadeEntryBinding] = {} property_owners: dict[str, set[str]] = {} for definition_id in binding_ids: binding = optimade_entry_binding(definition_id) if binding is None: raise OptimadeDiscoveryError( "(local registry)", f"binding {definition_id!r} disappeared during discovery" ) try: definition = load_entry_type_schema(definition_id) except Exception as exc: raise OptimadeDiscoveryError( "(local registry)", f"could not load entry-type definition for binding {definition_id!r}: {exc}" ) from exc bindings[definition_id] = binding for property_definition in definition.properties.values(): property_owners.setdefault(property_definition.definition_id, set()).add(definition_id) candidates = set(binding_ids) universal = set(binding_ids) for property_iri in remote_property_iris: owners = property_owners.get(property_iri, set()) # A locally unknown extension IRI carries no evidence about the # entry type. Known non-universal IRIs do: mutually exclusive # evidence still empties the candidate set and stays generic. if owners and owners != universal: candidates.intersection_update(owners) if len(candidates) == 1: return bindings[candidates.pop()] return None
[docs] def refresh(self) -> None: """Atomically replace discovery state after a fully successful rediscovery.""" with self._lock: self._require_open() api_version, entry_types, by_name = self._discover() self.api_version = api_version self._entry_types = entry_types self._entry_types_by_name = by_name
[docs] def close(self) -> None: """Close an internally owned HTTP client; borrowed clients stay open.""" with self._lock: if self._closed: return self._closed = True if self._owned_client: close = getattr(self._client, "close", None) if callable(close): close()
def __enter__(self) -> Self: self._require_open() return self def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None: self.close()
[docs] def searcher(self, *, response_fields: object = ...) -> "RemoteSearcher": """Create one synchronous, read-only remote search plan. Passing ``response_fields`` overrides the store-level selection. An omitted value inherits it, while explicit ``None`` requests the service default. """ from .remote_query import RemoteSearcher selected = self.response_fields if response_fields is ... else response_fields return RemoteSearcher(self, response_fields=selected)