← 返回首页
utils - OPTIMADE Python tools
Skip to content
OPTIMADE Python tools
utils
optimade-python-tools

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
165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266
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
166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202
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
204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266
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
48 49 50 51 52
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
44 45
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
65 66 67
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
60 61 62
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
70 71 72
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
55 56 57
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
269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401
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