utils

This submodule contains utility methods and models used by the validator. The two main features being:

1. The @test_case decorator can be used to decorate validation methods and performs error handling, output and logging of test successes and failures.
2. The patched Validator versions allow for stricter validation of server responses. The standard response classes allow entries to be provided as bare dictionaries, whilst these patched classes force them to be validated with the corresponding entry models themselves.

DEFAULT_CONN_TIMEOUT = 3.05 module-attribute

DEFAULT_READ_TIMEOUT = 60 module-attribute

DEFAULT_USER_AGENT_STRING = f'optimade-python-tools validator/{__version__}' module-attribute

__version__ = '1.0.1' module-attribute

Client

Source code in optimade/validator/utils.py

class Client:  # pragma: no cover
    def __init__(
        self,
        base_url: str,
        max_retries: int = 5,
        headers: Optional[dict[str, str]] = None,
        timeout: Optional[float] = DEFAULT_CONN_TIMEOUT,
        read_timeout: Optional[float] = DEFAULT_READ_TIMEOUT,
    ) -> None:
        """Initialises the Client with the given `base_url` without testing
        if it is valid.

        Parameters:
            base_url: the base URL of the optimade implementation, including
                request protocol (e.g. `'http://'`) and API version number if necessary.

                Examples:

                - `'http://example.org/optimade/v1'`,
                - `'www.crystallography.net/cod-test/optimade/v0.10.0/'`

                Note: A maximum of one slash ("/") is allowed as the last character.

            max_retries: The maximum number of attempts to make for each query.
            headers: Dictionary of additional headers to add to every request.
            timeout: Connection timeout in seconds.
            read_timeout: Read timeout in seconds.

        """
        self.base_url: str = base_url
        self.last_request: Optional[str] = None
        self.response: Optional[requests.Response] = None
        self.max_retries = max_retries
        self.headers = headers or {}
        if "User-Agent" not in self.headers:
            self.headers["User-Agent"] = DEFAULT_USER_AGENT_STRING
        self.timeout = timeout or DEFAULT_CONN_TIMEOUT
        self.read_timeout = read_timeout or DEFAULT_READ_TIMEOUT

    def get(self, request: str):
        """Makes the given request, with a number of retries if being rate limited. The
        request will be prepended with the `base_url` unless the request appears to be an
        absolute URL (i.e. starts with `http://` or `https://`).

        Parameters:
            request (str): the request to make against the base URL of this client.

        Returns:
            response (requests.models.Response): the response from the server.

        Raises:
            SystemExit: if there is no response from the server, or if the URL is invalid.
            ResponseError: if the server does not respond with a non-429 status code within
                the `MAX_RETRIES` attempts.

        """
        if urllib.parse.urlparse(request, allow_fragments=True).scheme:
            self.last_request = request
        else:
            if request and not request.startswith("/"):
                request = f"/{request}"
            self.last_request = f"{self.base_url}{request}"

        status_code = None
        retries = 0
        errors = []
        while retries < self.max_retries:
            retries += 1
            try:
                self.response = requests.get(
                    self.last_request,
                    headers=self.headers,
                    timeout=(self.timeout, self.read_timeout),
                )

                status_code = self.response.status_code
                # If we hit a 429 Too Many Requests status, then try again in 1 second
                if status_code != 429:
                    return self.response

            # If the connection times out, retry but cache the error
            except requests.exceptions.ConnectionError as exc:
                errors.append(str(exc))

            # Read timeouts should prevent further retries
            except requests.exceptions.ReadTimeout as exc:
                raise ResponseError(str(exc)) from exc

            except requests.exceptions.MissingSchema:
                sys.exit(
                    f"Unable to make request on {self.last_request}, did you mean http://{self.last_request}?"
                )

            # If the connection failed, or returned a 429, then wait 1 second before retrying
            time.sleep(1)

        else:
            message = f"Hit max retries ({self.max_retries}) on request {self.last_request!r}."
            if errors:
                error_str = "\n\t".join(errors)
                message += f"\nErrors:\n\t{error_str}"
            raise ResponseError(message)

__init__(base_url, max_retries=5, headers=None, timeout=DEFAULT_CONN_TIMEOUT, read_timeout=DEFAULT_READ_TIMEOUT)

Initialises the Client with the given base_url without testing if it is valid.

Parameters:

Name	Type	Description	Default
base_url	str	the base URL of the optimade implementation, including request protocol (e.g. 'http://') and API version number if necessary. Examples: 'http://example.org/optimade/v1', 'www.crystallography.net/cod-test/optimade/v0.10.0/' Note: A maximum of one slash ("/") is allowed as the last character.	required
max_retries	int	The maximum number of attempts to make for each query.	5
headers	Optional[dict[str, str]]	Dictionary of additional headers to add to every request.	None
timeout	Optional[float]	Connection timeout in seconds.	DEFAULT_CONN_TIMEOUT
read_timeout	Optional[float]	Read timeout in seconds.	DEFAULT_READ_TIMEOUT

Source code in optimade/validator/utils.py

def __init__(
    self,
    base_url: str,
    max_retries: int = 5,
    headers: Optional[dict[str, str]] = None,
    timeout: Optional[float] = DEFAULT_CONN_TIMEOUT,
    read_timeout: Optional[float] = DEFAULT_READ_TIMEOUT,
) -> None:
    """Initialises the Client with the given `base_url` without testing
    if it is valid.

    Parameters:
        base_url: the base URL of the optimade implementation, including
            request protocol (e.g. `'http://'`) and API version number if necessary.

            Examples:

            - `'http://example.org/optimade/v1'`,
            - `'www.crystallography.net/cod-test/optimade/v0.10.0/'`

            Note: A maximum of one slash ("/") is allowed as the last character.

        max_retries: The maximum number of attempts to make for each query.
        headers: Dictionary of additional headers to add to every request.
        timeout: Connection timeout in seconds.
        read_timeout: Read timeout in seconds.

    """
    self.base_url: str = base_url
    self.last_request: Optional[str] = None
    self.response: Optional[requests.Response] = None
    self.max_retries = max_retries
    self.headers = headers or {}
    if "User-Agent" not in self.headers:
        self.headers["User-Agent"] = DEFAULT_USER_AGENT_STRING
    self.timeout = timeout or DEFAULT_CONN_TIMEOUT
    self.read_timeout = read_timeout or DEFAULT_READ_TIMEOUT

get(request)

Makes the given request, with a number of retries if being rate limited. The request will be prepended with the base_url unless the request appears to be an absolute URL (i.e. starts with http:// or https://).

Parameters:

Name	Type	Description	Default
request	str	the request to make against the base URL of this client.	required

Returns:

Name	Type	Description
response	Response	the response from the server.

Raises:

Type	Description
SystemExit	if there is no response from the server, or if the URL is invalid.
ResponseError	if the server does not respond with a non-429 status code within the MAX_RETRIES attempts.

Source code in optimade/validator/utils.py

def get(self, request: str):
    """Makes the given request, with a number of retries if being rate limited. The
    request will be prepended with the `base_url` unless the request appears to be an
    absolute URL (i.e. starts with `http://` or `https://`).

    Parameters:
        request (str): the request to make against the base URL of this client.

    Returns:
        response (requests.models.Response): the response from the server.

    Raises:
        SystemExit: if there is no response from the server, or if the URL is invalid.
        ResponseError: if the server does not respond with a non-429 status code within
            the `MAX_RETRIES` attempts.

    """
    if urllib.parse.urlparse(request, allow_fragments=True).scheme:
        self.last_request = request
    else:
        if request and not request.startswith("/"):
            request = f"/{request}"
        self.last_request = f"{self.base_url}{request}"

    status_code = None
    retries = 0
    errors = []
    while retries < self.max_retries:
        retries += 1
        try:
            self.response = requests.get(
                self.last_request,
                headers=self.headers,
                timeout=(self.timeout, self.read_timeout),
            )

            status_code = self.response.status_code
            # If we hit a 429 Too Many Requests status, then try again in 1 second
            if status_code != 429:
                return self.response

        # If the connection times out, retry but cache the error
        except requests.exceptions.ConnectionError as exc:
            errors.append(str(exc))

        # Read timeouts should prevent further retries
        except requests.exceptions.ReadTimeout as exc:
            raise ResponseError(str(exc)) from exc

        except requests.exceptions.MissingSchema:
            sys.exit(
                f"Unable to make request on {self.last_request}, did you mean http://{self.last_request}?"
            )

        # If the connection failed, or returned a 429, then wait 1 second before retrying
        time.sleep(1)

    else:
        message = f"Hit max retries ({self.max_retries}) on request {self.last_request!r}."
        if errors:
            error_str = "\n\t".join(errors)
            message += f"\nErrors:\n\t{error_str}"
        raise ResponseError(message)

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

InternalError

Bases: Exception

This exception should be raised when validation throws an unexpected error. These should be counted separately from ResponseError's and ValidationError's.

Source code in optimade/validator/utils.py

class InternalError(Exception):
    """This exception should be raised when validation throws an unexpected error.
    These should be counted separately from `ResponseError`'s and `ValidationError`'s.

    """

LinksResource

Bases: EntryResource

A Links endpoint resource object

Source code in optimade/models/links.py

class LinksResource(EntryResource):
    """A Links endpoint resource object"""

    type: Annotated[
        Literal["links"],
        StrictField(
            description="These objects are described in detail in the section Links Endpoint",
            pattern="^links$",
        ),
    ] = "links"

    attributes: Annotated[
        LinksResourceAttributes,
        StrictField(
            description="A dictionary containing key-value pairs representing the Links resource's properties.",
        ),
    ]

    @model_validator(mode="after")
    def relationships_must_not_be_present(self) -> "LinksResource":
        if self.relationships or "relationships" in self.model_fields_set:
            raise ValueError('"relationships" is not allowed for links resources')
        return self

ReferenceResource

Bases: EntryResource

The references entries describe bibliographic references.

The following properties are used to provide the bibliographic details:

• address, annote, booktitle, chapter, crossref, edition, howpublished, institution, journal, key, month, note, number, organization, pages, publisher, school, series, title, volume, year: meanings of these properties match the BibTeX specification, values are strings;
• bib_type: type of the reference, corresponding to type property in the BibTeX specification, value is string;
• authors and editors: lists of person objects which are dictionaries with the following keys:
  • name: Full name of the person, REQUIRED.
  • firstname, lastname: Parts of the person's name, OPTIONAL.
• doi and url: values are strings.
• Requirements/Conventions:
  • Support: OPTIONAL support in implementations, i.e., any of the properties MAY be null.
  • Query: Support for queries on any of these properties is OPTIONAL. If supported, filters MAY support only a subset of comparison operators.
  • Every references entry MUST contain at least one of the properties.

Source code in optimade/models/references.py

class ReferenceResource(EntryResource):
    """The `references` entries describe bibliographic references.

    The following properties are used to provide the bibliographic details:

    - **address**, **annote**, **booktitle**, **chapter**, **crossref**, **edition**, **howpublished**, **institution**, **journal**, **key**, **month**, **note**, **number**, **organization**, **pages**, **publisher**, **school**, **series**, **title**, **volume**, **year**: meanings of these properties match the [BibTeX specification](http://bibtexml.sourceforge.net/btxdoc.pdf), values are strings;
    - **bib_type**: type of the reference, corresponding to **type** property in the BibTeX specification, value is string;
    - **authors** and **editors**: lists of *person objects* which are dictionaries with the following keys:
        - **name**: Full name of the person, REQUIRED.
        - **firstname**, **lastname**: Parts of the person's name, OPTIONAL.
    - **doi** and **url**: values are strings.
    - **Requirements/Conventions**:
        - **Support**: OPTIONAL support in implementations, i.e., any of the properties MAY be `null`.
        - **Query**: Support for queries on any of these properties is OPTIONAL.
            If supported, filters MAY support only a subset of comparison operators.
        - Every references entry MUST contain at least one of the properties.

    """

    type: Annotated[
        Literal["references"],
        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"`""",
            pattern="^references$",
            support=SupportLevel.MUST,
            queryable=SupportLevel.MUST,
        ),
    ] = "references"
    attributes: ReferenceResourceAttributes

    @field_validator("attributes", mode="before")
    @classmethod
    def validate_attributes(cls, value: Any) -> dict[str, Any]:
        if not isinstance(value, dict):
            if isinstance(value, BaseModel):
                value = value.model_dump()
            else:
                raise TypeError("attributes field must be a mapping")
        if not any(prop[1] is not None for prop in value):
            raise ValueError("reference object must have at least one field defined")
        return value

ResponseError

Bases: Exception

This exception should be raised for a manual hardcoded test failure.

Source code in optimade/validator/utils.py

class ResponseError(Exception):
    """This exception should be raised for a manual hardcoded test failure."""

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

StructureResource

Bases: EntryResource

Representing a structure.

Source code in optimade/models/structures.py

class StructureResource(EntryResource):
    """Representing a structure."""

    type: Annotated[
        Literal["structures"],
        StrictField(
            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.

- **Examples**:
    - `"structures"`""",
            pattern="^structures$",
            support=SupportLevel.MUST,
            queryable=SupportLevel.MUST,
        ),
    ] = "structures"

    attributes: StructureResourceAttributes

Success

Bases: Response

errors are not allowed

Source code in optimade/models/optimade_json.py

class Success(jsonapi.Response):
    """errors are not allowed"""

    meta: Annotated[
        ResponseMeta,
        StrictField(description="A meta object containing non-standard information"),
    ]

    @model_validator(mode="after")
    def either_data_meta_or_errors_must_be_set(self) -> "Success":
        """Overwriting the existing validation function, since 'errors' MUST NOT be set."""
        required_fields = ("data", "meta")
        if not any(field in self.model_fields_set for field in required_fields):
            raise ValueError(
                f"At least one of {required_fields} MUST be specified in the top-level response."
            )

        # errors MUST be skipped
        if self.errors or "errors" in self.model_fields_set:
            raise ValueError("'errors' MUST be skipped for a successful response.")

        return self

model_config = ConfigDict(json_encoders={datetime: lambda : v.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}) class-attribute instance-attribute

The specification mandates that datetimes must be encoded following RFC3339, which does not support fractional seconds, thus they must be stripped in the response. This can cause issues when the underlying database contains fields that do include microseconds, as filters may return unexpected results.

either_data_meta_or_errors_must_be_set()

Overwriting the existing validation function, since 'errors' MUST NOT be set.

Source code in optimade/models/optimade_json.py

@model_validator(mode="after")
def either_data_meta_or_errors_must_be_set(self) -> "Success":
    """Overwriting the existing validation function, since 'errors' MUST NOT be set."""
    required_fields = ("data", "meta")
    if not any(field in self.model_fields_set for field in required_fields):
        raise ValueError(
            f"At least one of {required_fields} MUST be specified in the top-level response."
        )

    # errors MUST be skipped
    if self.errors or "errors" in self.model_fields_set:
        raise ValueError("'errors' MUST be skipped for a successful response.")

    return self

ValidatorEntryResponseMany

Bases: Success

Source code in optimade/validator/utils.py

class ValidatorEntryResponseMany(Success):
    meta: ResponseMeta = Field(...)
    data: list[EntryResource] = Field(...)
    included: Optional[list[dict[str, Any]]] = Field(None)  # type: ignore[assignment]

model_config = ConfigDict(json_encoders={datetime: lambda : v.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}) class-attribute instance-attribute

The specification mandates that datetimes must be encoded following RFC3339, which does not support fractional seconds, thus they must be stripped in the response. This can cause issues when the underlying database contains fields that do include microseconds, as filters may return unexpected results.

either_data_meta_or_errors_must_be_set()

Overwriting the existing validation function, since 'errors' MUST NOT be set.

Source code in optimade/models/optimade_json.py

@model_validator(mode="after")
def either_data_meta_or_errors_must_be_set(self) -> "Success":
    """Overwriting the existing validation function, since 'errors' MUST NOT be set."""
    required_fields = ("data", "meta")
    if not any(field in self.model_fields_set for field in required_fields):
        raise ValueError(
            f"At least one of {required_fields} MUST be specified in the top-level response."
        )

    # errors MUST be skipped
    if self.errors or "errors" in self.model_fields_set:
        raise ValueError("'errors' MUST be skipped for a successful response.")

    return self

ValidatorEntryResponseOne

Bases: Success

Source code in optimade/validator/utils.py

class ValidatorEntryResponseOne(Success):
    meta: ResponseMeta = Field(...)
    data: EntryResource = Field(...)
    included: Optional[list[dict[str, Any]]] = Field(None)  # type: ignore[assignment]

model_config = ConfigDict(json_encoders={datetime: lambda : v.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}) class-attribute instance-attribute

The specification mandates that datetimes must be encoded following RFC3339, which does not support fractional seconds, thus they must be stripped in the response. This can cause issues when the underlying database contains fields that do include microseconds, as filters may return unexpected results.

either_data_meta_or_errors_must_be_set()

Overwriting the existing validation function, since 'errors' MUST NOT be set.

Source code in optimade/models/optimade_json.py

@model_validator(mode="after")
def either_data_meta_or_errors_must_be_set(self) -> "Success":
    """Overwriting the existing validation function, since 'errors' MUST NOT be set."""
    required_fields = ("data", "meta")
    if not any(field in self.model_fields_set for field in required_fields):
        raise ValueError(
            f"At least one of {required_fields} MUST be specified in the top-level response."
        )

    # errors MUST be skipped
    if self.errors or "errors" in self.model_fields_set:
        raise ValueError("'errors' MUST be skipped for a successful response.")

    return self

ValidatorLinksResponse

Bases: Success

Source code in optimade/validator/utils.py

class ValidatorLinksResponse(Success):
    meta: ResponseMeta = Field(...)
    data: list[LinksResource] = Field(...)

model_config = ConfigDict(json_encoders={datetime: lambda : v.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}) class-attribute instance-attribute

The specification mandates that datetimes must be encoded following RFC3339, which does not support fractional seconds, thus they must be stripped in the response. This can cause issues when the underlying database contains fields that do include microseconds, as filters may return unexpected results.

either_data_meta_or_errors_must_be_set()

Overwriting the existing validation function, since 'errors' MUST NOT be set.

Source code in optimade/models/optimade_json.py

@model_validator(mode="after")
def either_data_meta_or_errors_must_be_set(self) -> "Success":
    """Overwriting the existing validation function, since 'errors' MUST NOT be set."""
    required_fields = ("data", "meta")
    if not any(field in self.model_fields_set for field in required_fields):
        raise ValueError(
            f"At least one of {required_fields} MUST be specified in the top-level response."
        )

    # errors MUST be skipped
    if self.errors or "errors" in self.model_fields_set:
        raise ValueError("'errors' MUST be skipped for a successful response.")

    return self

ValidatorReferenceResponseMany

Bases: ValidatorEntryResponseMany

Source code in optimade/validator/utils.py

class ValidatorReferenceResponseMany(ValidatorEntryResponseMany):
    data: list[ReferenceResource] = Field(...)

model_config = ConfigDict(json_encoders={datetime: lambda : v.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}) class-attribute instance-attribute

The specification mandates that datetimes must be encoded following RFC3339, which does not support fractional seconds, thus they must be stripped in the response. This can cause issues when the underlying database contains fields that do include microseconds, as filters may return unexpected results.

either_data_meta_or_errors_must_be_set()

Overwriting the existing validation function, since 'errors' MUST NOT be set.

Source code in optimade/models/optimade_json.py

@model_validator(mode="after")
def either_data_meta_or_errors_must_be_set(self) -> "Success":
    """Overwriting the existing validation function, since 'errors' MUST NOT be set."""
    required_fields = ("data", "meta")
    if not any(field in self.model_fields_set for field in required_fields):
        raise ValueError(
            f"At least one of {required_fields} MUST be specified in the top-level response."
        )

    # errors MUST be skipped
    if self.errors or "errors" in self.model_fields_set:
        raise ValueError("'errors' MUST be skipped for a successful response.")

    return self

ValidatorReferenceResponseOne

Bases: ValidatorEntryResponseOne

Source code in optimade/validator/utils.py

class ValidatorReferenceResponseOne(ValidatorEntryResponseOne):
    data: ReferenceResource = Field(...)

model_config = ConfigDict(json_encoders={datetime: lambda : v.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}) class-attribute instance-attribute

The specification mandates that datetimes must be encoded following RFC3339, which does not support fractional seconds, thus they must be stripped in the response. This can cause issues when the underlying database contains fields that do include microseconds, as filters may return unexpected results.

either_data_meta_or_errors_must_be_set()

Overwriting the existing validation function, since 'errors' MUST NOT be set.

Source code in optimade/models/optimade_json.py

@model_validator(mode="after")
def either_data_meta_or_errors_must_be_set(self) -> "Success":
    """Overwriting the existing validation function, since 'errors' MUST NOT be set."""
    required_fields = ("data", "meta")
    if not any(field in self.model_fields_set for field in required_fields):
        raise ValueError(
            f"At least one of {required_fields} MUST be specified in the top-level response."
        )

    # errors MUST be skipped
    if self.errors or "errors" in self.model_fields_set:
        raise ValueError("'errors' MUST be skipped for a successful response.")

    return self

ValidatorResults dataclass

A dataclass to store and print the validation results.

Source code in optimade/validator/utils.py

@dataclasses.dataclass
class ValidatorResults:
    """A dataclass to store and print the validation results."""

    success_count: int = 0
    failure_count: int = 0
    internal_failure_count: int = 0
    optional_success_count: int = 0
    optional_failure_count: int = 0
    failure_messages: list[tuple[str, str]] = dataclasses.field(default_factory=list)
    internal_failure_messages: list[tuple[str, str]] = dataclasses.field(
        default_factory=list
    )
    optional_failure_messages: list[tuple[str, str]] = dataclasses.field(
        default_factory=list
    )
    verbosity: int = 0

    def add_success(self, summary: str, success_type: Optional[str] = None):
        """Register a validation success to the results class.

        Parameters:
            summary: A summary of the success to be printed.
            success_type: Either `None` or `"optional"` depending on the
                type of the check.

        """
        success_types = (None, "optional")
        if success_type not in success_types:
            raise RuntimeError(
                f"`success_type` must be one of {success_types}, not {success_type}."
            )

        if success_type is None:
            self.success_count += 1
        elif success_type == "optional":
            self.optional_success_count += 1

        message = f"✔: {summary}"
        pretty_print = print if success_type == "optional" else print_success

        if self.verbosity > 0:
            pretty_print(message)  # type: ignore[operator]
        elif self.verbosity == 0:
            pretty_print(".", end="", flush=True)  # type: ignore[operator]

    def add_failure(
        self, summary: str, message: str, failure_type: Optional[str] = None
    ):
        """Register a validation failure to the results class with
        corresponding summary, message and type.

        Parameters:
            summary: Short error message.
            message: Full error message, potentially containing a traceback.
            failure_type: Either `None`, `"internal"` or `"optional"`
                depending on the type of check that was failed.

        """
        failure_types = (None, "internal", "optional")
        if failure_type not in failure_types:
            raise RuntimeError(
                f"`failure_type` must be one of {failure_types}, not {failure_type}."
            )

        if failure_type is None:
            self.failure_count += 1
            self.failure_messages.append((summary, message))
        elif failure_type == "internal":
            self.internal_failure_count += 1
            self.internal_failure_messages.append((summary, message))
        elif failure_type == "optional":
            self.optional_failure_count += 1
            self.optional_failure_messages.append((summary, message))

        pprint_types: dict[str, tuple[Callable, Callable]] = {
            "internal": (print_notify, print_warning),
            "optional": (print, print),
        }
        pprint, warning_pprint = pprint_types.get(
            str(failure_type), (print_failure, print_warning)
        )

        symbol = "!" if failure_type == "internal" else "✖"
        if self.verbosity == 0:
            pprint(symbol, end="", flush=True)
        elif self.verbosity > 0:
            pprint(f"{symbol}: {summary}")
            for line in message.split("\n"):
                warning_pprint(f"\t{line}")

add_failure(summary, message, failure_type=None)

Register a validation failure to the results class with corresponding summary, message and type.

Parameters:

Name	Type	Description	Default
summary	str	Short error message.	required
message	str	Full error message, potentially containing a traceback.	required
failure_type	Optional[str]	Either None, "internal" or "optional" depending on the type of check that was failed.	None

Source code in optimade/validator/utils.py

def add_failure(
    self, summary: str, message: str, failure_type: Optional[str] = None
):
    """Register a validation failure to the results class with
    corresponding summary, message and type.

    Parameters:
        summary: Short error message.
        message: Full error message, potentially containing a traceback.
        failure_type: Either `None`, `"internal"` or `"optional"`
            depending on the type of check that was failed.

    """
    failure_types = (None, "internal", "optional")
    if failure_type not in failure_types:
        raise RuntimeError(
            f"`failure_type` must be one of {failure_types}, not {failure_type}."
        )

    if failure_type is None:
        self.failure_count += 1
        self.failure_messages.append((summary, message))
    elif failure_type == "internal":
        self.internal_failure_count += 1
        self.internal_failure_messages.append((summary, message))
    elif failure_type == "optional":
        self.optional_failure_count += 1
        self.optional_failure_messages.append((summary, message))

    pprint_types: dict[str, tuple[Callable, Callable]] = {
        "internal": (print_notify, print_warning),
        "optional": (print, print),
    }
    pprint, warning_pprint = pprint_types.get(
        str(failure_type), (print_failure, print_warning)
    )

    symbol = "!" if failure_type == "internal" else "✖"
    if self.verbosity == 0:
        pprint(symbol, end="", flush=True)
    elif self.verbosity > 0:
        pprint(f"{symbol}: {summary}")
        for line in message.split("\n"):
            warning_pprint(f"\t{line}")

add_success(summary, success_type=None)

Register a validation success to the results class.

Parameters:

Name	Type	Description	Default
summary	str	A summary of the success to be printed.	required
success_type	Optional[str]	Either None or "optional" depending on the type of the check.	None

Source code in optimade/validator/utils.py

def add_success(self, summary: str, success_type: Optional[str] = None):
    """Register a validation success to the results class.

    Parameters:
        summary: A summary of the success to be printed.
        success_type: Either `None` or `"optional"` depending on the
            type of the check.

    """
    success_types = (None, "optional")
    if success_type not in success_types:
        raise RuntimeError(
            f"`success_type` must be one of {success_types}, not {success_type}."
        )

    if success_type is None:
        self.success_count += 1
    elif success_type == "optional":
        self.optional_success_count += 1

    message = f"✔: {summary}"
    pretty_print = print if success_type == "optional" else print_success

    if self.verbosity > 0:
        pretty_print(message)  # type: ignore[operator]
    elif self.verbosity == 0:
        pretty_print(".", end="", flush=True)  # type: ignore[operator]

ValidatorStructureResponseMany

Bases: ValidatorEntryResponseMany

Source code in optimade/validator/utils.py

class ValidatorStructureResponseMany(ValidatorEntryResponseMany):
    data: list[StructureResource] = Field(...)

model_config = ConfigDict(json_encoders={datetime: lambda : v.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}) class-attribute instance-attribute

The specification mandates that datetimes must be encoded following RFC3339, which does not support fractional seconds, thus they must be stripped in the response. This can cause issues when the underlying database contains fields that do include microseconds, as filters may return unexpected results.

either_data_meta_or_errors_must_be_set()

Overwriting the existing validation function, since 'errors' MUST NOT be set.

Source code in optimade/models/optimade_json.py

@model_validator(mode="after")
def either_data_meta_or_errors_must_be_set(self) -> "Success":
    """Overwriting the existing validation function, since 'errors' MUST NOT be set."""
    required_fields = ("data", "meta")
    if not any(field in self.model_fields_set for field in required_fields):
        raise ValueError(
            f"At least one of {required_fields} MUST be specified in the top-level response."
        )

    # errors MUST be skipped
    if self.errors or "errors" in self.model_fields_set:
        raise ValueError("'errors' MUST be skipped for a successful response.")

    return self

ValidatorStructureResponseOne

Bases: ValidatorEntryResponseOne

Source code in optimade/validator/utils.py

class ValidatorStructureResponseOne(ValidatorEntryResponseOne):
    data: StructureResource = Field(...)

model_config = ConfigDict(json_encoders={datetime: lambda : v.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}) class-attribute instance-attribute

The specification mandates that datetimes must be encoded following RFC3339, which does not support fractional seconds, thus they must be stripped in the response. This can cause issues when the underlying database contains fields that do include microseconds, as filters may return unexpected results.

either_data_meta_or_errors_must_be_set()

Overwriting the existing validation function, since 'errors' MUST NOT be set.

Source code in optimade/models/optimade_json.py

@model_validator(mode="after")
def either_data_meta_or_errors_must_be_set(self) -> "Success":
    """Overwriting the existing validation function, since 'errors' MUST NOT be set."""
    required_fields = ("data", "meta")
    if not any(field in self.model_fields_set for field in required_fields):
        raise ValueError(
            f"At least one of {required_fields} MUST be specified in the top-level response."
        )

    # errors MUST be skipped
    if self.errors or "errors" in self.model_fields_set:
        raise ValueError("'errors' MUST be skipped for a successful response.")

    return self

print_failure(string, **kwargs)

Print but sad.

Source code in optimade/validator/utils.py

def print_failure(string: str, **kwargs) -> None:
    """Print but sad."""
    print(f"\033[91m\033[1m{string}\033[0m", **kwargs)

print_notify(string, **kwargs)

Print but louder.

Source code in optimade/validator/utils.py

def print_notify(string: str, **kwargs) -> None:
    """Print but louder."""
    print(f"\033[94m\033[1m{string}\033[0m", **kwargs)

print_success(string, **kwargs)

Print but happy.

Source code in optimade/validator/utils.py

def print_success(string: str, **kwargs) -> None:
    """Print but happy."""
    print(f"\033[92m\033[1m{string}\033[0m", **kwargs)

print_warning(string, **kwargs)

Print but angry.

Source code in optimade/validator/utils.py

def print_warning(string: str, **kwargs) -> None:
    """Print but angry."""
    print(f"\033[93m{string}\033[0m", **kwargs)

test_case(test_fn)

Wrapper for test case functions, which pretty-prints any errors depending on verbosity level, collates the number and severity of test failures, returns the response and summary string to the caller. Any additional positional or keyword arguments are passed directly to test_fn. The wrapper will intercept the named arguments optional, multistage and request and interpret them according to the docstring for wrapper(...) below.

Parameters:

Name	Type	Description	Default
test_fn	Callable[..., tuple[Any, str]]	Any function that returns an object and a message to print upon success. The function should raise a ResponseError, ValidationError or a ManualValidationError if the test case has failed. The function can return None to indicate that the test was not appropriate and should be ignored.	required

Source code in optimade/validator/utils.py

def test_case(test_fn: Callable[..., tuple[Any, str]]):
    """Wrapper for test case functions, which pretty-prints any errors
    depending on verbosity level, collates the number and severity of
    test failures, returns the response and summary string to the caller.
    Any additional positional or keyword arguments are passed directly
    to `test_fn`. The wrapper will intercept the named arguments
    `optional`, `multistage` and `request` and interpret them according
    to the docstring for `wrapper(...)` below.

    Parameters:
        test_fn: Any function that returns an object and a message to
            print upon success. The function should raise a `ResponseError`,
            `ValidationError` or a `ManualValidationError` if the test
            case has failed. The function can return `None` to indicate
            that the test was not appropriate and should be ignored.

    """
    from functools import wraps

    @wraps(test_fn)
    def wrapper(
        validator,
        *args,
        request: Optional[str] = None,
        optional: bool = False,
        multistage: bool = False,
        **kwargs,
    ):
        """Wraps a function or validator method and handles
        success, failure and output depending on the keyword
        arguments passed.

        Arguments:
            validator: The validator object to accumulate errors/counters.
            *args: Positional arguments passed to the test function.
            request: Description of the request made by the wrapped
                function (e.g. a URL or a summary).
            optional: Whether or not to treat the test as optional.
            multistage: If `True`, no output will be printed for this test,
                and it will not increment the success counter. Errors will be
                handled in the normal way. This can be used to avoid flooding
                the output for mutli-stage tests.
            **kwargs: Extra named arguments passed to the test function.

        """
        try:
            try:
                if optional and not validator.run_optional_tests:
                    result = None
                    msg = "skipping optional"
                else:
                    result, msg = test_fn(validator, *args, **kwargs)

            except (json.JSONDecodeError, ResponseError, ValidationError) as exc:
                msg = f"{exc.__class__.__name__}: {exc}"
                raise exc
            except Exception as exc:
                msg = f"{exc.__class__.__name__}: {exc}"
                raise InternalError(msg)

        # Catch SystemExit and KeyboardInterrupt explicitly so that we can pass
        # them to the finally block, where they are immediately raised
        except (Exception, SystemExit, KeyboardInterrupt) as exc:
            result = exc
            traceback = tb.format_exc()

        finally:
            # This catches the case of the Client throwing a SystemExit if the server
            # did not respond, the case of the validator "fail-fast"'ing and throwing
            # a SystemExit below, and the case of the user interrupting the process manually
            if isinstance(result, (SystemExit, KeyboardInterrupt)):
                raise result

            display_request = None
            try:
                display_request = validator.client.last_request
            except AttributeError:
                pass
            if display_request is None:
                display_request = validator.base_url
                if request is not None:
                    display_request += "/" + request

            request = display_request

            # If the result was None, return it here and ignore statuses
            if result is None:
                return result, msg
            display_request = requests.utils.requote_uri(request.replace("\n", ""))  # type: ignore[union-attr]

            if not isinstance(result, Exception):
                if not multistage:
                    success_type = "optional" if optional else None
                    validator.results.add_success(
                        f"{display_request} - {msg}", success_type
                    )
            else:
                message = msg.split("\n")
                if validator.verbosity > 1:
                    # ValidationErrors from pydantic already include very detailed errors
                    # that get duplicated in the traceback
                    if not isinstance(result, ValidationError):
                        message += traceback.split("\n")

                failure_type: Optional[str] = None
                if isinstance(result, InternalError):
                    summary = f"{display_request} - {test_fn.__name__} - failed with internal error"
                    failure_type = "internal"
                else:
                    summary = (
                        f"{display_request} - {test_fn.__name__} - failed with error"
                    )
                    failure_type = "optional" if optional else None

                validator.results.add_failure(
                    summary, "\n".join(message), failure_type=failure_type
                )

                # set failure result to None as this is expected by other functions
                result = None

                if validator.fail_fast and not optional:
                    validator.print_summary()
                    raise SystemExit

            # Reset the client request so that it can be properly
            # displayed if the next request fails
            if not multistage:
                validator.client.last_request = None

            return result, msg

    return wrapper