client

This module implements OPTIMADE client functionality for:

• making web requests to filter and harvest resources from OPTIMADE APIs,
• query multiple providers simultaneously.

ENDPOINTS = ('structures', 'references', 'calculations', 'info', 'extensions') module-attribute

__all__ = ('OptimadeClient') module-attribute

__api_version__ = '1.1.0' module-attribute

__version__ = '1.0.1' module-attribute

BadRequest

Bases: OptimadeHTTPException

400 Bad Request

Source code in optimade/exceptions.py

class BadRequest(OptimadeHTTPException):
    """400 Bad Request"""

    status_code: int = 400
    title: str = "Bad Request"

LarkParser

This class wraps a versioned OPTIMADE grammar and allows it to be parsed into Lark tree objects.

Source code in optimade/filterparser/lark_parser.py

class LarkParser:
    """This class wraps a versioned OPTIMADE grammar and allows
    it to be parsed into Lark tree objects.

    """

    def __init__(
        self, version: Optional[tuple[int, int, int]] = None, variant: str = "default"
    ):
        """For a given version and variant, try to load the corresponding grammar.

        Parameters:
            version: The grammar version number to use (e.g., `(1, 0, 1)` for v1.0.1).
            variant: The grammar variant to employ.

        Raises:
            ParserError: If the requested version/variant of the
                grammar does not exist.

        """

        if not version:
            version = max(
                _ for _ in AVAILABLE_PARSERS if AVAILABLE_PARSERS[_].get("default")
            )

        if version not in AVAILABLE_PARSERS:
            raise ParserError(f"Unknown parser grammar version: {version}")

        if variant not in AVAILABLE_PARSERS[version]:
            raise ParserError(f"Unknown variant of the parser: {variant}")

        self.version = version
        self.variant = variant

        with open(AVAILABLE_PARSERS[version][variant]) as f:
            self.lark = Lark(f, maybe_placeholders=False)

        self.tree: Optional[Tree] = None
        self.filter: Optional[str] = None

    def parse(self, filter_: str) -> Tree:
        """Parse a filter string into a `lark.Tree`.

        Parameters:
            filter_: The filter string to parse.

        Raises:
            BadRequest: If the filter cannot be parsed.

        Returns:
            The parsed filter.

        """
        try:
            self.tree = self.lark.parse(filter_)
            self.filter = filter_
            return self.tree
        except Exception as exc:
            raise BadRequest(
                detail=f"Unable to parse filter {filter_}. Lark traceback: \n{exc}"
            ) from exc

    def __repr__(self):
        if isinstance(self.tree, Tree):
            return self.tree.pretty()
        return repr(self.lark)

__init__(version=None, variant='default')

For a given version and variant, try to load the corresponding grammar.

Parameters:

Name	Type	Description	Default
version	Optional[tuple[int, int, int]]	The grammar version number to use (e.g., (1, 0, 1) for v1.0.1).	None
variant	str	The grammar variant to employ.	'default'

Raises:

Type	Description
ParserError	If the requested version/variant of the grammar does not exist.

Source code in optimade/filterparser/lark_parser.py

def __init__(
    self, version: Optional[tuple[int, int, int]] = None, variant: str = "default"
):
    """For a given version and variant, try to load the corresponding grammar.

    Parameters:
        version: The grammar version number to use (e.g., `(1, 0, 1)` for v1.0.1).
        variant: The grammar variant to employ.

    Raises:
        ParserError: If the requested version/variant of the
            grammar does not exist.

    """

    if not version:
        version = max(
            _ for _ in AVAILABLE_PARSERS if AVAILABLE_PARSERS[_].get("default")
        )

    if version not in AVAILABLE_PARSERS:
        raise ParserError(f"Unknown parser grammar version: {version}")

    if variant not in AVAILABLE_PARSERS[version]:
        raise ParserError(f"Unknown variant of the parser: {variant}")

    self.version = version
    self.variant = variant

    with open(AVAILABLE_PARSERS[version][variant]) as f:
        self.lark = Lark(f, maybe_placeholders=False)

    self.tree: Optional[Tree] = None
    self.filter: Optional[str] = None

parse(filter_)

Parse a filter string into a lark.Tree.

Parameters:

Name	Type	Description	Default
filter_	str	The filter string to parse.	required

Raises:

Type	Description
BadRequest	If the filter cannot be parsed.

Returns:

Type	Description
Tree	The parsed filter.

Source code in optimade/filterparser/lark_parser.py

def parse(self, filter_: str) -> Tree:
    """Parse a filter string into a `lark.Tree`.

    Parameters:
        filter_: The filter string to parse.

    Raises:
        BadRequest: If the filter cannot be parsed.

    Returns:
        The parsed filter.

    """
    try:
        self.tree = self.lark.parse(filter_)
        self.filter = filter_
        return self.tree
    except Exception as exc:
        raise BadRequest(
            detail=f"Unable to parse filter {filter_}. Lark traceback: \n{exc}"
        ) from exc

OptimadeClient

This class implemements a client for executing the same queries across multiple OPTIMADE APIs simultaneously, paging and caching the results.

By default, all registered OPTIMADE providers will be queried simulateneously and asynchronously, with the results collected into the all_results attribute, keyed by endpoint, filter and provider.

Source code in optimade/client/client.py

class OptimadeClient:
    """This class implemements a client for executing the same queries
    across multiple OPTIMADE APIs simultaneously, paging and caching the
    results.

    By default, all registered OPTIMADE providers will be queried
    simulateneously and asynchronously, with the results collected
    into the `all_results` attribute, keyed by endpoint, filter
    and provider.

    """

    base_urls: Union[str, Iterable[str]]
    """A list (or any iterable) of OPTIMADE base URLs to query."""

    all_results: dict[str, dict[str, dict[str, QueryResults]]] = defaultdict(dict)
    """A nested dictionary keyed by endpoint and OPTIMADE filter string that contains
    the results from each base URL for that particular filter.
    """

    count_results: dict[str, dict[str, dict[str, int]]] = defaultdict(dict)
    """A nested dictionary keyed by endpoint and OPTIMADE filter string that contains
    the number of results from each base URL for that particular filter.
    """

    max_results_per_provider: Optional[int] = None
    """Maximum number of results to downlod per provider. If None, will
    download all.
    """

    property_lists: dict[str, dict[str, list[str]]] = defaultdict(dict)
    """A dictionary containing list of properties served by each database,
    broken down by entry type, then database.
    """

    headers: dict = {"User-Agent": f"optimade-python-tools/{__version__}"}
    """Additional HTTP headers."""

    http_timeout: httpx.Timeout = httpx.Timeout(10.0, read=1000.0)
    """The timeout to use for each HTTP request."""

    max_attempts: int
    """The maximum number of times to repeat a failed query before giving up."""

    use_async: bool
    """Whether or not to make all requests asynchronously using asyncio."""

    callbacks: Optional[list[Callable[[str, dict], Union[None, dict]]]] = None
    """A list of callbacks to execute after each successful request, used
    to e.g., write to a file, add results to a database or perform additional
    filtering.

    The callbacks will receive the request URL and the results extracted
    from the JSON response, with keys 'data', 'meta', 'links', 'errors'
    and 'included'.

    Each callback can return a dictionary that can modify the `next_url` with the
    key `next` and the progress bar with the key `advance_results`.
    In the case of multiple provided callbacks, only the value returned by the final
    callback in the stack will be used.

    """

    silent: bool
    """Whether to disable progress bar printing."""

    _excluded_providers: Optional[set[str]] = None
    """A set of providers IDs excluded from future queries."""

    _included_providers: Optional[set[str]] = None
    """A set of providers IDs included from future queries."""

    _excluded_databases: Optional[set[str]] = None
    """A set of child database URLs excluded from future queries."""

    __current_endpoint: Optional[str] = None
    """Used internally when querying via `client.structures.get()` to set the
    chosen endpoint. Should be reset to `None` outside of all `get()` calls."""

    _http_client: Optional[
        Union[type[httpx.AsyncClient], type[requests.Session]]
    ] = None
    """Override the HTTP client class, primarily used for testing."""

    __strict_async: bool = False
    """Whether or not to fallover if `use_async` is true yet asynchronous mode
    is impossible due to, e.g., a running event loop.
    """

    def __init__(
        self,
        base_urls: Optional[Union[str, Iterable[str]]] = None,
        max_results_per_provider: int = 1000,
        headers: Optional[dict] = None,
        http_timeout: Optional[Union[httpx.Timeout, float]] = None,
        max_attempts: int = 5,
        use_async: bool = True,
        silent: bool = False,
        exclude_providers: Optional[list[str]] = None,
        include_providers: Optional[list[str]] = None,
        exclude_databases: Optional[list[str]] = None,
        http_client: Optional[
            Union[type[httpx.AsyncClient], type[requests.Session]]
        ] = None,
        verbose: int = 0,
        callbacks: Optional[list[Callable[[str, dict], Union[None, dict]]]] = None,
    ):
        """Create the OPTIMADE client object.

        Parameters:
            base_urls: A list of OPTIMADE base URLs to query.
            max_results_per_provider: The maximum number of results to download
                from each provider (-1 or 0 indicate unlimited).
            headers: Any additional HTTP headers to use for the queries.
            http_timeout: The timeout to use per request. Defaults to 10
                seconds with 1000 seconds for reads specifically. Overriding this value
                will replace all timeouts (connect, read, write and pool) with this value.
            max_attempts: The maximum number of times to repeat a failing query.
            use_async: Whether or not to make all requests asynchronously.
            exclude_providers: A set or collection of provider IDs to exclude from queries.
            include_providers: A set or collection of provider IDs to include in queries.
            exclude_databases: A set or collection of child database URLs to exclude from queries.
            http_client: An override for the underlying HTTP client, primarily used for testing.
            callbacks: A list of functions to call after each successful response, see the
                attribute [`OptimadeClient.callbacks`][optimade.client.OptimadeClient.callbacks]
                docstring for more details.

        """

        self.max_results_per_provider = max_results_per_provider
        if self.max_results_per_provider in (-1, 0):
            self.max_results_per_provider = None

        self._excluded_providers = set(exclude_providers) if exclude_providers else None
        self._included_providers = set(include_providers) if include_providers else None
        self._excluded_databases = set(exclude_databases) if exclude_databases else None

        self.max_attempts = max_attempts
        self.silent = silent

        if headers:
            self.headers.update(headers)

        if not base_urls:
            progress = None
            if not self.silent:
                progress = OptimadeClientProgress()
            self.base_urls = list(
                get_all_databases(
                    exclude_providers=self._excluded_providers,
                    include_providers=self._included_providers,
                    exclude_databases=self._excluded_databases,
                    progress=progress,
                )
            )
        else:
            if exclude_providers or include_providers or exclude_databases:
                raise RuntimeError(
                    "Cannot provide both a list of base URLs and included/excluded databases."
                )

            self.base_urls = base_urls

            if isinstance(self.base_urls, str):
                self.base_urls = [self.base_urls]
            self.base_urls = list(self.base_urls)

        if not self.base_urls:
            raise SystemExit(
                "Unable to access any OPTIMADE base URLs. If you believe this is an error, try manually specifying some base URLs."
            )

        if http_timeout:
            if isinstance(http_timeout, httpx.Timeout):
                self.http_timeout = http_timeout
            else:
                self.http_timeout = httpx.Timeout(http_timeout)

        self.use_async = use_async

        if http_client:
            self._http_client = http_client
            if issubclass(self._http_client, httpx.AsyncClient):
                if not self.use_async and self.__strict_async:
                    raise RuntimeError(
                        "Cannot use synchronous mode with an asynchronous HTTP client, please set `use_async=True` or pass an asynchronous HTTP client."
                    )
                self.use_async = True
            elif issubclass(self._http_client, requests.Session):
                if self.use_async and self.__strict_async:
                    raise RuntimeError(
                        "Cannot use async mode with a synchronous HTTP client, please set `use_async=False` or pass an synchronous HTTP client."
                    )
                self.use_async = False
        else:
            if use_async:
                self._http_client = httpx.AsyncClient
            else:
                self._http_client = requests.Session

        self.callbacks = callbacks

    def __getattribute__(self, name):
        """Allows entry endpoints to be queried via attribute access, using the
        allowed list for this module.

        Should also pass through any `extensions/<example>` endpoints.

        Any non-entry-endpoint name requested will be passed to the
        original `__getattribute__`.

        !!! example
            ```python
            from optimade.client import OptimadeClient
            cli = OptimadeClient()
            structures = cli.structures.get()
            references = cli.references.get()
            info_structures = cli.info.structures.get()
            ```

        """
        if name in ENDPOINTS:
            if self.__current_endpoint == "info":
                self.__current_endpoint = f"info/{name}"
            elif self.__current_endpoint == "extensions":
                self.__current_endpoint = f"extensions/{name}"
            else:
                self.__current_endpoint = name
            return self

        return super().__getattribute__(name)

    def get(
        self,
        filter: Optional[str] = None,
        endpoint: Optional[str] = None,
        response_fields: Optional[list[str]] = None,
        sort: Optional[str] = None,
    ) -> dict[str, dict[str, dict[str, dict]]]:
        """Gets the results from the endpoint and filter across the
        defined OPTIMADE APIs.

        Parameters:
            filter: The OPTIMADE filter string for the query.
            endpoint: The endpoint to query.
            response_fields: A list of response fields to request
                from the server.
            sort: The field by which to sort the results.

        Raises:
            RuntimeError: If the query could not be executed.

        Returns:
            A nested mapping from endpoint, filter and base URL to the query results.

        """

        if endpoint is None:
            if self.__current_endpoint is not None:
                endpoint = self.__current_endpoint
                self.__current_endpoint = None
            else:
                endpoint = "structures"

        if filter is None:
            filter = ""

        self._progress = OptimadeClientProgress()
        if self.silent:
            self._progress.disable = True

        self._check_filter(filter, endpoint)

        with self._progress:
            if not self.silent:
                self._progress.print(
                    Panel(
                        f"Performing query [bold yellow]{endpoint}[/bold yellow]/?filter=[bold magenta][i]{filter}[/i][/bold magenta]",
                        expand=False,
                    )
                )
            results = self._execute_queries(
                filter,
                endpoint,
                response_fields=response_fields,
                page_limit=None,
                paginate=True,
                sort=sort,
            )
            self.all_results[endpoint][filter] = results
            return {endpoint: {filter: {k: results[k].asdict() for k in results}}}

    def count(
        self, filter: Optional[str] = None, endpoint: Optional[str] = None
    ) -> dict[str, dict[str, dict[str, Optional[int]]]]:
        """Counts the number of results for the filter, requiring
        only 1 request per provider by making use of the `meta->data_returned`
        key.

        Raises:
            RuntimeError: If the query could not be executed.

        Returns:
            A nested mapping from endpoint, filter and base URL to the number of query results.

        """

        if endpoint is None:
            if self.__current_endpoint is not None:
                endpoint = self.__current_endpoint
                self.__current_endpoint = None
            else:
                endpoint = "structures"

        if filter is None:
            filter = ""

        self._progress = OptimadeClientProgress()
        if self.silent:
            self._progress.disable = True

        self._check_filter(filter, endpoint)

        with self._progress:
            if not self.silent:
                self._progress.print(
                    Panel(
                        f"Counting results for [bold yellow]{endpoint}[/bold yellow]/?filter=[bold magenta][i]{filter}[/i][/bold magenta]",
                        expand=False,
                    )
                )
            results = self._execute_queries(
                filter,
                endpoint,
                page_limit=1,
                paginate=False,
                response_fields=[],
                sort=None,
            )
            count_results = {}

            for base_url in results:
                count_results[base_url] = results[base_url].meta.get(
                    "data_returned", None
                )

                if count_results[base_url] is None:
                    self._progress.print(
                        f"Warning: {base_url} did not return a value for `meta->data_returned`, unable to count results. Full response: {results[base_url]}"
                    )

            self.count_results[endpoint][filter] = count_results
            return {endpoint: {filter: count_results}}

    def list_properties(
        self,
        entry_type: str,
    ) -> dict[str, list[str]]:
        """Returns the list of properties reported at `/info/<entry_type>`
        for the given entry type, for each database.

        """
        self._progress = OptimadeClientProgress()
        if self.silent:
            self._progress.disable = True

        with self._progress:
            if not self.silent:
                self._progress.print(
                    Panel(
                        f"Listing properties for [bold yellow]{entry_type}[/bold yellow]",
                        expand=False,
                    )
                )
            results = self._execute_queries(
                "",
                f"info/{entry_type}",
                paginate=False,
                page_limit=1,
                response_fields=[],
                sort=None,
            )
        self.property_lists = {entry_type: {}}
        for database in results:
            self.property_lists[entry_type][database] = list(
                results[database].data.get("properties", {}).keys()  # type: ignore
            )
        return self.property_lists[entry_type]

    def search_property(self, query: str, entry_type: str) -> dict[str, list[str]]:
        """Searches for the query substring within the listed properties
        served by each database.

        Parameters:
            query: The substring to search for.
            entry_type: The entry type to query.

        Returns:
            A nested dictionary of matching property lists, arranged by
            entry type and database.

        """
        if not self.property_lists:
            self.list_properties(entry_type=entry_type)

        matching_properties: dict[str, dict[str, list[str]]] = {
            entry_type: defaultdict(list)
        }
        if entry_type in self.property_lists:
            for database in self.property_lists[entry_type]:
                for property in self.property_lists[entry_type][database]:
                    if query in property:
                        matching_properties[entry_type][database].append(property)
        return matching_properties[entry_type]

    def _execute_queries(
        self,
        filter: str,
        endpoint: str,
        page_limit: Optional[int],
        paginate: bool,
        response_fields: Optional[list[str]],
        sort: Optional[str],
    ) -> dict[str, QueryResults]:
        """Executes the queries over the base URLs either asynchronously or
        serially, depending on the `self.use_async` setting.

        Parameters:
            filter: The OPTIMADE filter string.
            endpoint: The OPTIMADE endpoint to query.
            page_limit: A page limit to enforce for each query (used in
                conjunction with `paginate`).
            paginate: Whether to pull all pages of results (up to the
                value of `max_results_per_provider`) or whether to return
                after one page.
            response_fields: A list of response fields to request
                from the server.
            sort: The field by which to sort the results.

        Returns:
            A mapping from base URL to `QueryResults` for each queried API.

        """

        if self.use_async:
            # Check for a pre-existing event loop (e.g. within a Jupyter notebook)
            # and use it if present
            try:
                event_loop = asyncio.get_running_loop()
                if event_loop:
                    if self.__strict_async:
                        raise RuntimeError(
                            "Detected a running event loop, cannot run in async mode."
                        )
                    self._progress.print(
                        "Detected a running event loop (e.g., Jupyter, pytest). Trying to use nest_asyncio."
                    )
                    self.use_async = False
            except RuntimeError:
                event_loop = None

        if self.use_async and not event_loop:
            return asyncio.run(
                self._get_all_async(
                    endpoint,
                    filter,
                    page_limit=page_limit,
                    paginate=paginate,
                    response_fields=response_fields,
                    sort=sort,
                )
            )

        return self._get_all(
            endpoint,
            filter,
            page_limit=page_limit,
            paginate=paginate,
            response_fields=response_fields,
            sort=sort,
        )

    def get_one(
        self,
        endpoint: str,
        filter: str,
        base_url: str,
        response_fields: Optional[list[str]] = None,
        sort: Optional[str] = None,
        page_limit: Optional[int] = None,
        paginate: bool = True,
    ) -> dict[str, QueryResults]:
        """Executes the query synchronously on one API.

        Parameters:
            endpoint: The OPTIMADE endpoint to query.
            filter: The OPTIMADE filter string.
            response_fields: A list of response fields to request
                from the server.
            sort: The field by which to sort the results.
            page_limit: A page limit to enforce for each query (used in
                conjunction with `paginate`).
            paginate: Whether to pull all pages of results (up to the
                value of `max_results_per_provider`) or whether to return
                after one page.

        Returns:
            A dictionary mapping from base URL to the results of the query.

        """
        try:
            return self._get_one(
                endpoint,
                filter,
                base_url,
                page_limit=page_limit,
                paginate=paginate,
                response_fields=response_fields,
                sort=sort,
            )
        except Exception as exc:
            error_query_results = QueryResults()
            error_query_results.errors = [
                f"{exc.__class__.__name__}: {str(exc.args[0])}"
            ]
            self._progress.print(
                f"[red]Error[/red]: Provider {str(base_url)!r} returned: [red i]{exc}[/red i]"
            )
            return {base_url: error_query_results}

    async def _get_all_async(
        self,
        endpoint: str,
        filter: str,
        response_fields: Optional[list[str]] = None,
        sort: Optional[str] = None,
        page_limit: Optional[int] = None,
        paginate: bool = True,
    ) -> dict[str, QueryResults]:
        """Executes the query asynchronously across all defined APIs.

        Parameters:
            endpoint: The OPTIMADE endpoint to query.
            filter: The OPTIMADE filter string.
            response_fields: A list of response fields to request
                from the server.
            sort: The field by which to sort the results.
            page_limit: A page limit to enforce for each query (used in
                conjunction with `paginate`).
            paginate: Whether to pull all pages of results (up to the
                value of `max_results_per_provider`) or whether to return
                after one page.

        Returns:
            A dictionary mapping from base URL to the results of the query.

        """
        results = await asyncio.gather(
            *[
                self.get_one_async(
                    endpoint,
                    filter,
                    base_url,
                    page_limit=page_limit,
                    paginate=paginate,
                    response_fields=response_fields,
                    sort=sort,
                )
                for base_url in self.base_urls
            ]
        )
        return functools.reduce(lambda r1, r2: {**r1, **r2}, results)

    def _get_all(
        self,
        endpoint: str,
        filter: str,
        page_limit: Optional[int] = None,
        response_fields: Optional[list[str]] = None,
        sort: Optional[str] = None,
        paginate: bool = True,
    ) -> dict[str, QueryResults]:
        """Executes the query synchronously across all defined APIs.

        Parameters:
            endpoint: The OPTIMADE endpoint to query.
            filter: The OPTIMADE filter string.
            response_fields: A list of response fields to request
                from the server.
            sort: The field by which to sort the results.
            page_limit: A page limit to enforce for each query (used in
                conjunction with `paginate`).
            paginate: Whether to pull all pages of results (up to the
                value of `max_results_per_provider`) or whether to return
                after one page.

        Returns:
            A dictionary mapping from base URL to the results of the query.

        """
        results = [
            self.get_one(
                endpoint,
                filter,
                base_url,
                page_limit=page_limit,
                paginate=paginate,
                response_fields=response_fields,
                sort=sort,
            )
            for base_url in self.base_urls
        ]
        if results:
            return functools.reduce(lambda r1, r2: {**r1, **r2}, results)

        return {}

    async def get_one_async(
        self,
        endpoint: str,
        filter: str,
        base_url: str,
        response_fields: Optional[list[str]] = None,
        sort: Optional[str] = None,
        page_limit: Optional[int] = None,
        paginate: bool = True,
    ) -> dict[str, QueryResults]:
        """Executes the query asynchronously on one API.

        !!! note
            This method currently makes non-blocking requests
            to a single API, but these requests are executed
            serially on that API, i.e., results are pulled one
            page at a time, but requests will not block other
            async requests to other APIs.

        Parameters:
            endpoint: The OPTIMADE endpoint to query.
            filter: The OPTIMADE filter string.
            response_fields: A list of response fields to request
                from the server.
            sort: The field by which to sort the results.
            page_limit: A page limit to enforce for each query (used in
                conjunction with `paginate`).
            paginate: Whether to pull all pages of results (up to the
                value of `max_results_per_provider`) or whether to return
                after one page.

        Returns:
            A dictionary mapping from base URL to the results of the query.

        """
        try:
            return await self._get_one_async(
                endpoint,
                filter,
                base_url,
                page_limit=page_limit,
                paginate=paginate,
                response_fields=response_fields,
                sort=sort,
            )
        except Exception as exc:
            error_query_results = QueryResults()
            error_query_results.errors = [
                f"{exc.__class__.__name__}: {str(exc.args[0])}"
            ]
            self._progress.print(
                f"[red]Error[/red]: Provider {str(base_url)!r} returned: [red i]{error_query_results.errors}[/red i]"
            )
            return {base_url: error_query_results}

    async def _get_one_async(
        self,
        endpoint: str,
        filter: str,
        base_url: str,
        response_fields: Optional[list[str]] = None,
        sort: Optional[str] = None,
        page_limit: Optional[int] = None,
        paginate: bool = True,
    ) -> dict[str, QueryResults]:
        """See [`OptimadeClient.get_one_async`][optimade.client.OptimadeClient.get_one_async]."""
        next_url, _task = self._setup(
            endpoint=endpoint,
            base_url=base_url,
            filter=filter,
            page_limit=page_limit,
            response_fields=response_fields,
            sort=sort,
        )
        results = QueryResults()
        try:
            async with self._http_client(headers=self.headers) as client:  # type: ignore[union-attr,call-arg,misc]
                while next_url:
                    attempts = 0
                    try:
                        r = await client.get(
                            next_url, follow_redirects=True, timeout=self.http_timeout
                        )
                        page_results, next_url = self._handle_response(r, _task)
                    except RecoverableHTTPError:
                        attempts += 1
                        if attempts > self.max_attempts:
                            raise RuntimeError(
                                f"Exceeded maximum number of retries for {next_url}"
                            )
                        await asyncio.sleep(1)
                        continue

                    results.update(page_results)

                    if not paginate:
                        break

                    if (
                        self.max_results_per_provider
                        and len(results.data) >= self.max_results_per_provider
                    ):
                        if not self.silent:
                            self._progress.print(
                                f"Reached {len(results.data)} results for {base_url}, exceeding `max_results_per_provider` parameter ({self.max_results_per_provider}). Stopping download."
                            )
                        break

            return {str(base_url): results}

        finally:
            self._teardown(_task, len(results.data))

    def _get_one(
        self,
        endpoint: str,
        filter: str,
        base_url: str,
        sort: Optional[str] = None,
        page_limit: Optional[int] = None,
        response_fields: Optional[list[str]] = None,
        paginate: bool = True,
    ) -> dict[str, QueryResults]:
        """See [`OptimadeClient.get_one`][optimade.client.OptimadeClient.get_one]."""
        next_url, _task = self._setup(
            endpoint=endpoint,
            base_url=base_url,
            filter=filter,
            page_limit=page_limit,
            response_fields=response_fields,
            sort=sort,
        )
        results = QueryResults()
        try:
            with self._http_client() as client:  # type: ignore[misc]
                client.headers.update(self.headers)

                if isinstance(client, requests.Session):
                    # Convert configured httpx timeout to requests-style tuple
                    timeout = (self.http_timeout.connect, self.http_timeout.read)

                while next_url:
                    attempts = 0
                    try:
                        r = client.get(next_url, timeout=timeout)
                        page_results, next_url = self._handle_response(r, _task)
                    except RecoverableHTTPError:
                        attempts += 1
                        if attempts > self.max_attempts:
                            raise RuntimeError(
                                f"Exceeded maximum number of retries for {next_url}"
                            )
                        time.sleep(1)
                        continue

                    results.update(page_results)

                    if (
                        self.max_results_per_provider
                        and len(results.data) >= self.max_results_per_provider
                    ):
                        if not self.silent:
                            self._progress.print(
                                f"Reached {len(results.data)} results for {base_url}, exceeding `max_results_per_provider` parameter ({self.max_results_per_provider}). Stopping download."
                            )
                        break

                    if not paginate:
                        break

            return {str(base_url): results}

        finally:
            self._teardown(_task, len(results.data))

    def _setup(
        self,
        endpoint: str,
        base_url: str,
        filter: str,
        page_limit: Optional[int],
        response_fields: Optional[list[str]],
        sort: Optional[str],
    ) -> tuple[str, TaskID]:
        """Constructs the first query URL and creates the progress bar task.

        Returns:
            The URL for the first query and the Rich TaskID for progress logging.

        """
        url = self._build_url(
            base_url=base_url,
            endpoint=endpoint,
            filter=filter,
            page_limit=page_limit,
            response_fields=response_fields,
            sort=sort,
        )
        parsed_url = urlparse(url)
        _task = self._progress.add_task(
            description=parsed_url.netloc + parsed_url.path,
            total=None,
        )
        return url, _task

    def _build_url(
        self,
        base_url: str,
        endpoint: Optional[str] = "structures",
        version: Optional[str] = None,
        filter: Optional[str] = None,
        response_fields: Optional[list[str]] = None,
        sort: Optional[str] = None,
        page_limit: Optional[int] = None,
    ) -> str:
        """Builds the URL to query based on the passed parameters.

        Parameters:
            base_url: The server's base URL.
            endpoint: The endpoint to query.
            version: The OPTIMADE version string.
            filter: The filter to apply to the endpoint.
            response_fields: A list of response fields to request from the server.
            sort: The field by which to sort the results.
            page_limit: The page limit for an individual request.

        Returns:
            The overall query URL, including parameters.

        """

        if not version:
            version = f'v{__api_version__.split(".")[0]}'
        while base_url.endswith("/"):
            base_url = base_url[:-1]

        url = f"{base_url}/{version}/{endpoint}"

        # Handle params
        _filter: Optional[str] = None
        _response_fields: Optional[str] = None
        _page_limit: Optional[str] = None
        _sort: Optional[str] = None

        if filter:
            _filter = f"filter={filter}"
        if response_fields is not None:
            # If we have requested no response fields (e.g., in the case of --count) then just ask for IDs
            if len(response_fields) == 0:
                _response_fields = "response_fields=id"
            else:
                _response_fields = f'response_fields={",".join(response_fields)}'
        if page_limit:
            _page_limit = f"page_limit={page_limit}"
        if sort:
            _sort = f"sort={sort}"

        params = "&".join(
            p for p in (_filter, _response_fields, _page_limit, _sort) if p
        )
        if params:
            url += f"?{params}"

        return url

    def _check_filter(self, filter: str, endpoint: str) -> None:
        """Passes the filter through [`LarkParser`][optimade.filterparser.LarkParser]
        from the optimade-python-tools reference server implementation.

        Parameters:
            filter: The filter string.
            endpoint: The endpoint being queried. If this endpoint is not "known" to
                OPTIMADE, the filter will automatically pass.

        Raises:
            RuntimeError: If the filter cannot be parsed.

        """
        try:
            if endpoint in ENDPOINTS:
                parser = LarkParser()
                parser.parse(filter)
        except BadRequest as exc:
            self._progress.print(
                f"[bold red]Filter [blue i]{filter!r}[/blue i] could not be parsed as an OPTIMADE filter.[/bold red]",
                Panel(f"[magenta]{exc}[/magenta]"),
            )
            with silent_raise():
                raise RuntimeError(exc) from None

    def _handle_response(
        self, response: Union[httpx.Response, requests.Response], _task: TaskID
    ) -> tuple[dict[str, Any], str]:
        """Handle the response from the server.

        Parameters:
            response: The response from the server.
            _task: The Rich TaskID for this task's progressbar.

        Returns:
            A dictionary containing the results, and a link to the next page,
            if it exists.

        """

        # Handle error statuses
        if response.status_code == 429:
            raise TooManyRequestsException(response.content)
        if response.status_code != 200:
            try:
                errors = response.json().get("errors")
                error_message = "\n".join(
                    [f"{error['title']}: {error['detail']}" for error in errors]
                )
            except Exception:
                error_message = str(response.content)

            raise RuntimeError(
                f"{response.status_code} - {response.url}: {error_message}"
            )

        try:
            r = response.json()
        except json.JSONDecodeError as exc:
            raise RuntimeError(
                f"Could not decode response as JSON: {response.content!r}"
            ) from exc

        # Accumulate results with correct empty containers if missing
        results = {
            "data": r.get("data", []),
            "meta": r.get("meta", {}),
            "links": r.get("links", {}),
            "included": r.get("included", []),
            "errors": r.get("errors", []),
        }

        callback_response = None
        if self.callbacks:
            callback_response = self._execute_callbacks(results, response)
        callback_response = callback_response or {}

        # Advance the progress bar for this provider
        self._progress.update(
            _task,
            advance=callback_response.get("advance_results", len(results["data"])),
            total=results["meta"].get("data_returned", None),
        )

        next_url = callback_response.get("next") or results["links"].get("next", None)
        if isinstance(next_url, dict):
            next_url = next_url.pop("href")

        return results, next_url

    def _teardown(self, _task: TaskID, num_results: int) -> None:
        """Update the finished status of the progress bar depending on the number of results.

        Parameters:
            _task: The Rich TaskID for this task's progressbar.
            num_results: The number of data entries returned.

        """
        if num_results == 0:
            self._progress.update(_task, total=None, finished=False, complete=True)
        else:
            self._progress.update(
                _task, total=num_results, finished=True, complete=True
            )

    def _execute_callbacks(
        self, results: dict, response: Union[httpx.Response, requests.Response]
    ) -> Union[None, dict]:
        """Execute any callbacks registered with the client.

        Parameters:
            results: The results from the query.
            response: The full response from the server.

        Returns:
            Either `None` or the string value returned from the *final* callback.

        """
        request_url = str(response.request.url)
        if not self.callbacks:
            return None
        for callback in self.callbacks:
            cb_response = callback(request_url, results)
        return cb_response

__current_endpoint: Optional[str] = None class-attribute instance-attribute

Used internally when querying via client.structures.get() to set the chosen endpoint. Should be reset to None outside of all get() calls.

__strict_async: bool = False class-attribute instance-attribute

Whether or not to fallover if use_async is true yet asynchronous mode is impossible due to, e.g., a running event loop.

all_results: dict[str, dict[str, dict[str, QueryResults]]] = defaultdict(dict) class-attribute instance-attribute

A nested dictionary keyed by endpoint and OPTIMADE filter string that contains the results from each base URL for that particular filter.

base_urls: Union[str, Iterable[str]] instance-attribute

A list (or any iterable) of OPTIMADE base URLs to query.

callbacks: Optional[list[Callable[[str, dict], Union[None, dict]]]] = callbacks class-attribute instance-attribute

A list of callbacks to execute after each successful request, used to e.g., write to a file, add results to a database or perform additional filtering.

The callbacks will receive the request URL and the results extracted from the JSON response, with keys 'data', 'meta', 'links', 'errors' and 'included'.

Each callback can return a dictionary that can modify the next_url with the key next and the progress bar with the key advance_results. In the case of multiple provided callbacks, only the value returned by the final callback in the stack will be used.

count_results: dict[str, dict[str, dict[str, int]]] = defaultdict(dict) class-attribute instance-attribute

A nested dictionary keyed by endpoint and OPTIMADE filter string that contains the number of results from each base URL for that particular filter.

headers: dict = {'User-Agent': f'optimade-python-tools/{__version__}'} class-attribute instance-attribute

Additional HTTP headers.

http_timeout: httpx.Timeout = httpx.Timeout(10.0, read=1000.0) class-attribute instance-attribute

The timeout to use for each HTTP request.

max_attempts: int = max_attempts instance-attribute

The maximum number of times to repeat a failed query before giving up.

max_results_per_provider: Optional[int] = max_results_per_provider class-attribute instance-attribute

Maximum number of results to downlod per provider. If None, will download all.

property_lists: dict[str, dict[str, list[str]]] = defaultdict(dict) class-attribute instance-attribute

A dictionary containing list of properties served by each database, broken down by entry type, then database.

silent: bool = silent instance-attribute

Whether to disable progress bar printing.

use_async: bool = use_async instance-attribute

Whether or not to make all requests asynchronously using asyncio.

__getattribute__(name)

Allows entry endpoints to be queried via attribute access, using the allowed list for this module.

Should also pass through any extensions/<example> endpoints.

Any non-entry-endpoint name requested will be passed to the original __getattribute__.

Example

from optimade.client import OptimadeClient
cli = OptimadeClient()
structures = cli.structures.get()
references = cli.references.get()
info_structures = cli.info.structures.get()

Source code in optimade/client/client.py

def __getattribute__(self, name):
    """Allows entry endpoints to be queried via attribute access, using the
    allowed list for this module.

    Should also pass through any `extensions/<example>` endpoints.

    Any non-entry-endpoint name requested will be passed to the
    original `__getattribute__`.

    !!! example
        ```python
        from optimade.client import OptimadeClient
        cli = OptimadeClient()
        structures = cli.structures.get()
        references = cli.references.get()
        info_structures = cli.info.structures.get()
        ```

    """
    if name in ENDPOINTS:
        if self.__current_endpoint == "info":
            self.__current_endpoint = f"info/{name}"
        elif self.__current_endpoint == "extensions":
            self.__current_endpoint = f"extensions/{name}"
        else:
            self.__current_endpoint = name
        return self

    return super().__getattribute__(name)

__init__(base_urls=None, max_results_per_provider=1000, headers=None, http_timeout=None, max_attempts=5, use_async=True, silent=False, exclude_providers=None, include_providers=None, exclude_databases=None, http_client=None, verbose=0, callbacks=None)

Create the OPTIMADE client object.

Parameters:

Name	Type	Description	Default
base_urls	Optional[Union[str, Iterable[str]]]	A list of OPTIMADE base URLs to query.	None
max_results_per_provider	int	The maximum number of results to download from each provider (-1 or 0 indicate unlimited).	1000
headers	Optional[dict]	Any additional HTTP headers to use for the queries.	None
http_timeout	Optional[Union[Timeout, float]]	The timeout to use per request. Defaults to 10 seconds with 1000 seconds for reads specifically. Overriding this value will replace all timeouts (connect, read, write and pool) with this value.	None
max_attempts	int	The maximum number of times to repeat a failing query.	5
use_async	bool	Whether or not to make all requests asynchronously.	True
exclude_providers	Optional[list[str]]	A set or collection of provider IDs to exclude from queries.	None
include_providers	Optional[list[str]]	A set or collection of provider IDs to include in queries.	None
exclude_databases	Optional[list[str]]	A set or collection of child database URLs to exclude from queries.	None
http_client	Optional[Union[type[AsyncClient], type[Session]]]	An override for the underlying HTTP client, primarily used for testing.	None
callbacks	Optional[list[Callable[[str, dict], Union[None, dict]]]]	A list of functions to call after each successful response, see the attribute OptimadeClient.callbacks docstring for more details.	None

Source code in optimade/client/client.py

def __init__(
    self,
    base_urls: Optional[Union[str, Iterable[str]]] = None,
    max_results_per_provider: int = 1000,
    headers: Optional[dict] = None,
    http_timeout: Optional[Union[httpx.Timeout, float]] = None,
    max_attempts: int = 5,
    use_async: bool = True,
    silent: bool = False,
    exclude_providers: Optional[list[str]] = None,
    include_providers: Optional[list[str]] = None,
    exclude_databases: Optional[list[str]] = None,
    http_client: Optional[
        Union[type[httpx.AsyncClient], type[requests.Session]]
    ] = None,
    verbose: int = 0,
    callbacks: Optional[list[Callable[[str, dict], Union[None, dict]]]] = None,
):
    """Create the OPTIMADE client object.

    Parameters:
        base_urls: A list of OPTIMADE base URLs to query.
        max_results_per_provider: The maximum number of results to download
            from each provider (-1 or 0 indicate unlimited).
        headers: Any additional HTTP headers to use for the queries.
        http_timeout: The timeout to use per request. Defaults to 10
            seconds with 1000 seconds for reads specifically. Overriding this value
            will replace all timeouts (connect, read, write and pool) with this value.
        max_attempts: The maximum number of times to repeat a failing query.
        use_async: Whether or not to make all requests asynchronously.
        exclude_providers: A set or collection of provider IDs to exclude from queries.
        include_providers: A set or collection of provider IDs to include in queries.
        exclude_databases: A set or collection of child database URLs to exclude from queries.
        http_client: An override for the underlying HTTP client, primarily used for testing.
        callbacks: A list of functions to call after each successful response, see the
            attribute [`OptimadeClient.callbacks`][optimade.client.OptimadeClient.callbacks]
            docstring for more details.

    """

    self.max_results_per_provider = max_results_per_provider
    if self.max_results_per_provider in (-1, 0):
        self.max_results_per_provider = None

    self._excluded_providers = set(exclude_providers) if exclude_providers else None
    self._included_providers = set(include_providers) if include_providers else None
    self._excluded_databases = set(exclude_databases) if exclude_databases else None

    self.max_attempts = max_attempts
    self.silent = silent

    if headers:
        self.headers.update(headers)

    if not base_urls:
        progress = None
        if not self.silent:
            progress = OptimadeClientProgress()
        self.base_urls = list(
            get_all_databases(
                exclude_providers=self._excluded_providers,
                include_providers=self._included_providers,
                exclude_databases=self._excluded_databases,
                progress=progress,
            )
        )
    else:
        if exclude_providers or include_providers or exclude_databases:
            raise RuntimeError(
                "Cannot provide both a list of base URLs and included/excluded databases."
            )

        self.base_urls = base_urls

        if isinstance(self.base_urls, str):
            self.base_urls = [self.base_urls]
        self.base_urls = list(self.base_urls)

    if not self.base_urls:
        raise SystemExit(
            "Unable to access any OPTIMADE base URLs. If you believe this is an error, try manually specifying some base URLs."
        )

    if http_timeout:
        if isinstance(http_timeout, httpx.Timeout):
            self.http_timeout = http_timeout
        else:
            self.http_timeout = httpx.Timeout(http_timeout)

    self.use_async = use_async

    if http_client:
        self._http_client = http_client
        if issubclass(self._http_client, httpx.AsyncClient):
            if not self.use_async and self.__strict_async:
                raise RuntimeError(
                    "Cannot use synchronous mode with an asynchronous HTTP client, please set `use_async=True` or pass an asynchronous HTTP client."
                )
            self.use_async = True
        elif issubclass(self._http_client, requests.Session):
            if self.use_async and self.__strict_async:
                raise RuntimeError(
                    "Cannot use async mode with a synchronous HTTP client, please set `use_async=False` or pass an synchronous HTTP client."
                )
            self.use_async = False
    else:
        if use_async:
            self._http_client = httpx.AsyncClient
        else:
            self._http_client = requests.Session

    self.callbacks = callbacks

count(filter=None, endpoint=None)

Counts the number of results for the filter, requiring only 1 request per provider by making use of the meta->data_returned key.

Raises:

Type	Description
RuntimeError	If the query could not be executed.

Returns:

Type	Description
dict[str, dict[str, dict[str, Optional[int]]]]	A nested mapping from endpoint, filter and base URL to the number of query results.

Source code in optimade/client/client.py

def count(
    self, filter: Optional[str] = None, endpoint: Optional[str] = None
) -> dict[str, dict[str, dict[str, Optional[int]]]]:
    """Counts the number of results for the filter, requiring
    only 1 request per provider by making use of the `meta->data_returned`
    key.

    Raises:
        RuntimeError: If the query could not be executed.

    Returns:
        A nested mapping from endpoint, filter and base URL to the number of query results.

    """

    if endpoint is None:
        if self.__current_endpoint is not None:
            endpoint = self.__current_endpoint
            self.__current_endpoint = None
        else:
            endpoint = "structures"

    if filter is None:
        filter = ""

    self._progress = OptimadeClientProgress()
    if self.silent:
        self._progress.disable = True

    self._check_filter(filter, endpoint)

    with self._progress:
        if not self.silent:
            self._progress.print(
                Panel(
                    f"Counting results for [bold yellow]{endpoint}[/bold yellow]/?filter=[bold magenta][i]{filter}[/i][/bold magenta]",
                    expand=False,
                )
            )
        results = self._execute_queries(
            filter,
            endpoint,
            page_limit=1,
            paginate=False,
            response_fields=[],
            sort=None,
        )
        count_results = {}

        for base_url in results:
            count_results[base_url] = results[base_url].meta.get(
                "data_returned", None
            )

            if count_results[base_url] is None:
                self._progress.print(
                    f"Warning: {base_url} did not return a value for `meta->data_returned`, unable to count results. Full response: {results[base_url]}"
                )

        self.count_results[endpoint][filter] = count_results
        return {endpoint: {filter: count_results}}

get(filter=None, endpoint=None, response_fields=None, sort=None)

Gets the results from the endpoint and filter across the defined OPTIMADE APIs.

Parameters:

Name	Type	Description	Default
filter	Optional[str]	The OPTIMADE filter string for the query.	None
endpoint	Optional[str]	The endpoint to query.	None
response_fields	Optional[list[str]]	A list of response fields to request from the server.	None
sort	Optional[str]	The field by which to sort the results.	None

Raises:

Type	Description
RuntimeError	If the query could not be executed.

Returns:

Type	Description
dict[str, dict[str, dict[str, dict]]]	A nested mapping from endpoint, filter and base URL to the query results.

Source code in optimade/client/client.py

def get(
    self,
    filter: Optional[str] = None,
    endpoint: Optional[str] = None,
    response_fields: Optional[list[str]] = None,
    sort: Optional[str] = None,
) -> dict[str, dict[str, dict[str, dict]]]:
    """Gets the results from the endpoint and filter across the
    defined OPTIMADE APIs.

    Parameters:
        filter: The OPTIMADE filter string for the query.
        endpoint: The endpoint to query.
        response_fields: A list of response fields to request
            from the server.
        sort: The field by which to sort the results.

    Raises:
        RuntimeError: If the query could not be executed.

    Returns:
        A nested mapping from endpoint, filter and base URL to the query results.

    """

    if endpoint is None:
        if self.__current_endpoint is not None:
            endpoint = self.__current_endpoint
            self.__current_endpoint = None
        else:
            endpoint = "structures"

    if filter is None:
        filter = ""

    self._progress = OptimadeClientProgress()
    if self.silent:
        self._progress.disable = True

    self._check_filter(filter, endpoint)

    with self._progress:
        if not self.silent:
            self._progress.print(
                Panel(
                    f"Performing query [bold yellow]{endpoint}[/bold yellow]/?filter=[bold magenta][i]{filter}[/i][/bold magenta]",
                    expand=False,
                )
            )
        results = self._execute_queries(
            filter,
            endpoint,
            response_fields=response_fields,
            page_limit=None,
            paginate=True,
            sort=sort,
        )
        self.all_results[endpoint][filter] = results
        return {endpoint: {filter: {k: results[k].asdict() for k in results}}}

get_one(endpoint, filter, base_url, response_fields=None, sort=None, page_limit=None, paginate=True)

Executes the query synchronously on one API.

Parameters:

Name	Type	Description	Default
endpoint	str	The OPTIMADE endpoint to query.	required
filter	str	The OPTIMADE filter string.	required
response_fields	Optional[list[str]]	A list of response fields to request from the server.	None
sort	Optional[str]	The field by which to sort the results.	None
page_limit	Optional[int]	A page limit to enforce for each query (used in conjunction with paginate).	None
paginate	bool	Whether to pull all pages of results (up to the value of max_results_per_provider) or whether to return after one page.	True

Returns:

Type	Description
dict[str, QueryResults]	A dictionary mapping from base URL to the results of the query.

Source code in optimade/client/client.py

def get_one(
    self,
    endpoint: str,
    filter: str,
    base_url: str,
    response_fields: Optional[list[str]] = None,
    sort: Optional[str] = None,
    page_limit: Optional[int] = None,
    paginate: bool = True,
) -> dict[str, QueryResults]:
    """Executes the query synchronously on one API.

    Parameters:
        endpoint: The OPTIMADE endpoint to query.
        filter: The OPTIMADE filter string.
        response_fields: A list of response fields to request
            from the server.
        sort: The field by which to sort the results.
        page_limit: A page limit to enforce for each query (used in
            conjunction with `paginate`).
        paginate: Whether to pull all pages of results (up to the
            value of `max_results_per_provider`) or whether to return
            after one page.

    Returns:
        A dictionary mapping from base URL to the results of the query.

    """
    try:
        return self._get_one(
            endpoint,
            filter,
            base_url,
            page_limit=page_limit,
            paginate=paginate,
            response_fields=response_fields,
            sort=sort,
        )
    except Exception as exc:
        error_query_results = QueryResults()
        error_query_results.errors = [
            f"{exc.__class__.__name__}: {str(exc.args[0])}"
        ]
        self._progress.print(
            f"[red]Error[/red]: Provider {str(base_url)!r} returned: [red i]{exc}[/red i]"
        )
        return {base_url: error_query_results}

get_one_async(endpoint, filter, base_url, response_fields=None, sort=None, page_limit=None, paginate=True) async

Executes the query asynchronously on one API.

Note

This method currently makes non-blocking requests to a single API, but these requests are executed serially on that API, i.e., results are pulled one page at a time, but requests will not block other async requests to other APIs.

Parameters:

Name	Type	Description	Default
endpoint	str	The OPTIMADE endpoint to query.	required
filter	str	The OPTIMADE filter string.	required
response_fields	Optional[list[str]]	A list of response fields to request from the server.	None
sort	Optional[str]	The field by which to sort the results.	None
page_limit	Optional[int]	A page limit to enforce for each query (used in conjunction with paginate).	None
paginate	bool	Whether to pull all pages of results (up to the value of max_results_per_provider) or whether to return after one page.	True

Returns:

Type	Description
dict[str, QueryResults]	A dictionary mapping from base URL to the results of the query.

Source code in optimade/client/client.py

async def get_one_async(
    self,
    endpoint: str,
    filter: str,
    base_url: str,
    response_fields: Optional[list[str]] = None,
    sort: Optional[str] = None,
    page_limit: Optional[int] = None,
    paginate: bool = True,
) -> dict[str, QueryResults]:
    """Executes the query asynchronously on one API.

    !!! note
        This method currently makes non-blocking requests
        to a single API, but these requests are executed
        serially on that API, i.e., results are pulled one
        page at a time, but requests will not block other
        async requests to other APIs.

    Parameters:
        endpoint: The OPTIMADE endpoint to query.
        filter: The OPTIMADE filter string.
        response_fields: A list of response fields to request
            from the server.
        sort: The field by which to sort the results.
        page_limit: A page limit to enforce for each query (used in
            conjunction with `paginate`).
        paginate: Whether to pull all pages of results (up to the
            value of `max_results_per_provider`) or whether to return
            after one page.

    Returns:
        A dictionary mapping from base URL to the results of the query.

    """
    try:
        return await self._get_one_async(
            endpoint,
            filter,
            base_url,
            page_limit=page_limit,
            paginate=paginate,
            response_fields=response_fields,
            sort=sort,
        )
    except Exception as exc:
        error_query_results = QueryResults()
        error_query_results.errors = [
            f"{exc.__class__.__name__}: {str(exc.args[0])}"
        ]
        self._progress.print(
            f"[red]Error[/red]: Provider {str(base_url)!r} returned: [red i]{error_query_results.errors}[/red i]"
        )
        return {base_url: error_query_results}

list_properties(entry_type)

Returns the list of properties reported at /info/<entry_type> for the given entry type, for each database.

Source code in optimade/client/client.py

def list_properties(
    self,
    entry_type: str,
) -> dict[str, list[str]]:
    """Returns the list of properties reported at `/info/<entry_type>`
    for the given entry type, for each database.

    """
    self._progress = OptimadeClientProgress()
    if self.silent:
        self._progress.disable = True

    with self._progress:
        if not self.silent:
            self._progress.print(
                Panel(
                    f"Listing properties for [bold yellow]{entry_type}[/bold yellow]",
                    expand=False,
                )
            )
        results = self._execute_queries(
            "",
            f"info/{entry_type}",
            paginate=False,
            page_limit=1,
            response_fields=[],
            sort=None,
        )
    self.property_lists = {entry_type: {}}
    for database in results:
        self.property_lists[entry_type][database] = list(
            results[database].data.get("properties", {}).keys()  # type: ignore
        )
    return self.property_lists[entry_type]

search_property(query, entry_type)

Searches for the query substring within the listed properties served by each database.

Parameters:

Name	Type	Description	Default
query	str	The substring to search for.	required
entry_type	str	The entry type to query.	required

Returns:

Type	Description
dict[str, list[str]]	A nested dictionary of matching property lists, arranged by
dict[str, list[str]]	entry type and database.

Source code in optimade/client/client.py

def search_property(self, query: str, entry_type: str) -> dict[str, list[str]]:
    """Searches for the query substring within the listed properties
    served by each database.

    Parameters:
        query: The substring to search for.
        entry_type: The entry type to query.

    Returns:
        A nested dictionary of matching property lists, arranged by
        entry type and database.

    """
    if not self.property_lists:
        self.list_properties(entry_type=entry_type)

    matching_properties: dict[str, dict[str, list[str]]] = {
        entry_type: defaultdict(list)
    }
    if entry_type in self.property_lists:
        for database in self.property_lists[entry_type]:
            for property in self.property_lists[entry_type][database]:
                if query in property:
                    matching_properties[entry_type][database].append(property)
    return matching_properties[entry_type]

OptimadeClientProgress

Bases: Progress

A wrapper around Rich.Progress that defines the OPTIMADE client progressbars.

Source code in optimade/client/utils.py

class OptimadeClientProgress(Progress):
    """A wrapper around `Rich.Progress` that defines the OPTIMADE client progressbars."""

    def __init__(self):
        super().__init__(
            SpinnerColumn(finished_text="[green]✓"),
            TextColumn("[progress.description]{task.description}"),
            BarColumn(),
            TaskProgressColumn(
                text_format="[progress.completed]{task.completed}/[progress.total]{task.total}",
                text_format_no_percentage="[progress.completed]{task.completed}",
            ),
            TimeElapsedColumn(),
            console=Console(stderr=True),
            auto_refresh=True,
            refresh_per_second=10,
        )

    def print(self, *args, **kwargs):
        if not self.disable:
            super().print(*args, **kwargs)

QueryResults dataclass

A container dataclass for the results from a given query.

Source code in optimade/client/utils.py

@dataclass
class QueryResults:
    """A container dataclass for the results from a given query."""

    data: Union[dict, list[dict]] = field(default_factory=list, init=False)  # type: ignore[assignment]
    errors: list[str] = field(default_factory=list, init=False)
    links: dict = field(default_factory=dict, init=False)
    included: list[dict] = field(default_factory=list, init=False)
    meta: dict = field(default_factory=dict, init=False)

    @property
    def included_index(self) -> set[str]:
        if not getattr(self, "_included_index", None):
            self._included_index: set[str] = set()
        return self._included_index

    def asdict(self):
        return asdict(self)

    def update(self, page_results: dict) -> None:
        """Combine the results from one page with the existing results for a given query.

        Parameters:
            page_results: The results for the current page.

        """

        if "data" in page_results:
            # If the `data` field is a list, add it to our existing results.
            # Otherwise, as is the case for `info` endpoints, `data` is a dictionary (or null)
            # and should be added as the only `data` field for these results.
            if isinstance(page_results["data"], list):
                self.data.extend(page_results["data"])  # type: ignore[union-attr]
            elif not self.data:
                self.data = page_results["data"]
            else:
                raise RuntimeError(
                    "Not overwriting old `data` field in `QueryResults`."
                )

        if "errors" in page_results:
            self.errors.extend(page_results["errors"])

        # Combine meta/links fields across all pages in a sensible way, i.e.,
        # if we really reached the last page of results, then make sure `links->next`
        # is null in the final response, and make sure `meta->more_data_available` is None or False.
        keys_to_filter = {
            "links": ("next", "prev"),
            "meta": ("query", "more_data_available"),
        }
        for top_level_key in keys_to_filter:
            if top_level_key not in page_results:
                page_results[top_level_key] = {}
            for k in keys_to_filter[top_level_key]:
                if k not in page_results[top_level_key]:
                    page_results[top_level_key][k] = None
            getattr(self, top_level_key).update(
                {k: page_results[top_level_key][k] for k in page_results[top_level_key]}
            )

        # Only add new unique entries to the included list
        for d in page_results.get("included", []):
            typed_id = f"{d['type']}/{d['id']}"
            if typed_id not in self.included_index:
                self.included_index.add(typed_id)
                self.included.append(d)

update(page_results)

Combine the results from one page with the existing results for a given query.

Parameters:

Name	Type	Description	Default
page_results	dict	The results for the current page.	required

Source code in optimade/client/utils.py

def update(self, page_results: dict) -> None:
    """Combine the results from one page with the existing results for a given query.

    Parameters:
        page_results: The results for the current page.

    """

    if "data" in page_results:
        # If the `data` field is a list, add it to our existing results.
        # Otherwise, as is the case for `info` endpoints, `data` is a dictionary (or null)
        # and should be added as the only `data` field for these results.
        if isinstance(page_results["data"], list):
            self.data.extend(page_results["data"])  # type: ignore[union-attr]
        elif not self.data:
            self.data = page_results["data"]
        else:
            raise RuntimeError(
                "Not overwriting old `data` field in `QueryResults`."
            )

    if "errors" in page_results:
        self.errors.extend(page_results["errors"])

    # Combine meta/links fields across all pages in a sensible way, i.e.,
    # if we really reached the last page of results, then make sure `links->next`
    # is null in the final response, and make sure `meta->more_data_available` is None or False.
    keys_to_filter = {
        "links": ("next", "prev"),
        "meta": ("query", "more_data_available"),
    }
    for top_level_key in keys_to_filter:
        if top_level_key not in page_results:
            page_results[top_level_key] = {}
        for k in keys_to_filter[top_level_key]:
            if k not in page_results[top_level_key]:
                page_results[top_level_key][k] = None
        getattr(self, top_level_key).update(
            {k: page_results[top_level_key][k] for k in page_results[top_level_key]}
        )

    # Only add new unique entries to the included list
    for d in page_results.get("included", []):
        typed_id = f"{d['type']}/{d['id']}"
        if typed_id not in self.included_index:
            self.included_index.add(typed_id)
            self.included.append(d)

RecoverableHTTPError

Bases: Exception

Base class for any HTTP issues that may be recoverable by just repeating the query.

Source code in optimade/client/utils.py

class RecoverableHTTPError(Exception):
    """Base class for any HTTP issues that may be recoverable by just
    repeating the query."""

TooManyRequestsException

Bases: RecoverableHTTPError

For when the underlying HTTP request returns 429: Too Many Requests.

Source code in optimade/client/utils.py

class TooManyRequestsException(RecoverableHTTPError):
    """For when the underlying HTTP request returns 429: Too Many Requests."""

get_all_databases(include_providers=None, exclude_providers=None, exclude_databases=None, progress=None)

Iterate through all databases reported by registered OPTIMADE providers.

Parameters:

Name	Type	Description	Default
include_providers	Optional[Container[str]]	A set/container of provider IDs to include child databases for.	None
exclude_providers	Optional[Container[str]]	A set/container of provider IDs to exclude child databases for.	None
exclude_databases	Optional[Container[str]]	A set/container of specific database URLs to exclude.	None

Returns:

Type	Description
Iterable[str]	A generator of child database links that obey the given parameters.

Source code in optimade/utils.py

def get_all_databases(
    include_providers: Optional[Container[str]] = None,
    exclude_providers: Optional[Container[str]] = None,
    exclude_databases: Optional[Container[str]] = None,
    progress: "Optional[rich.Progress]" = None,
) -> Iterable[str]:
    """Iterate through all databases reported by registered OPTIMADE providers.

    Parameters:
        include_providers: A set/container of provider IDs to include child databases for.
        exclude_providers: A set/container of provider IDs to exclude child databases for.
        exclude_databases: A set/container of specific database URLs to exclude.

    Returns:
        A generator of child database links that obey the given parameters.

    """
    if progress is not None:
        _task = progress.add_task(
            description="Retrieving all databases from registered OPTIMADE providers...",
            total=None,
        )
    else:
        progress = contextlib.nullcontext()
        progress.print = lambda _: None  # type: ignore[attr-defined]
        progress.advance = lambda *_: None  # type: ignore[attr-defined]
        _task = None

    with progress:
        for provider in get_providers():
            if exclude_providers and provider["id"] in exclude_providers:
                continue
            if include_providers and provider["id"] not in include_providers:
                continue

            try:
                links = get_child_database_links(provider)
                for link in links:
                    if link.attributes.base_url:
                        if (
                            exclude_databases
                            and link.attributes.base_url in exclude_databases
                        ):
                            continue
                        yield str(link.attributes.base_url)
                if links and progress is not None:
                    progress.advance(_task, 1)
                    progress.print(
                        f"Retrieved databases from [bold green]{provider['id']}[/bold green]"
                    )
            except RuntimeError as exc:
                if progress is not None:
                    progress.print(
                        f"Unable to retrieve databases from [bold red]{provider['id']}[/bold red]: {exc}",
                    )
                pass

silent_raise()

Raise an exception without printing a traceback, or the exception message itself.

Source code in optimade/client/utils.py

@contextmanager
def silent_raise():
    """Raise an exception without printing a traceback, or the exception message itself."""
    default_value = getattr(
        sys, "tracebacklimit", 1000
    )  # `1000` is a Python's default value
    default_excepthook = getattr(sys, "excepthook")
    sys.tracebacklimit = 0
    sys.excepthook = lambda type, value, traceback: None
    yield
    sys.tracebacklimit = default_value  # revert changes
    sys.excepthook = default_excepthook