utils

BASE_URL_PREFIXES = {'major': f'/v{__api_version__.split('-')[0].split('+')[0].split('.')[0]}', 'minor': f'/v{'.'.join(__api_version__.split('-')[0].split('+')[0].split('.')[:2])}', 'patch': f'/v{'.'.join(__api_version__.split('-')[0].split('+')[0].split('.')[:3])}'} module-attribute

CONFIG: ServerConfig = ServerConfig() module-attribute

This singleton loads the config from a hierarchy of sources (see customise_sources) and makes it importable in the server code.

PROVIDER_LIST_URLS = ('https://providers.optimade.org/v1/links', 'https://raw.githubusercontent.com/Materials-Consortia/providers/master/src/links/v1/providers.json') module-attribute

__all__ = ('BASE_URL_PREFIXES', 'meta_values', 'handle_response_fields', 'get_included_relationships', 'get_base_url', 'get_entries', 'get_single_entry', 'mongo_id_for_database', 'get_providers', 'PROVIDER_LIST_URLS') module-attribute

__api_version__ = '1.1.0' 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"

EntryCollection

Bases: ABC

Backend-agnostic base class for querying collections of EntryResources.

Source code in optimade/server/entry_collections/entry_collections.py

class EntryCollection(ABC):
    """Backend-agnostic base class for querying collections of
    [`EntryResource`][optimade.models.entries.EntryResource]s."""

    pagination_mechanism = PaginationMechanism("page_offset")
    """The default pagination mechansim to use with a given collection,
    if the user does not provide any pagination query parameters.
    """

    def __init__(
        self,
        resource_cls: type[EntryResource],
        resource_mapper: type[BaseResourceMapper],
        transformer: Transformer,
    ):
        """Initialize the collection for the given parameters.

        Parameters:
            resource_cls (EntryResource): The `EntryResource` model
                that is stored by the collection.
            resource_mapper (BaseResourceMapper): A resource mapper
                object that handles aliases and format changes between
                deserialization and response.
            transformer (Transformer): The Lark `Transformer` used to
                interpret the filter.

        """
        self.parser = LarkParser()
        self.resource_cls = resource_cls
        self.resource_mapper = resource_mapper
        self.transformer = transformer

        self.provider_prefix = CONFIG.provider.prefix
        self.provider_fields = [
            field if isinstance(field, str) else field["name"]
            for field in CONFIG.provider_fields.get(resource_mapper.ENDPOINT, [])
        ]

        self._all_fields: set[str] = set()

    @abstractmethod
    def __len__(self) -> int:
        """Returns the total number of entries in the collection."""

    @abstractmethod
    def insert(self, data: list[EntryResource]) -> None:
        """Add the given entries to the underlying database.

        Arguments:
            data: The entry resource objects to add to the database.

        """

    @abstractmethod
    def count(self, **kwargs: Any) -> Optional[int]:
        """Returns the number of entries matching the query specified
        by the keyword arguments.

        Parameters:
            **kwargs: Query parameters as keyword arguments.

        """

    def find(
        self, params: Union[EntryListingQueryParams, SingleEntryQueryParams]
    ) -> tuple[
        Optional[Union[dict[str, Any], list[dict[str, Any]]]],
        Optional[int],
        bool,
        set[str],
        set[str],
    ]:
        """
        Fetches results and indicates if more data is available.

        Also gives the total number of data available in the absence of `page_limit`.
        See [`EntryListingQueryParams`][optimade.server.query_params.EntryListingQueryParams]
        for more information.

        Returns a list of the mapped database reponse.

        If no results match the query, then `results` is set to `None`.

        Parameters:
            params: Entry listing URL query params.

        Returns:
            A tuple of various relevant values:
            (`results`, `data_returned`, `more_data_available`, `exclude_fields`, `include_fields`).

        """
        criteria = self.handle_query_params(params)
        single_entry = isinstance(params, SingleEntryQueryParams)
        response_fields: set[str] = criteria.pop("fields")

        raw_results, data_returned, more_data_available = self._run_db_query(
            criteria, single_entry
        )

        exclude_fields = self.all_fields - response_fields
        include_fields = (
            response_fields - self.resource_mapper.TOP_LEVEL_NON_ATTRIBUTES_FIELDS
        )

        bad_optimade_fields: set[str] = set()
        bad_provider_fields: set[str] = set()
        supported_prefixes = self.resource_mapper.SUPPORTED_PREFIXES
        all_attributes: set[str] = self.resource_mapper.ALL_ATTRIBUTES
        for field in include_fields:
            if field not in all_attributes:
                if field.startswith("_"):
                    if any(
                        field.startswith(f"_{prefix}_") for prefix in supported_prefixes
                    ):
                        bad_provider_fields.add(field)
                else:
                    bad_optimade_fields.add(field)

        if bad_provider_fields:
            warnings.warn(
                message=f"Unrecognised field(s) for this provider requested in `response_fields`: {bad_provider_fields}.",
                category=UnknownProviderProperty,
            )

        if bad_optimade_fields:
            raise BadRequest(
                detail=f"Unrecognised OPTIMADE field(s) in requested `response_fields`: {bad_optimade_fields}."
            )

        results: Optional[Union[list[dict[str, Any]], dict[str, Any]]] = None

        if raw_results:
            results = [self.resource_mapper.map_back(doc) for doc in raw_results]

            if single_entry:
                results = results[0]

                if (
                    CONFIG.validate_api_response
                    and data_returned is not None
                    and data_returned > 1
                ):
                    raise NotFound(
                        detail=f"Instead of a single entry, {data_returned} entries were found",
                    )
                else:
                    data_returned = 1

        return (
            results,
            data_returned,
            more_data_available,
            exclude_fields,
            include_fields,
        )

    @abstractmethod
    def _run_db_query(
        self, criteria: dict[str, Any], single_entry: bool = False
    ) -> tuple[list[dict[str, Any]], Optional[int], bool]:
        """Run the query on the backend and collect the results.

        Arguments:
            criteria: A dictionary representation of the query parameters.
            single_entry: Whether or not the caller is expecting a single entry response.

        Returns:
            The list of entries from the database (without any re-mapping), the total number of
            entries matching the query and a boolean for whether or not there is more data available.

        """

    @property
    def all_fields(self) -> set[str]:
        """Get the set of all fields handled in this collection,
        from attribute fields in the schema, provider fields and top-level OPTIMADE fields.

        The set of all fields are lazily created and then cached.
        This means the set is created the first time the property is requested and then cached.

        Returns:
            All fields handled in this collection.

        """
        if not self._all_fields:
            # All OPTIMADE fields
            self._all_fields = (
                self.resource_mapper.TOP_LEVEL_NON_ATTRIBUTES_FIELDS.copy()
            )
            self._all_fields |= self.get_attribute_fields()
            # All provider-specific fields
            self._all_fields |= {
                f"_{self.provider_prefix}_{field_name}"
                if not field_name.startswith("_")
                else field_name
                for field_name in self.provider_fields
            }

        return self._all_fields

    def get_attribute_fields(self) -> set[str]:
        """Get the set of attribute fields

        Return only the _first-level_ attribute fields from the schema of the resource class,
        resolving references along the way if needed.

        Note:
            It is not needed to take care of other special OpenAPI schema keys than `allOf`,
            since only `allOf` will be found in this context.
            Other special keys can be found in [the Swagger documentation](https://swagger.io/docs/specification/data-models/oneof-anyof-allof-not/).

        Returns:
            Property names.

        """
        annotation = _get_origin_type(
            self.resource_cls.model_fields["attributes"].annotation
        )

        if annotation in (None, NoneType) or not issubclass(annotation, Attributes):
            raise TypeError(
                "resource class 'attributes' field must be a subclass of 'EntryResourceAttributes'"
            )

        return set(annotation.model_fields)  # type: ignore[attr-defined]

    def handle_query_params(
        self, params: Union[EntryListingQueryParams, SingleEntryQueryParams]
    ) -> dict[str, Any]:
        """Parse and interpret the backend-agnostic query parameter models into a dictionary
        that can be used by the specific backend.

        Note:
            Currently this method returns the pymongo interpretation of the parameters,
            which will need modification for modified for other backends.

        Parameters:
            params: The initialized query parameter model from the server.

        Raises:
            Forbidden: If too large of a page limit is provided.
            BadRequest: If an invalid request is made, e.g., with incorrect fields
                or response format.

        Returns:
            A dictionary representation of the query parameters.

        """
        cursor_kwargs = {}

        # filter
        if getattr(params, "filter", False):
            cursor_kwargs["filter"] = self.transformer.transform(
                self.parser.parse(params.filter)  # type: ignore[union-attr]
            )
        else:
            cursor_kwargs["filter"] = {}

        # response_format
        if (
            getattr(params, "response_format", False)
            and params.response_format != "json"
        ):
            raise BadRequest(
                detail=f"Response format {params.response_format} is not supported, please use response_format='json'"
            )

        # page_limit
        if getattr(params, "page_limit", False):
            limit = params.page_limit  # type: ignore[union-attr]
            if limit > CONFIG.page_limit_max:
                raise Forbidden(
                    detail=f"Max allowed page_limit is {CONFIG.page_limit_max}, you requested {limit}",
                )
            cursor_kwargs["limit"] = limit
        else:
            cursor_kwargs["limit"] = CONFIG.page_limit

        # response_fields
        cursor_kwargs["projection"] = {
            f"{self.resource_mapper.get_backend_field(f)}": True
            for f in self.all_fields
        }

        if getattr(params, "response_fields", False):
            response_fields = set(params.response_fields.split(","))
            response_fields |= self.resource_mapper.get_required_fields()
        else:
            response_fields = self.all_fields.copy()

        cursor_kwargs["fields"] = response_fields

        # sort
        if getattr(params, "sort", False):
            cursor_kwargs["sort"] = self.parse_sort_params(params.sort)  # type: ignore[union-attr]

        # warn if multiple pagination keys are present, and only use the first from this list
        received_pagination_option = False
        warn_multiple_keys = False

        if getattr(params, "page_offset", False):
            received_pagination_option = True
            cursor_kwargs["skip"] = params.page_offset  # type: ignore[union-attr]

        if isinstance(getattr(params, "page_number", None), int):
            if received_pagination_option:
                warn_multiple_keys = True
            else:
                received_pagination_option = True
                if params.page_number < 1:  # type: ignore[union-attr]
                    warnings.warn(
                        message=f"'page_number' is 1-based, using 'page_number=1' instead of {params.page_number}",  # type: ignore[union-attr]
                        category=QueryParamNotUsed,
                    )
                    page_number = 1
                else:
                    page_number = params.page_number  # type: ignore[union-attr]
                cursor_kwargs["skip"] = (page_number - 1) * cursor_kwargs["limit"]

        if isinstance(getattr(params, "page_above", None), str):
            if received_pagination_option:
                warn_multiple_keys = True
            else:
                received_pagination_option = True
                cursor_kwargs["page_above"] = params.page_above  # type: ignore[union-attr]

        if warn_multiple_keys:
            warnings.warn(
                message="Multiple pagination keys were provided, only using the first one of 'page_offset', 'page_number' or 'page_above'",
                category=QueryParamNotUsed,
            )

        return cursor_kwargs

    def parse_sort_params(self, sort_params: str) -> Iterable[tuple[str, int]]:
        """Handles any sort parameters passed to the collection,
        resolving aliases and dealing with any invalid fields.

        Raises:
            BadRequest: if an invalid sort is requested.

        Returns:
            A list of tuples containing the aliased field name and
            sort direction encoded as 1 (ascending) or -1 (descending).

        """
        sort_spec: list[tuple[str, int]] = []
        for field in sort_params.split(","):
            sort_dir = 1
            if field.startswith("-"):
                field = field[1:]
                sort_dir = -1
            aliased_field = self.resource_mapper.get_backend_field(field)
            sort_spec.append((aliased_field, sort_dir))

        unknown_fields = [
            field
            for field, _ in sort_spec
            if self.resource_mapper.get_optimade_field(field) not in self.all_fields
        ]

        if unknown_fields:
            error_detail = "Unable to sort on unknown field{} '{}'".format(
                "s" if len(unknown_fields) > 1 else "",
                "', '".join(unknown_fields),
            )

            # If all unknown fields are "other" provider-specific, then only provide a warning
            if all(
                (
                    re.match(r"_[a-z_0-9]+_[a-z_0-9]*", field)
                    and not field.startswith(f"_{self.provider_prefix}_")
                )
                for field in unknown_fields
            ):
                warnings.warn(error_detail, FieldValueNotRecognized)

            # Otherwise, if all fields are unknown, or some fields are unknown and do not
            # have other provider prefixes, then return 400: Bad Request
            else:
                raise BadRequest(detail=error_detail)

        # If at least one valid field has been provided for sorting, then use that
        sort_spec = [
            (field, sort_dir)
            for field, sort_dir in sort_spec
            if field not in unknown_fields
        ]

        return sort_spec

    def get_next_query_params(
        self,
        params: EntryListingQueryParams,
        results: Optional[Union[dict[str, Any], list[dict[str, Any]]]],
    ) -> dict[str, list[str]]:
        """Provides url query pagination parameters that will be used in the next
        link.

        Arguments:
            results: The results produced by find.
            params: The parsed request params produced by handle_query_params.

        Returns:
            A dictionary with the necessary query parameters.

        """
        query: dict[str, list[str]] = dict()
        if isinstance(results, list) and results:
            # If a user passed a particular pagination mechanism, keep using it
            # Otherwise, use the default pagination mechanism of the collection
            pagination_mechanism = PaginationMechanism.OFFSET
            for pagination_key in (
                "page_offset",
                "page_number",
                "page_above",
            ):
                if getattr(params, pagination_key, None) is not None:
                    pagination_mechanism = PaginationMechanism(pagination_key)
                    break

            if pagination_mechanism == PaginationMechanism.OFFSET:
                query["page_offset"] = [
                    str(params.page_offset + len(results))  # type: ignore[list-item]
                ]

        return query

all_fields: set[str] property

Get the set of all fields handled in this collection, from attribute fields in the schema, provider fields and top-level OPTIMADE fields.

The set of all fields are lazily created and then cached. This means the set is created the first time the property is requested and then cached.

Returns:

Type	Description
set[str]	All fields handled in this collection.

pagination_mechanism = PaginationMechanism('page_offset') class-attribute instance-attribute

The default pagination mechansim to use with a given collection, if the user does not provide any pagination query parameters.

__init__(resource_cls, resource_mapper, transformer)

Initialize the collection for the given parameters.

Parameters:

Name	Type	Description	Default
resource_cls	EntryResource	The EntryResource model that is stored by the collection.	required
resource_mapper	BaseResourceMapper	A resource mapper object that handles aliases and format changes between deserialization and response.	required
transformer	Transformer	The Lark Transformer used to interpret the filter.	required

Source code in optimade/server/entry_collections/entry_collections.py

def __init__(
    self,
    resource_cls: type[EntryResource],
    resource_mapper: type[BaseResourceMapper],
    transformer: Transformer,
):
    """Initialize the collection for the given parameters.

    Parameters:
        resource_cls (EntryResource): The `EntryResource` model
            that is stored by the collection.
        resource_mapper (BaseResourceMapper): A resource mapper
            object that handles aliases and format changes between
            deserialization and response.
        transformer (Transformer): The Lark `Transformer` used to
            interpret the filter.

    """
    self.parser = LarkParser()
    self.resource_cls = resource_cls
    self.resource_mapper = resource_mapper
    self.transformer = transformer

    self.provider_prefix = CONFIG.provider.prefix
    self.provider_fields = [
        field if isinstance(field, str) else field["name"]
        for field in CONFIG.provider_fields.get(resource_mapper.ENDPOINT, [])
    ]

    self._all_fields: set[str] = set()

__len__() abstractmethod

Returns the total number of entries in the collection.

Source code in optimade/server/entry_collections/entry_collections.py

@abstractmethod
def __len__(self) -> int:
    """Returns the total number of entries in the collection."""

count(**kwargs) abstractmethod

Returns the number of entries matching the query specified by the keyword arguments.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Query parameters as keyword arguments.	{}

Source code in optimade/server/entry_collections/entry_collections.py

@abstractmethod
def count(self, **kwargs: Any) -> Optional[int]:
    """Returns the number of entries matching the query specified
    by the keyword arguments.

    Parameters:
        **kwargs: Query parameters as keyword arguments.

    """

find(params)

Fetches results and indicates if more data is available.

Also gives the total number of data available in the absence of page_limit. See EntryListingQueryParams for more information.

Returns a list of the mapped database reponse.

If no results match the query, then results is set to None.

Parameters:

Name	Type	Description	Default
params	Union[EntryListingQueryParams, SingleEntryQueryParams]	Entry listing URL query params.	required

Returns:

Type	Description
Optional[Union[dict[str, Any], list[dict[str, Any]]]]	A tuple of various relevant values:
Optional[int]	(results, data_returned, more_data_available, exclude_fields, include_fields).

Source code in optimade/server/entry_collections/entry_collections.py

def find(
    self, params: Union[EntryListingQueryParams, SingleEntryQueryParams]
) -> tuple[
    Optional[Union[dict[str, Any], list[dict[str, Any]]]],
    Optional[int],
    bool,
    set[str],
    set[str],
]:
    """
    Fetches results and indicates if more data is available.

    Also gives the total number of data available in the absence of `page_limit`.
    See [`EntryListingQueryParams`][optimade.server.query_params.EntryListingQueryParams]
    for more information.

    Returns a list of the mapped database reponse.

    If no results match the query, then `results` is set to `None`.

    Parameters:
        params: Entry listing URL query params.

    Returns:
        A tuple of various relevant values:
        (`results`, `data_returned`, `more_data_available`, `exclude_fields`, `include_fields`).

    """
    criteria = self.handle_query_params(params)
    single_entry = isinstance(params, SingleEntryQueryParams)
    response_fields: set[str] = criteria.pop("fields")

    raw_results, data_returned, more_data_available = self._run_db_query(
        criteria, single_entry
    )

    exclude_fields = self.all_fields - response_fields
    include_fields = (
        response_fields - self.resource_mapper.TOP_LEVEL_NON_ATTRIBUTES_FIELDS
    )

    bad_optimade_fields: set[str] = set()
    bad_provider_fields: set[str] = set()
    supported_prefixes = self.resource_mapper.SUPPORTED_PREFIXES
    all_attributes: set[str] = self.resource_mapper.ALL_ATTRIBUTES
    for field in include_fields:
        if field not in all_attributes:
            if field.startswith("_"):
                if any(
                    field.startswith(f"_{prefix}_") for prefix in supported_prefixes
                ):
                    bad_provider_fields.add(field)
            else:
                bad_optimade_fields.add(field)

    if bad_provider_fields:
        warnings.warn(
            message=f"Unrecognised field(s) for this provider requested in `response_fields`: {bad_provider_fields}.",
            category=UnknownProviderProperty,
        )

    if bad_optimade_fields:
        raise BadRequest(
            detail=f"Unrecognised OPTIMADE field(s) in requested `response_fields`: {bad_optimade_fields}."
        )

    results: Optional[Union[list[dict[str, Any]], dict[str, Any]]] = None

    if raw_results:
        results = [self.resource_mapper.map_back(doc) for doc in raw_results]

        if single_entry:
            results = results[0]

            if (
                CONFIG.validate_api_response
                and data_returned is not None
                and data_returned > 1
            ):
                raise NotFound(
                    detail=f"Instead of a single entry, {data_returned} entries were found",
                )
            else:
                data_returned = 1

    return (
        results,
        data_returned,
        more_data_available,
        exclude_fields,
        include_fields,
    )

get_attribute_fields()

Get the set of attribute fields

Return only the first-level attribute fields from the schema of the resource class, resolving references along the way if needed.

Note

It is not needed to take care of other special OpenAPI schema keys than allOf, since only allOf will be found in this context. Other special keys can be found in the Swagger documentation.

Returns:

Type	Description
set[str]	Property names.

Source code in optimade/server/entry_collections/entry_collections.py

def get_attribute_fields(self) -> set[str]:
    """Get the set of attribute fields

    Return only the _first-level_ attribute fields from the schema of the resource class,
    resolving references along the way if needed.

    Note:
        It is not needed to take care of other special OpenAPI schema keys than `allOf`,
        since only `allOf` will be found in this context.
        Other special keys can be found in [the Swagger documentation](https://swagger.io/docs/specification/data-models/oneof-anyof-allof-not/).

    Returns:
        Property names.

    """
    annotation = _get_origin_type(
        self.resource_cls.model_fields["attributes"].annotation
    )

    if annotation in (None, NoneType) or not issubclass(annotation, Attributes):
        raise TypeError(
            "resource class 'attributes' field must be a subclass of 'EntryResourceAttributes'"
        )

    return set(annotation.model_fields)  # type: ignore[attr-defined]

get_next_query_params(params, results)

Provides url query pagination parameters that will be used in the next link.

Parameters:

Name	Type	Description	Default
results	Optional[Union[dict[str, Any], list[dict[str, Any]]]]	The results produced by find.	required
params	EntryListingQueryParams	The parsed request params produced by handle_query_params.	required

Returns:

Type	Description
dict[str, list[str]]	A dictionary with the necessary query parameters.

Source code in optimade/server/entry_collections/entry_collections.py

def get_next_query_params(
    self,
    params: EntryListingQueryParams,
    results: Optional[Union[dict[str, Any], list[dict[str, Any]]]],
) -> dict[str, list[str]]:
    """Provides url query pagination parameters that will be used in the next
    link.

    Arguments:
        results: The results produced by find.
        params: The parsed request params produced by handle_query_params.

    Returns:
        A dictionary with the necessary query parameters.

    """
    query: dict[str, list[str]] = dict()
    if isinstance(results, list) and results:
        # If a user passed a particular pagination mechanism, keep using it
        # Otherwise, use the default pagination mechanism of the collection
        pagination_mechanism = PaginationMechanism.OFFSET
        for pagination_key in (
            "page_offset",
            "page_number",
            "page_above",
        ):
            if getattr(params, pagination_key, None) is not None:
                pagination_mechanism = PaginationMechanism(pagination_key)
                break

        if pagination_mechanism == PaginationMechanism.OFFSET:
            query["page_offset"] = [
                str(params.page_offset + len(results))  # type: ignore[list-item]
            ]

    return query

handle_query_params(params)

Parse and interpret the backend-agnostic query parameter models into a dictionary that can be used by the specific backend.

Note

Currently this method returns the pymongo interpretation of the parameters, which will need modification for modified for other backends.

Parameters:

Name	Type	Description	Default
params	Union[EntryListingQueryParams, SingleEntryQueryParams]	The initialized query parameter model from the server.	required

Raises:

Type	Description
Forbidden	If too large of a page limit is provided.
BadRequest	If an invalid request is made, e.g., with incorrect fields or response format.

Returns:

Type	Description
dict[str, Any]	A dictionary representation of the query parameters.

Source code in optimade/server/entry_collections/entry_collections.py

def handle_query_params(
    self, params: Union[EntryListingQueryParams, SingleEntryQueryParams]
) -> dict[str, Any]:
    """Parse and interpret the backend-agnostic query parameter models into a dictionary
    that can be used by the specific backend.

    Note:
        Currently this method returns the pymongo interpretation of the parameters,
        which will need modification for modified for other backends.

    Parameters:
        params: The initialized query parameter model from the server.

    Raises:
        Forbidden: If too large of a page limit is provided.
        BadRequest: If an invalid request is made, e.g., with incorrect fields
            or response format.

    Returns:
        A dictionary representation of the query parameters.

    """
    cursor_kwargs = {}

    # filter
    if getattr(params, "filter", False):
        cursor_kwargs["filter"] = self.transformer.transform(
            self.parser.parse(params.filter)  # type: ignore[union-attr]
        )
    else:
        cursor_kwargs["filter"] = {}

    # response_format
    if (
        getattr(params, "response_format", False)
        and params.response_format != "json"
    ):
        raise BadRequest(
            detail=f"Response format {params.response_format} is not supported, please use response_format='json'"
        )

    # page_limit
    if getattr(params, "page_limit", False):
        limit = params.page_limit  # type: ignore[union-attr]
        if limit > CONFIG.page_limit_max:
            raise Forbidden(
                detail=f"Max allowed page_limit is {CONFIG.page_limit_max}, you requested {limit}",
            )
        cursor_kwargs["limit"] = limit
    else:
        cursor_kwargs["limit"] = CONFIG.page_limit

    # response_fields
    cursor_kwargs["projection"] = {
        f"{self.resource_mapper.get_backend_field(f)}": True
        for f in self.all_fields
    }

    if getattr(params, "response_fields", False):
        response_fields = set(params.response_fields.split(","))
        response_fields |= self.resource_mapper.get_required_fields()
    else:
        response_fields = self.all_fields.copy()

    cursor_kwargs["fields"] = response_fields

    # sort
    if getattr(params, "sort", False):
        cursor_kwargs["sort"] = self.parse_sort_params(params.sort)  # type: ignore[union-attr]

    # warn if multiple pagination keys are present, and only use the first from this list
    received_pagination_option = False
    warn_multiple_keys = False

    if getattr(params, "page_offset", False):
        received_pagination_option = True
        cursor_kwargs["skip"] = params.page_offset  # type: ignore[union-attr]

    if isinstance(getattr(params, "page_number", None), int):
        if received_pagination_option:
            warn_multiple_keys = True
        else:
            received_pagination_option = True
            if params.page_number < 1:  # type: ignore[union-attr]
                warnings.warn(
                    message=f"'page_number' is 1-based, using 'page_number=1' instead of {params.page_number}",  # type: ignore[union-attr]
                    category=QueryParamNotUsed,
                )
                page_number = 1
            else:
                page_number = params.page_number  # type: ignore[union-attr]
            cursor_kwargs["skip"] = (page_number - 1) * cursor_kwargs["limit"]

    if isinstance(getattr(params, "page_above", None), str):
        if received_pagination_option:
            warn_multiple_keys = True
        else:
            received_pagination_option = True
            cursor_kwargs["page_above"] = params.page_above  # type: ignore[union-attr]

    if warn_multiple_keys:
        warnings.warn(
            message="Multiple pagination keys were provided, only using the first one of 'page_offset', 'page_number' or 'page_above'",
            category=QueryParamNotUsed,
        )

    return cursor_kwargs

insert(data) abstractmethod

Add the given entries to the underlying database.

Parameters:

Name	Type	Description	Default
data	list[EntryResource]	The entry resource objects to add to the database.	required

Source code in optimade/server/entry_collections/entry_collections.py

@abstractmethod
def insert(self, data: list[EntryResource]) -> None:
    """Add the given entries to the underlying database.

    Arguments:
        data: The entry resource objects to add to the database.

    """

parse_sort_params(sort_params)

Handles any sort parameters passed to the collection, resolving aliases and dealing with any invalid fields.

Raises:

Type	Description
BadRequest	if an invalid sort is requested.

Returns:

Type	Description
Iterable[tuple[str, int]]	A list of tuples containing the aliased field name and
Iterable[tuple[str, int]]	sort direction encoded as 1 (ascending) or -1 (descending).

Source code in optimade/server/entry_collections/entry_collections.py

def parse_sort_params(self, sort_params: str) -> Iterable[tuple[str, int]]:
    """Handles any sort parameters passed to the collection,
    resolving aliases and dealing with any invalid fields.

    Raises:
        BadRequest: if an invalid sort is requested.

    Returns:
        A list of tuples containing the aliased field name and
        sort direction encoded as 1 (ascending) or -1 (descending).

    """
    sort_spec: list[tuple[str, int]] = []
    for field in sort_params.split(","):
        sort_dir = 1
        if field.startswith("-"):
            field = field[1:]
            sort_dir = -1
        aliased_field = self.resource_mapper.get_backend_field(field)
        sort_spec.append((aliased_field, sort_dir))

    unknown_fields = [
        field
        for field, _ in sort_spec
        if self.resource_mapper.get_optimade_field(field) not in self.all_fields
    ]

    if unknown_fields:
        error_detail = "Unable to sort on unknown field{} '{}'".format(
            "s" if len(unknown_fields) > 1 else "",
            "', '".join(unknown_fields),
        )

        # If all unknown fields are "other" provider-specific, then only provide a warning
        if all(
            (
                re.match(r"_[a-z_0-9]+_[a-z_0-9]*", field)
                and not field.startswith(f"_{self.provider_prefix}_")
            )
            for field in unknown_fields
        ):
            warnings.warn(error_detail, FieldValueNotRecognized)

        # Otherwise, if all fields are unknown, or some fields are unknown and do not
        # have other provider prefixes, then return 400: Bad Request
        else:
            raise BadRequest(detail=error_detail)

    # If at least one valid field has been provided for sorting, then use that
    sort_spec = [
        (field, sort_dir)
        for field, sort_dir in sort_spec
        if field not in unknown_fields
    ]

    return sort_spec

EntryListingQueryParams

Bases: BaseQueryParams

Common query params for all Entry listing endpoints.

Attributes:

Name	Type	Description
filter	str	A filter string, in the format described in section API Filtering Format Specification of the specification.
response_format	str	The output format requested (see section Response Format). Defaults to the format string 'json', which specifies the standard output format described in this specification. Example: http://example.com/v1/structures?response_format=xml
email_address	EmailStr	An email address of the user making the request. The email SHOULD be that of a person and not an automatic system. Example: http://example.com/v1/structures?email_address=user@example.com
response_fields	str	A comma-delimited set of fields to be provided in the output. If provided, these fields MUST be returned along with the REQUIRED fields. Other OPTIONAL fields MUST NOT be returned when this parameter is present. Example: http://example.com/v1/structures?response_fields=last_modified,nsites
sort	str	If supporting sortable queries, an implementation MUST use the sort query parameter with format as specified by JSON API 1.0. An implementation MAY support multiple sort fields for a single query. If it does, it again MUST conform to the JSON API 1.0 specification. If an implementation supports sorting for an entry listing endpoint, then the /info/<entries> endpoint MUST include, for each field name <fieldname> in its data.properties.<fieldname> response value that can be used for sorting, the key sortable with value true. If a field name under an entry listing endpoint supporting sorting cannot be used for sorting, the server MUST either leave out the sortable key or set it equal to false for the specific field name. The set of field names, with sortable equal to true are allowed to be used in the "sort fields" list according to its definition in the JSON API 1.0 specification. The field sortable is in addition to each property description and other OPTIONAL fields. An example is shown in the section Entry Listing Info Endpoints.
page_limit	int	Sets a numerical limit on the number of entries returned. See JSON API 1.0. The API implementation MUST return no more than the number specified. It MAY return fewer. The database MAY have a maximum limit and not accept larger numbers (in which case an error code -- 403 Forbidden -- MUST be returned). The default limit value is up to the API implementation to decide. Example: http://example.com/optimade/v1/structures?page_limit=100
page_offset	int	RECOMMENDED for use with offset-based pagination: using page_offset and page_limit is RECOMMENDED. Example: Skip 50 structures and fetch up to 100: /structures?page_offset=50&page_limit=100.
page_number	int	RECOMMENDED for use with page-based pagination: using page_number and page_limit is RECOMMENDED. It is RECOMMENDED that the first page has number 1, i.e., that page_number is 1-based. Example: Fetch page 2 of up to 50 structures per page: /structures?page_number=2&page_limit=50.
page_cursor	int	RECOMMENDED for use with cursor-based pagination: using page_cursor and page_limit is RECOMMENDED.
page_above	str	RECOMMENDED for use with value-based pagination: using page_above/page_below and page_limit is RECOMMENDED. Example: Fetch up to 100 structures above sort-field value 4000 (in this example, server chooses to fetch results sorted by increasing id, so page_above value refers to an id value): /structures?page_above=4000&page_limit=100.
page_below	str	RECOMMENDED for use with value-based pagination: using page_above/page_below and page_limit is RECOMMENDED.
include	str	A server MAY implement the JSON API concept of returning compound documents by utilizing the include query parameter as specified by JSON API 1.0. All related resource objects MUST be returned as part of an array value for the top-level included field, see the section JSON Response Schema: Common Fields. The value of include MUST be a comma-separated list of "relationship paths", as defined in the JSON API. If relationship paths are not supported, or a server is unable to identify a relationship path a 400 Bad Request response MUST be made. The default value for include is references. This means references entries MUST always be included under the top-level field included as default, since a server assumes if include is not specified by a client in the request, it is still specified as include=references. Note, if a client explicitly specifies include and leaves out references, references resource objects MUST NOT be included under the top-level field included, as per the definition of included, see section JSON Response Schema: Common Fields. Note: A query with the parameter include set to the empty string means no related resource objects are to be returned under the top-level field included.
api_hint	str	If the client provides the parameter, the value SHOULD have the format vMAJOR or vMAJOR.MINOR, where MAJOR is a major version and MINOR is a minor version of the API. For example, if a client appends api_hint=v1.0 to the query string, the hint provided is for major version 1 and minor version 0.

Source code in optimade/server/query_params.py

class EntryListingQueryParams(BaseQueryParams):
    """
    Common query params for all Entry listing endpoints.

    Attributes:
        filter (str): A filter string, in the format described in section API Filtering Format Specification of the specification.

        response_format (str): The output format requested (see section Response Format).
            Defaults to the format string 'json', which specifies the standard output format described in this specification.

            **Example**: `http://example.com/v1/structures?response_format=xml`

        email_address (EmailStr): An email address of the user making the request.
            The email SHOULD be that of a person and not an automatic system.

            **Example**: `http://example.com/v1/structures?email_address=user@example.com`

        response_fields (str): A comma-delimited set of fields to be provided in the output.
            If provided, these fields MUST be returned along with the REQUIRED fields.
            Other OPTIONAL fields MUST NOT be returned when this parameter is present.

            **Example**: `http://example.com/v1/structures?response_fields=last_modified,nsites`

        sort (str): If supporting sortable queries, an implementation MUST use the `sort` query parameter with format as specified
            by [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-sorting).

            An implementation MAY support multiple sort fields for a single query.
            If it does, it again MUST conform to the JSON API 1.0 specification.

            If an implementation supports sorting for an entry listing endpoint, then the `/info/<entries>` endpoint MUST include,
            for each field name `<fieldname>` in its `data.properties.<fieldname>` response value that can be used for sorting,
            the key `sortable` with value `true`.
            If a field name under an entry listing endpoint supporting sorting cannot be used for sorting, the server MUST either
            leave out the `sortable` key or set it equal to `false` for the specific field name.
            The set of field names, with `sortable` equal to `true` are allowed to be used in the "sort fields" list according to
            its definition in the JSON API 1.0 specification.
            The field `sortable` is in addition to each property description and other OPTIONAL fields.
            An example is shown in the section Entry Listing Info Endpoints.

        page_limit (int): Sets a numerical limit on the number of entries returned.
            See [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-pagination).
            The API implementation MUST return no more than the number specified.
            It MAY return fewer. The database MAY have a maximum limit and not accept larger numbers
            (in which case an error code -- `403 Forbidden` -- MUST be returned).
            The default limit value is up to the API implementation to decide.

            **Example**: `http://example.com/optimade/v1/structures?page_limit=100`

        page_offset (int): RECOMMENDED for use with _offset-based_ pagination: using `page_offset` and `page_limit` is RECOMMENDED.

            **Example**: Skip 50 structures and fetch up to 100: `/structures?page_offset=50&page_limit=100`.

        page_number (int): RECOMMENDED for use with _page-based_ pagination: using `page_number` and `page_limit` is RECOMMENDED.
            It is RECOMMENDED that the first page has number 1, i.e., that `page_number` is 1-based.

            **Example**: Fetch page 2 of up to 50 structures per page: `/structures?page_number=2&page_limit=50`.

        page_cursor (int): RECOMMENDED for use with _cursor-based_ pagination: using `page_cursor` and `page_limit` is RECOMMENDED.

        page_above (str): RECOMMENDED for use with _value-based_ pagination: using `page_above`/`page_below` and `page_limit` is RECOMMENDED.

            **Example**: Fetch up to 100 structures above sort-field value 4000 (in this example, server chooses to fetch results sorted by
            increasing `id`, so `page_above` value refers to an `id` value): `/structures?page_above=4000&page_limit=100`.

        page_below (str): RECOMMENDED for use with _value-based_ pagination: using `page_above`/`page_below` and `page_limit` is RECOMMENDED.

        include (str): A server MAY implement the JSON API concept of returning [compound documents](https://jsonapi.org/format/1.0/#document-compound-documents)
            by utilizing the `include` query parameter as specified by [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-includes).

            All related resource objects MUST be returned as part of an array value for the top-level `included` field,
            see the section JSON Response Schema: Common Fields.

            The value of `include` MUST be a comma-separated list of "relationship paths", as defined in the [JSON API](https://jsonapi.org/format/1.0/#fetching-includes).
            If relationship paths are not supported, or a server is unable to identify a relationship path a `400 Bad Request` response MUST be made.

            The **default value** for `include` is `references`. This means `references` entries MUST always be included under the top-level field
            `included` as default, since a server assumes if `include` is not specified by a client in the request, it is still specified as `include=references`.
            Note, if a client explicitly specifies `include` and leaves out `references`, `references` resource objects MUST NOT be included under the top-level
            field `included`, as per the definition of `included`, see section JSON Response Schema: Common Fields.

            **Note**: A query with the parameter `include` set to the empty string means no related resource objects are to be returned under the top-level field `included`.

        api_hint (str): If the client provides the parameter, the value SHOULD have the format `vMAJOR` or `vMAJOR.MINOR`,
            where MAJOR is a major version and MINOR is a minor version of the API.
            For example, if a client appends `api_hint=v1.0` to the query string, the hint provided is for major version 1 and minor version 0.

    """

    # The reference server implementation only supports offset/number-based pagination
    unsupported_params: list[str] = [
        "page_cursor",
        "page_below",
    ]

    def __init__(
        self,
        *,
        filter: Annotated[
            str,
            Query(  # pylint: disable=redefined-builtin
                description="A filter string, in the format described in section API Filtering Format Specification of the specification.",
            ),
        ] = "",
        response_format: Annotated[
            str,
            Query(
                description="The output format requested (see section Response Format).\nDefaults to the format string 'json', which specifies the standard output format described in this specification.\nExample: `http://example.com/v1/structures?response_format=xml`",
            ),
        ] = "json",
        email_address: Annotated[
            EmailStr,
            Query(
                description="An email address of the user making the request.\nThe email SHOULD be that of a person and not an automatic system.\nExample: `http://example.com/v1/structures?email_address=user@example.com`",
            ),
        ] = "",
        response_fields: Annotated[
            str,
            Query(
                description="A comma-delimited set of fields to be provided in the output.\nIf provided, these fields MUST be returned along with the REQUIRED fields.\nOther OPTIONAL fields MUST NOT be returned when this parameter is present.\nExample: `http://example.com/v1/structures?response_fields=last_modified,nsites`",
                pattern=r"([a-z_][a-z_0-9]*(,[a-z_][a-z_0-9]*)*)?",
            ),
        ] = "",
        sort: Annotated[
            str,
            Query(
                description='If supporting sortable queries, an implementation MUST use the `sort` query parameter with format as specified by [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-sorting).\n\nAn implementation MAY support multiple sort fields for a single query.\nIf it does, it again MUST conform to the JSON API 1.0 specification.\n\nIf an implementation supports sorting for an entry listing endpoint, then the `/info/<entries>` endpoint MUST include, for each field name `<fieldname>` in its `data.properties.<fieldname>` response value that can be used for sorting, the key `sortable` with value `true`.\nIf a field name under an entry listing endpoint supporting sorting cannot be used for sorting, the server MUST either leave out the `sortable` key or set it equal to `false` for the specific field name.\nThe set of field names, with `sortable` equal to `true` are allowed to be used in the "sort fields" list according to its definition in the JSON API 1.0 specification.\nThe field `sortable` is in addition to each property description and other OPTIONAL fields.\nAn example is shown in the section Entry Listing Info Endpoints.',
                pattern=r"([a-z_][a-z_0-9]*(,[a-z_][a-z_0-9]*)*)?",
            ),
        ] = "",
        page_limit: Annotated[
            int,
            Query(
                description="Sets a numerical limit on the number of entries returned.\nSee [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-pagination).\nThe API implementation MUST return no more than the number specified.\nIt MAY return fewer.\nThe database MAY have a maximum limit and not accept larger numbers (in which case an error code -- 403 Forbidden -- MUST be returned).\nThe default limit value is up to the API implementation to decide.\nExample: `http://example.com/optimade/v1/structures?page_limit=100`",
                ge=0,
            ),
        ] = CONFIG.page_limit,
        page_offset: Annotated[
            int,
            Query(
                description="RECOMMENDED for use with _offset-based_ pagination: using `page_offset` and `page_limit` is RECOMMENDED.\nExample: Skip 50 structures and fetch up to 100: `/structures?page_offset=50&page_limit=100`.",
                ge=0,
            ),
        ] = 0,
        page_number: Annotated[
            int,
            Query(
                description="RECOMMENDED for use with _page-based_ pagination: using `page_number` and `page_limit` is RECOMMENDED.\nIt is RECOMMENDED that the first page has number 1, i.e., that `page_number` is 1-based.\nExample: Fetch page 2 of up to 50 structures per page: `/structures?page_number=2&page_limit=50`.",
                # ge=1,  # This constraint is only 'RECOMMENDED' in the specification, so should not be included here or in the OpenAPI schema
            ),
        ] = None,  # type: ignore[assignment]
        page_cursor: Annotated[
            int,
            Query(
                description="RECOMMENDED for use with _cursor-based_ pagination: using `page_cursor` and `page_limit` is RECOMMENDED.",
                ge=0,
            ),
        ] = 0,
        page_above: Annotated[
            str,
            Query(
                description="RECOMMENDED for use with _value-based_ pagination: using `page_above`/`page_below` and `page_limit` is RECOMMENDED.\nExample: Fetch up to 100 structures above sort-field value 4000 (in this example, server chooses to fetch results sorted by increasing `id`, so `page_above` value refers to an `id` value): `/structures?page_above=4000&page_limit=100`.",
            ),
        ] = None,  # type: ignore[assignment]
        page_below: Annotated[
            str,
            Query(
                description="RECOMMENDED for use with _value-based_ pagination: using `page_above`/`page_below` and `page_limit` is RECOMMENDED.",
            ),
        ] = None,  # type: ignore[assignment]
        include: Annotated[
            str,
            Query(
                description='A server MAY implement the JSON API concept of returning [compound documents](https://jsonapi.org/format/1.0/#document-compound-documents) by utilizing the `include` query parameter as specified by [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-includes).\n\nAll related resource objects MUST be returned as part of an array value for the top-level `included` field, see the section JSON Response Schema: Common Fields.\n\nThe value of `include` MUST be a comma-separated list of "relationship paths", as defined in the [JSON API](https://jsonapi.org/format/1.0/#fetching-includes).\nIf relationship paths are not supported, or a server is unable to identify a relationship path a `400 Bad Request` response MUST be made.\n\nThe **default value** for `include` is `references`.\nThis means `references` entries MUST always be included under the top-level field `included` as default, since a server assumes if `include` is not specified by a client in the request, it is still specified as `include=references`.\nNote, if a client explicitly specifies `include` and leaves out `references`, `references` resource objects MUST NOT be included under the top-level field `included`, as per the definition of `included`, see section JSON Response Schema: Common Fields.\n\n> **Note**: A query with the parameter `include` set to the empty string means no related resource objects are to be returned under the top-level field `included`.',
            ),
        ] = "references",
        api_hint: Annotated[
            str,
            Query(
                description="If the client provides the parameter, the value SHOULD have the format `vMAJOR` or `vMAJOR.MINOR`, where MAJOR is a major version and MINOR is a minor version of the API. For example, if a client appends `api_hint=v1.0` to the query string, the hint provided is for major version 1 and minor version 0.",
                pattern=r"(v[0-9]+(\.[0-9]+)?)?",
            ),
        ] = "",
    ):
        self.filter = filter
        self.response_format = response_format
        self.email_address = email_address
        self.response_fields = response_fields
        self.sort = sort
        self.page_limit = page_limit
        self.page_offset = page_offset
        self.page_number = page_number
        self.page_cursor = page_cursor
        self.page_above = page_above
        self.page_below = page_below
        self.include = include
        self.api_hint = api_hint

check_params(query_params)

This method checks whether all the query parameters that are specified in the URL string are implemented in the relevant *QueryParams class.

This method handles four cases:

• If a query parameter is passed that is not defined in the relevant *QueryParams class, and it is not prefixed with a known provider prefix, then a BadRequest is raised.
• If a query parameter is passed that is not defined in the relevant *QueryParams class, that is prefixed with a known provider prefix, then the parameter is silently ignored
• If a query parameter is passed that is not defined in the relevant *QueryParams class, that is prefixed with an unknown provider prefix, then a UnknownProviderQueryParameter warning is emitted.
• If a query parameter is passed that is on the unsupported_params list for the inherited class, then a QueryParamNotUsed warning is emitted.

Parameters:

Name	Type	Description	Default
query_params	Iterable[str]	An iterable of the request's string query parameters.	required

Raises:

Type	Description
`BadRequest`	if the query parameter was not found in the relevant class, or if it does not have a valid prefix.

Source code in optimade/server/query_params.py

def check_params(self, query_params: Iterable[str]) -> None:
    """This method checks whether all the query parameters that are specified
    in the URL string are implemented in the relevant `*QueryParams` class.

    This method handles four cases:

    * If a query parameter is passed that is not defined in the relevant `*QueryParams` class,
      and it is not prefixed with a known provider prefix, then a `BadRequest` is raised.
    * If a query parameter is passed that is not defined in the relevant `*QueryParams` class,
      that is prefixed with a known provider prefix, then the parameter is silently ignored
    * If a query parameter is passed that is not defined in the relevant `*QueryParams` class,
      that is prefixed with an unknown provider prefix, then a `UnknownProviderQueryParameter`
      warning is emitted.
    * If a query parameter is passed that is on the `unsupported_params` list for the inherited
      class, then a `QueryParamNotUsed` warning is emitted.

    Arguments:
        query_params: An iterable of the request's string query parameters.

    Raises:
        `BadRequest`: if the query parameter was not found in the relevant class, or if it
            does not have a valid prefix.

    """
    if not getattr(CONFIG, "validate_query_parameters", False):
        return
    errors = []
    warnings = []
    unsupported_warnings = []
    for param in query_params:
        if param in self.unsupported_params:
            unsupported_warnings.append(param)
        if not hasattr(self, param):
            split_param = param.split("_")
            if param.startswith("_") and len(split_param) > 2:
                prefix = split_param[1]
                if prefix in BaseResourceMapper.SUPPORTED_PREFIXES:
                    errors.append(param)
                elif prefix not in BaseResourceMapper.KNOWN_PROVIDER_PREFIXES:
                    warnings.append(param)
            else:
                errors.append(param)

    if warnings:
        warn(
            f"The query parameter(s) '{warnings}' are unrecognised and have been ignored.",
            UnknownProviderQueryParameter,
        )

    if unsupported_warnings:
        warn(
            f"The query parameter(s) '{unsupported_warnings}' are not supported by this server and have been ignored.",
            QueryParamNotUsed,
        )

    if errors:
        raise BadRequest(
            f"The query parameter(s) '{errors}' are not recognised by this endpoint."
        )

EntryResource

Bases: Resource

The base model for an entry resource.

Source code in optimade/models/entries.py

class EntryResource(Resource):
    """The base model for an entry resource."""

    id: Annotated[
        str,
        OptimadeField(
            description="""An entry's ID as defined in section Definition of Terms.

- **Type**: string.

- **Requirements/Conventions**:
    - **Support**: MUST be supported by all implementations, MUST NOT be `null`.
    - **Query**: MUST be a queryable property with support for all mandatory filter features.
    - **Response**: REQUIRED in the response.

- **Examples**:
    - `"db/1234567"`
    - `"cod/2000000"`
    - `"cod/2000000@1234567"`
    - `"nomad/L1234567890"`
    - `"42"`""",
            support=SupportLevel.MUST,
            queryable=SupportLevel.MUST,
        ),
    ]

    type: Annotated[
        str,
        OptimadeField(
            description="""The name of the type of an entry.

- **Type**: string.

- **Requirements/Conventions**:
    - **Support**: MUST be supported by all implementations, MUST NOT be `null`.
    - **Query**: MUST be a queryable property with support for all mandatory filter features.
    - **Response**: REQUIRED in the response.
    - MUST be an existing entry type.
    - The entry of type `<type>` and ID `<id>` MUST be returned in response to a request for `/<type>/<id>` under the versioned base URL.

- **Example**: `"structures"`""",
            support=SupportLevel.MUST,
            queryable=SupportLevel.MUST,
        ),
    ]

    attributes: Annotated[
        EntryResourceAttributes,
        StrictField(
            description="""A dictionary, containing key-value pairs representing the entry's properties, except for `type` and `id`.
Database-provider-specific properties need to include the database-provider-specific prefix (see section on Database-Provider-Specific Namespace Prefixes).""",
        ),
    ]

    relationships: Annotated[
        Optional[EntryRelationships],
        StrictField(
            description="""A dictionary containing references to other entries according to the description in section Relationships encoded as [JSON API Relationships](https://jsonapi.org/format/1.0/#document-resource-object-relationships).
The OPTIONAL human-readable description of the relationship MAY be provided in the `description` field inside the `meta` dictionary of the JSON API resource identifier object.""",
        ),
    ] = None

InternalServerError

Bases: OptimadeHTTPException

500 Internal Server Error

Source code in optimade/exceptions.py

class InternalServerError(OptimadeHTTPException):
    """500 Internal Server Error"""

    status_code: int = 500
    title: str = "Internal Server Error"

JSONAPIResponse

Bases: JSONResponse

This class simply patches fastapi.responses.JSONResponse to use the JSON:API 'application/vnd.api+json' MIME type.

Source code in optimade/server/routers/utils.py

class JSONAPIResponse(JSONResponse):
    """This class simply patches `fastapi.responses.JSONResponse` to use the
    JSON:API 'application/vnd.api+json' MIME type.

    """

    media_type = "application/vnd.api+json"

ResponseMeta

Bases: Meta

A JSON API meta member that contains JSON API meta objects of non-standard meta-information.

OPTIONAL additional information global to the query that is not specified in this document, MUST start with a database-provider-specific prefix.

Source code in optimade/models/optimade_json.py

class ResponseMeta(jsonapi.Meta):
    """
    A [JSON API meta member](https://jsonapi.org/format/1.0#document-meta)
    that contains JSON API meta objects of non-standard
    meta-information.

    OPTIONAL additional information global to the query that is not
    specified in this document, MUST start with a
    database-provider-specific prefix.
    """

    query: Annotated[
        ResponseMetaQuery,
        StrictField(description="Information on the Query that was requested"),
    ]

    api_version: Annotated[
        SemanticVersion,
        StrictField(
            description="""Presently used full version of the OPTIMADE API.
The version number string MUST NOT be prefixed by, e.g., "v".
Examples: `1.0.0`, `1.0.0-rc.2`.""",
        ),
    ]

    more_data_available: Annotated[
        bool,
        StrictField(
            description="`false` if the response contains all data for the request (e.g., a request issued to a single entry endpoint, or a `filter` query at the last page of a paginated response) and `true` if the response is incomplete in the sense that multiple objects match the request, and not all of them have been included in the response (e.g., a query with multiple pages that is not at the last page).",
        ),
    ]

    # start of "SHOULD" fields for meta response
    optimade_schema: Annotated[
        Optional[jsonapi.JsonLinkType],
        StrictField(
            alias="schema",
            description="""A [JSON API links object](http://jsonapi.org/format/1.0/#document-links) that points to a schema for the response.
If it is a string, or a dictionary containing no `meta` field, the provided URL MUST point at an [OpenAPI](https://swagger.io/specification/) schema.
It is possible that future versions of this specification allows for alternative schema types.
Hence, if the `meta` field of the JSON API links object is provided and contains a field `schema_type` that is not equal to the string `OpenAPI` the client MUST not handle failures to parse the schema or to validate the response against the schema as errors.""",
        ),
    ] = None

    time_stamp: Annotated[
        Optional[datetime],
        StrictField(
            description="A timestamp containing the date and time at which the query was executed.",
        ),
    ] = None

    data_returned: Annotated[
        Optional[int],
        StrictField(
            description="An integer containing the total number of data resource objects returned for the current `filter` query, independent of pagination.",
            ge=0,
        ),
    ] = None

    provider: Annotated[
        Optional[Provider],
        StrictField(
            description="information on the database provider of the implementation."
        ),
    ] = None

    # start of "MAY" fields for meta response
    data_available: Annotated[
        Optional[int],
        StrictField(
            description="An integer containing the total number of data resource objects available in the database for the endpoint.",
        ),
    ] = None

    last_id: Annotated[
        Optional[str],
        StrictField(description="a string containing the last ID returned"),
    ] = None

    response_message: Annotated[
        Optional[str], StrictField(description="response string from the server")
    ] = None

    implementation: Annotated[
        Optional[Implementation],
        StrictField(description="a dictionary describing the server implementation"),
    ] = None

    warnings: Annotated[
        Optional[list[Warnings]],
        StrictField(
            description="""A list of warning resource objects representing non-critical errors or warnings.
A warning resource object is defined similarly to a [JSON API error object](http://jsonapi.org/format/1.0/#error-objects), but MUST also include the field `type`, which MUST have the value `"warning"`.
The field `detail` MUST be present and SHOULD contain a non-critical message, e.g., reporting unrecognized search attributes or deprecated features.
The field `status`, representing a HTTP response status code, MUST NOT be present for a warning resource object.
This is an exclusive field for error resource objects.""",
            uniqueItems=True,
        ),
    ] = None

SingleEntryQueryParams

Bases: BaseQueryParams

Common query params for single entry endpoints.

Attributes:

Name	Type	Description
response_format	str	The output format requested (see section Response Format). Defaults to the format string 'json', which specifies the standard output format described in this specification. Example: http://example.com/v1/structures?response_format=xml
email_address	EmailStr	An email address of the user making the request. The email SHOULD be that of a person and not an automatic system. Example: http://example.com/v1/structures?email_address=user@example.com
response_fields	str	A comma-delimited set of fields to be provided in the output. If provided, these fields MUST be returned along with the REQUIRED fields. Other OPTIONAL fields MUST NOT be returned when this parameter is present. Example: http://example.com/v1/structures?response_fields=last_modified,nsites
include	str	A server MAY implement the JSON API concept of returning compound documents by utilizing the include query parameter as specified by JSON API 1.0. All related resource objects MUST be returned as part of an array value for the top-level included field, see the section JSON Response Schema: Common Fields. The value of include MUST be a comma-separated list of "relationship paths", as defined in the JSON API. If relationship paths are not supported, or a server is unable to identify a relationship path a 400 Bad Request response MUST be made. The default value for include is references. This means references entries MUST always be included under the top-level field included as default, since a server assumes if include is not specified by a client in the request, it is still specified as include=references. Note, if a client explicitly specifies include and leaves out references, references resource objects MUST NOT be included under the top-level field included, as per the definition of included, see section JSON Response Schema: Common Fields. Note: A query with the parameter include set to the empty string means no related resource objects are to be returned under the top-level field included.
api_hint	str	If the client provides the parameter, the value SHOULD have the format vMAJOR or vMAJOR.MINOR, where MAJOR is a major version and MINOR is a minor version of the API. For example, if a client appends api_hint=v1.0 to the query string, the hint provided is for major version 1 and minor version 0.

Source code in optimade/server/query_params.py

class SingleEntryQueryParams(BaseQueryParams):
    """
    Common query params for single entry endpoints.

    Attributes:
        response_format (str): The output format requested (see section Response Format).
            Defaults to the format string 'json', which specifies the standard output format described in this specification.

            **Example**: `http://example.com/v1/structures?response_format=xml`

        email_address (EmailStr): An email address of the user making the request.
            The email SHOULD be that of a person and not an automatic system.

            **Example**: `http://example.com/v1/structures?email_address=user@example.com`

        response_fields (str): A comma-delimited set of fields to be provided in the output.
            If provided, these fields MUST be returned along with the REQUIRED fields.
            Other OPTIONAL fields MUST NOT be returned when this parameter is present.

            **Example**: `http://example.com/v1/structures?response_fields=last_modified,nsites`

        include (str): A server MAY implement the JSON API concept of returning [compound documents](https://jsonapi.org/format/1.0/#document-compound-documents)
            by utilizing the `include` query parameter as specified by [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-includes).

            All related resource objects MUST be returned as part of an array value for the top-level `included` field,
            see the section JSON Response Schema: Common Fields.

            The value of `include` MUST be a comma-separated list of "relationship paths", as defined in the [JSON API](https://jsonapi.org/format/1.0/#fetching-includes).
            If relationship paths are not supported, or a server is unable to identify a relationship path a `400 Bad Request` response MUST be made.

            The **default value** for `include` is `references`. This means `references` entries MUST always be included under the top-level field
            `included` as default, since a server assumes if `include` is not specified by a client in the request, it is still specified as `include=references`.
            Note, if a client explicitly specifies `include` and leaves out `references`, `references` resource objects MUST NOT be included under the top-level
            field `included`, as per the definition of `included`, see section JSON Response Schema: Common Fields.

            **Note**: A query with the parameter `include` set to the empty string means no related resource objects are to be returned under the top-level field `included`.

        api_hint (str): If the client provides the parameter, the value SHOULD have the format `vMAJOR` or `vMAJOR.MINOR`,
            where MAJOR is a major version and MINOR is a minor version of the API.
            For example, if a client appends `api_hint=v1.0` to the query string, the hint provided is for major version 1 and minor version 0.

    """

    def __init__(
        self,
        *,
        response_format: Annotated[
            str,
            Query(
                description="The output format requested (see section Response Format).\nDefaults to the format string 'json', which specifies the standard output format described in this specification.\nExample: `http://example.com/v1/structures?response_format=xml`",
            ),
        ] = "json",
        email_address: Annotated[
            EmailStr,
            Query(
                description="An email address of the user making the request.\nThe email SHOULD be that of a person and not an automatic system.\nExample: `http://example.com/v1/structures?email_address=user@example.com`",
            ),
        ] = "",
        response_fields: Annotated[
            str,
            Query(
                description="A comma-delimited set of fields to be provided in the output.\nIf provided, these fields MUST be returned along with the REQUIRED fields.\nOther OPTIONAL fields MUST NOT be returned when this parameter is present.\nExample: `http://example.com/v1/structures?response_fields=last_modified,nsites`",
                pattern=r"([a-z_][a-z_0-9]*(,[a-z_][a-z_0-9]*)*)?",
            ),
        ] = "",
        include: Annotated[
            str,
            Query(
                description='A server MAY implement the JSON API concept of returning [compound documents](https://jsonapi.org/format/1.0/#document-compound-documents) by utilizing the `include` query parameter as specified by [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-includes).\n\nAll related resource objects MUST be returned as part of an array value for the top-level `included` field, see the section JSON Response Schema: Common Fields.\n\nThe value of `include` MUST be a comma-separated list of "relationship paths", as defined in the [JSON API](https://jsonapi.org/format/1.0/#fetching-includes).\nIf relationship paths are not supported, or a server is unable to identify a relationship path a `400 Bad Request` response MUST be made.\n\nThe **default value** for `include` is `references`.\nThis means `references` entries MUST always be included under the top-level field `included` as default, since a server assumes if `include` is not specified by a client in the request, it is still specified as `include=references`.\nNote, if a client explicitly specifies `include` and leaves out `references`, `references` resource objects MUST NOT be included under the top-level field `included`, as per the definition of `included`, see section JSON Response Schema: Common Fields.\n\n> **Note**: A query with the parameter `include` set to the empty string means no related resource objects are to be returned under the top-level field `included`.',
            ),
        ] = "references",
        api_hint: Annotated[
            str,
            Query(
                description="If the client provides the parameter, the value SHOULD have the format `vMAJOR` or `vMAJOR.MINOR`, where MAJOR is a major version and MINOR is a minor version of the API. For example, if a client appends `api_hint=v1.0` to the query string, the hint provided is for major version 1 and minor version 0.",
                pattern=r"(v[0-9]+(\.[0-9]+)?)?",
            ),
        ] = "",
    ):
        self.response_format = response_format
        self.email_address = email_address
        self.response_fields = response_fields
        self.include = include
        self.api_hint = api_hint

check_params(query_params)

This method checks whether all the query parameters that are specified in the URL string are implemented in the relevant *QueryParams class.

This method handles four cases:

• If a query parameter is passed that is not defined in the relevant *QueryParams class, and it is not prefixed with a known provider prefix, then a BadRequest is raised.
• If a query parameter is passed that is not defined in the relevant *QueryParams class, that is prefixed with a known provider prefix, then the parameter is silently ignored
• If a query parameter is passed that is not defined in the relevant *QueryParams class, that is prefixed with an unknown provider prefix, then a UnknownProviderQueryParameter warning is emitted.
• If a query parameter is passed that is on the unsupported_params list for the inherited class, then a QueryParamNotUsed warning is emitted.

Parameters:

Name	Type	Description	Default
query_params	Iterable[str]	An iterable of the request's string query parameters.	required

Raises:

Type	Description
`BadRequest`	if the query parameter was not found in the relevant class, or if it does not have a valid prefix.

Source code in optimade/server/query_params.py

def check_params(self, query_params: Iterable[str]) -> None:
    """This method checks whether all the query parameters that are specified
    in the URL string are implemented in the relevant `*QueryParams` class.

    This method handles four cases:

    * If a query parameter is passed that is not defined in the relevant `*QueryParams` class,
      and it is not prefixed with a known provider prefix, then a `BadRequest` is raised.
    * If a query parameter is passed that is not defined in the relevant `*QueryParams` class,
      that is prefixed with a known provider prefix, then the parameter is silently ignored
    * If a query parameter is passed that is not defined in the relevant `*QueryParams` class,
      that is prefixed with an unknown provider prefix, then a `UnknownProviderQueryParameter`
      warning is emitted.
    * If a query parameter is passed that is on the `unsupported_params` list for the inherited
      class, then a `QueryParamNotUsed` warning is emitted.

    Arguments:
        query_params: An iterable of the request's string query parameters.

    Raises:
        `BadRequest`: if the query parameter was not found in the relevant class, or if it
            does not have a valid prefix.

    """
    if not getattr(CONFIG, "validate_query_parameters", False):
        return
    errors = []
    warnings = []
    unsupported_warnings = []
    for param in query_params:
        if param in self.unsupported_params:
            unsupported_warnings.append(param)
        if not hasattr(self, param):
            split_param = param.split("_")
            if param.startswith("_") and len(split_param) > 2:
                prefix = split_param[1]
                if prefix in BaseResourceMapper.SUPPORTED_PREFIXES:
                    errors.append(param)
                elif prefix not in BaseResourceMapper.KNOWN_PROVIDER_PREFIXES:
                    warnings.append(param)
            else:
                errors.append(param)

    if warnings:
        warn(
            f"The query parameter(s) '{warnings}' are unrecognised and have been ignored.",
            UnknownProviderQueryParameter,
        )

    if unsupported_warnings:
        warn(
            f"The query parameter(s) '{unsupported_warnings}' are not supported by this server and have been ignored.",
            QueryParamNotUsed,
        )

    if errors:
        raise BadRequest(
            f"The query parameter(s) '{errors}' are not recognised by this endpoint."
        )

ToplevelLinks

Bases: BaseModel

A set of Links objects, possibly including pagination

Source code in optimade/models/jsonapi.py

class ToplevelLinks(BaseModel):
    """A set of Links objects, possibly including pagination"""

    model_config = ConfigDict(extra="allow")

    self: Annotated[
        Optional[JsonLinkType], StrictField(description="A link to itself")
    ] = None
    related: Annotated[
        Optional[JsonLinkType], StrictField(description="A related resource link")
    ] = None

    # Pagination
    first: Annotated[
        Optional[JsonLinkType], StrictField(description="The first page of data")
    ] = None
    last: Annotated[
        Optional[JsonLinkType], StrictField(description="The last page of data")
    ] = None
    prev: Annotated[
        Optional[JsonLinkType], StrictField(description="The previous page of data")
    ] = None
    next: Annotated[
        Optional[JsonLinkType], StrictField(description="The next page of data")
    ] = None

    @model_validator(mode="after")
    def check_additional_keys_are_links(self) -> "ToplevelLinks":
        """The `ToplevelLinks` class allows any additional keys, as long as
        they are also Links or Urls themselves.

        """
        for field, value in self:
            if field not in self.model_fields:
                setattr(
                    self,
                    field,
                    TypeAdapter(Optional[JsonLinkType]).validate_python(value),
                )

        return self

check_additional_keys_are_links()

The ToplevelLinks class allows any additional keys, as long as they are also Links or Urls themselves.

Source code in optimade/models/jsonapi.py

@model_validator(mode="after")
def check_additional_keys_are_links(self) -> "ToplevelLinks":
    """The `ToplevelLinks` class allows any additional keys, as long as
    they are also Links or Urls themselves.

    """
    for field, value in self:
        if field not in self.model_fields:
            setattr(
                self,
                field,
                TypeAdapter(Optional[JsonLinkType]).validate_python(value),
            )

    return self

get_base_url(parsed_url_request)

Get base URL for current server

Take the base URL from the config file, if it exists, otherwise use the request.

Source code in optimade/server/routers/utils.py

def get_base_url(
    parsed_url_request: Union[
        urllib.parse.ParseResult, urllib.parse.SplitResult, StarletteURL, str
    ]
) -> str:
    """Get base URL for current server

    Take the base URL from the config file, if it exists, otherwise use the request.
    """
    parsed_url_request = (
        urllib.parse.urlparse(parsed_url_request)
        if isinstance(parsed_url_request, str)
        else parsed_url_request
    )
    return (
        CONFIG.base_url.rstrip("/")
        if CONFIG.base_url
        else f"{parsed_url_request.scheme}://{parsed_url_request.netloc}"
    )

get_entries(collection, request, params)

Generalized /{entry} endpoint getter

Source code in optimade/server/routers/utils.py

def get_entries(
    collection: EntryCollection,
    request: Request,
    params: EntryListingQueryParams,
) -> dict[str, Any]:
    """Generalized /{entry} endpoint getter"""
    from optimade.server.routers import ENTRY_COLLECTIONS

    params.check_params(request.query_params)
    (
        results,
        data_returned,
        more_data_available,
        fields,
        include_fields,
    ) = collection.find(params)

    include = []
    if getattr(params, "include", False):
        include.extend(params.include.split(","))

    included = []
    if results is not None:
        included = get_included_relationships(results, ENTRY_COLLECTIONS, include)

    if more_data_available:
        # Deduce the `next` link from the current request
        query = urllib.parse.parse_qs(request.url.query)
        query.update(collection.get_next_query_params(params, results))

        urlencoded = urllib.parse.urlencode(query, doseq=True)
        base_url = get_base_url(request.url)

        links = ToplevelLinks(next=f"{base_url}{request.url.path}?{urlencoded}")
    else:
        links = ToplevelLinks(next=None)

    if results is not None and (fields or include_fields):
        results = handle_response_fields(results, fields, include_fields)  # type: ignore[assignment]

    return {
        "links": links,
        "data": results if results else [],
        "meta": meta_values(
            url=request.url,
            data_returned=data_returned,
            data_available=len(collection),
            more_data_available=more_data_available,
            schema=CONFIG.schema_url
            if not CONFIG.is_index
            else CONFIG.index_schema_url,
        ),
        "included": included,
    }

get_included_relationships(results, ENTRY_COLLECTIONS, include_param)

Filters the included relationships and makes the appropriate compound request to include them in the response.

Parameters:

Name	Type	Description	Default
results	Union[EntryResource, list[EntryResource], dict, list[dict]]	list of returned documents.	required
ENTRY_COLLECTIONS	dict[str, EntryCollection]	dictionary containing collections to query, with key based on endpoint type.	required
include_param	list[str]	list of queried related resources that should be included in included.	required

Returns:

Type	Description
list[Union[EntryResource, dict[str, Any]]]	Dictionary with the same keys as ENTRY_COLLECTIONS, each containing the list of resource objects for that entry type.

Source code in optimade/server/routers/utils.py

def get_included_relationships(
    results: Union[EntryResource, list[EntryResource], dict, list[dict]],
    ENTRY_COLLECTIONS: dict[str, EntryCollection],
    include_param: list[str],
) -> list[Union[EntryResource, dict[str, Any]]]:
    """Filters the included relationships and makes the appropriate compound request
    to include them in the response.

    Parameters:
        results: list of returned documents.
        ENTRY_COLLECTIONS: dictionary containing collections to query, with key
            based on endpoint type.
        include_param: list of queried related resources that should be included in
            `included`.

    Returns:
        Dictionary with the same keys as ENTRY_COLLECTIONS, each containing the list
            of resource objects for that entry type.

    """
    from collections import defaultdict

    if not isinstance(results, list):
        results = [results]

    for entry_type in include_param:
        if entry_type not in ENTRY_COLLECTIONS and entry_type != "":
            raise BadRequest(
                detail=f"'{entry_type}' cannot be identified as a valid relationship type. "
                f"Known relationship types: {sorted(ENTRY_COLLECTIONS.keys())}"
            )

    endpoint_includes: dict[Any, dict] = defaultdict(dict)
    for doc in results:
        # convert list of references into dict by ID to only included unique IDs
        if doc is None:
            continue

        try:
            relationships = doc.relationships  # type: ignore
        except AttributeError:
            relationships = doc.get("relationships", None)

        if relationships is None:
            continue

        if not isinstance(relationships, dict):
            relationships = relationships.model_dump()

        for entry_type in ENTRY_COLLECTIONS:
            # Skip entry type if it is not in `include_param`
            if entry_type not in include_param:
                continue

            entry_relationship = relationships.get(entry_type, {})
            if entry_relationship is not None:
                refs = entry_relationship.get("data", [])
                for ref in refs:
                    if ref["id"] not in endpoint_includes[entry_type]:
                        endpoint_includes[entry_type][ref["id"]] = ref

    included: dict[
        str,
        Union[list[EntryResource], list[dict[str, Any]]],
    ] = {}
    for entry_type in endpoint_includes:
        compound_filter = " OR ".join(
            [f'id="{ref_id}"' for ref_id in endpoint_includes[entry_type]]
        )
        params = EntryListingQueryParams(
            filter=compound_filter,
            response_format="json",
            response_fields="",
            sort="",
            page_limit=0,
            page_offset=0,
        )

        # still need to handle pagination
        ref_results, _, _, _, _ = ENTRY_COLLECTIONS[entry_type].find(params)
        if ref_results is None:
            ref_results = []
        included[entry_type] = ref_results  # type: ignore[assignment]

    # flatten dict by endpoint to list
    return [obj for endp in included.values() for obj in endp]

get_providers(add_mongo_id=False)

Retrieve Materials-Consortia providers (from https://providers.optimade.org/v1/links).

Fallback order if providers.optimade.org is not available:

1. Try Materials-Consortia/providers on GitHub.
2. Try submodule providers' list of providers.
3. Log warning that providers list from Materials-Consortia is not included in the /links-endpoint.

Parameters:

Name	Type	Description	Default
add_mongo_id	bool	Whether to populate the _id field of the provider with MongoDB ObjectID.	False

Returns:

Type	Description
list	List of raw JSON-decoded providers including MongoDB object IDs.

Source code in optimade/utils.py

def get_providers(add_mongo_id: bool = False) -> list:
    """Retrieve Materials-Consortia providers (from https://providers.optimade.org/v1/links).

    Fallback order if providers.optimade.org is not available:

    1. Try Materials-Consortia/providers on GitHub.
    2. Try submodule `providers`' list of providers.
    3. Log warning that providers list from Materials-Consortia is not included in the
       `/links`-endpoint.

    Arguments:
        add_mongo_id: Whether to populate the `_id` field of the provider with MongoDB
            ObjectID.

    Returns:
        List of raw JSON-decoded providers including MongoDB object IDs.

    """
    import json

    import requests

    for provider_list_url in PROVIDER_LIST_URLS:
        try:
            providers = requests.get(provider_list_url).json()
        except (
            requests.exceptions.ConnectionError,
            requests.exceptions.ConnectTimeout,
            json.JSONDecodeError,
        ):
            pass
        else:
            break
    else:
        try:
            from optimade.server.data import providers  # type: ignore
        except ImportError:
            from optimade.server.logger import LOGGER

            LOGGER.warning(
                """Could not retrieve a list of providers!

    Tried the following resources:

{}
    The list of providers will not be included in the `/links`-endpoint.
""".format(
                    "".join([f"    * {_}\n" for _ in PROVIDER_LIST_URLS])
                )
            )
            return []

    providers_list = []
    for provider in providers.get("data", []):
        # Remove/skip "exmpl"
        if provider["id"] == "exmpl":
            continue

        provider.update(provider.pop("attributes", {}))

        # Add MongoDB ObjectId
        if add_mongo_id:
            provider["_id"] = {
                "$oid": mongo_id_for_database(provider["id"], provider["type"])
            }

        providers_list.append(provider)

    return providers_list

get_single_entry(collection, entry_id, request, params)

Source code in optimade/server/routers/utils.py

def get_single_entry(
    collection: EntryCollection,
    entry_id: str,
    request: Request,
    params: SingleEntryQueryParams,
) -> dict[str, Any]:
    from optimade.server.routers import ENTRY_COLLECTIONS

    params.check_params(request.query_params)
    params.filter = f'id="{entry_id}"'  # type: ignore[attr-defined]
    (
        results,
        data_returned,
        more_data_available,
        fields,
        include_fields,
    ) = collection.find(params)

    if more_data_available:
        raise InternalServerError(
            detail=f"more_data_available MUST be False for single entry response, however it is {more_data_available}",
        )

    include = []
    if getattr(params, "include", False):
        include.extend(params.include.split(","))

    included = []
    if results is not None:
        included = get_included_relationships(results, ENTRY_COLLECTIONS, include)

    links = ToplevelLinks(next=None)

    if results is not None and (fields or include_fields):
        results = handle_response_fields(results, fields, include_fields)[0]  # type: ignore[assignment]

    return {
        "links": links,
        "data": results if results else None,
        "meta": meta_values(
            url=request.url,
            data_returned=data_returned,
            data_available=len(collection),
            more_data_available=more_data_available,
            schema=CONFIG.schema_url
            if not CONFIG.is_index
            else CONFIG.index_schema_url,
        ),
        "included": included,
    }

handle_response_fields(results, exclude_fields, include_fields)

Handle query parameter response_fields.

It is assumed that all fields are under attributes. This is due to all other top-level fields are REQUIRED in the response.

Parameters:

Name	Type	Description	Default
exclude_fields	set[str]	Fields under attributes to be excluded from the response.	required
include_fields	set[str]	Fields under attributes that were requested that should be set to null if missing in the entry.	required

Returns:

Type	Description
list[dict[str, Any]]	List of resulting resources as dictionaries after pruning according to
list[dict[str, Any]]	the response_fields OPTIMADE URL query parameter.

Source code in optimade/server/routers/utils.py

def handle_response_fields(
    results: Union[list[EntryResource], EntryResource, list[dict], dict],
    exclude_fields: set[str],
    include_fields: set[str],
) -> list[dict[str, Any]]:
    """Handle query parameter `response_fields`.

    It is assumed that all fields are under `attributes`.
    This is due to all other top-level fields are REQUIRED in the response.

    Parameters:
        exclude_fields: Fields under `attributes` to be excluded from the response.
        include_fields: Fields under `attributes` that were requested that should be
            set to null if missing in the entry.

    Returns:
        List of resulting resources as dictionaries after pruning according to
        the `response_fields` OPTIMADE URL query parameter.

    """
    if not isinstance(results, list):
        results = [results]

    new_results = []
    while results:
        new_entry = results.pop(0)
        try:
            new_entry = new_entry.model_dump(exclude_unset=True, by_alias=True)  # type: ignore[union-attr]
        except AttributeError:
            pass

        # Remove fields excluded by their omission in `response_fields`
        for field in exclude_fields:
            if field in new_entry["attributes"]:
                del new_entry["attributes"][field]

        # Include missing fields that were requested in `response_fields`
        for field in include_fields:
            if field not in new_entry["attributes"]:
                new_entry["attributes"][field] = None

        new_results.append(new_entry)

    return new_results

meta_values(url, data_returned, data_available, more_data_available, schema=None, **kwargs)

Helper to initialize the meta values

Source code in optimade/server/routers/utils.py

def meta_values(
    url: Union[urllib.parse.ParseResult, urllib.parse.SplitResult, StarletteURL, str],
    data_returned: Optional[int],
    data_available: int,
    more_data_available: bool,
    schema: Optional[str] = None,
    **kwargs,
) -> ResponseMeta:
    """Helper to initialize the meta values"""
    from optimade.models import ResponseMetaQuery

    if isinstance(url, str):
        url = urllib.parse.urlparse(url)

    # To catch all (valid) variations of the version part of the URL, a regex is used
    if re.match(r"/v[0-9]+(\.[0-9]+){,2}/.*", url.path) is not None:
        url_path = re.sub(r"/v[0-9]+(\.[0-9]+){,2}/", "/", url.path)
    else:
        url_path = url.path

    if schema is None:
        schema = CONFIG.schema_url if not CONFIG.is_index else CONFIG.index_schema_url

    return ResponseMeta(
        query=ResponseMetaQuery(representation=f"{url_path}?{url.query}"),
        api_version=__api_version__,
        time_stamp=datetime.now(),
        data_returned=data_returned,
        more_data_available=more_data_available,
        provider=CONFIG.provider,
        data_available=data_available,
        implementation=CONFIG.implementation,
        schema=schema,
        **kwargs,
    )

mongo_id_for_database(database_id, database_type)

Produce a MongoDB ObjectId for a database

Source code in optimade/utils.py

def mongo_id_for_database(database_id: str, database_type: str) -> str:
    """Produce a MongoDB ObjectId for a database"""
    from bson.objectid import ObjectId

    oid = f"{database_id}{database_type}"
    if len(oid) > 12:
        oid = oid[:12]
    elif len(oid) < 12:
        oid = f"{oid}{'0' * (12 - len(oid))}"

    return str(ObjectId(oid.encode("UTF-8")))