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.

Client

Source code in optimade/validator/utils.py

class Client:  # pragma: no cover
    def __init__(
        self,
        base_url: str,
        max_retries: int = 5,
        headers: dict[str, str] | None = None,
        timeout: float | None = DEFAULT_CONN_TIMEOUT,
        read_timeout: float | None = 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: str | None = None
        self.response: requests.Response | None = 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	dict[str, str] | None	Dictionary of additional headers to add to every request.	None
timeout	float | None	Connection timeout in seconds.	DEFAULT_CONN_TIMEOUT
read_timeout	float | None	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: dict[str, str] | None = None,
    timeout: float | None = DEFAULT_CONN_TIMEOUT,
    read_timeout: float | None = 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: str | None = None
    self.response: requests.Response | None = 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)

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.

    """

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."""

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: str | None = 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: str | None = 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