Source code for httk.serve.optimade.model.request
from dataclasses import dataclass, field
from typing import Any
@dataclass(slots=True)
[docs]
class RawRequest:
"""Represent an incoming OPTIMADE request from the web layer.
Only ``baseurl`` and ``representation`` are mandatory; missing information
is derived from ``representation`` during validation.
:param baseurl: Base URL used when generating response links.
:param representation: Request path and query representation.
:param relurl: Relative request URL, when supplied by the web layer.
:param querystr: Raw query string.
:param query: Parsed query parameters.
:param endpoint: Preselected endpoint, when supplied by the caller.
:param request_id: Preselected entry identifier, when supplied by the caller.
:param version: API version declared by the caller.
"""
[docs]
relurl: str | None = None
[docs]
querystr: str | None = None
[docs]
query: dict[str, str] | None = None
[docs]
endpoint: str | None = None
[docs]
request_id: str | None = None
[docs]
version: str | None = None
@dataclass(slots=True)
[docs]
class RequestedSlice:
"""Represent a slice requested via ``dimension_slices``.
The values are stored exactly as given (a ``None`` component means the
client omitted it and the default applies). Per the OPTIMADE specification,
``stop`` is *inclusive*.
:param start: Inclusive first index, or ``None`` for the default.
:param stop: Inclusive last index, or ``None`` for the default.
:param step: Slice step, or ``None`` for the default.
"""
[docs]
start: int | None = None
[docs]
stop: int | None = None
[docs]
step: int | None = None
@dataclass(slots=True)
[docs]
class ValidatedParameters:
"""Represent validated URL query parameters of an OPTIMADE request.
:param response_format: Requested response format.
:param page_limit: Maximum number of entries in a page.
:param page_offset: Number of matching entries to skip.
:param response_fields: Comma-separated requested response fields.
:param filter: Raw OPTIMADE filter expression.
:param sort: Raw OPTIMADE sort expression.
:param include: Raw related-entry inclusion request.
:param as_of: Nanosecond timestamp cutoff for timestamp-capable stored sources;
timestamp-disabled sources may serve current state and generic providers ignore it.
:param dimension_slices: Requested slices keyed by dimension name.
"""
[docs]
page_limit: int = 50
[docs]
page_offset: int = 0
[docs]
response_fields: str | None = None
[docs]
filter: str | None = None
[docs]
sort: str | None = None
[docs]
include: str | None = None
[docs]
as_of: int | None = None
[docs]
dimension_slices: dict[str, RequestedSlice] = field(default_factory=dict)
[docs]
def as_query_dict(self) -> dict[str, str]:
"""Return the parameters as a URL query mapping.
:return: Query values with unset optional parameters omitted.
"""
query: dict[str, str] = {
'response_format': self.response_format,
'page_limit': str(self.page_limit),
'page_offset': str(self.page_offset),
}
if self.response_fields is not None:
query['response_fields'] = self.response_fields
if self.filter is not None:
query['filter'] = self.filter
if self.sort is not None:
query['sort'] = self.sort
if self.include is not None:
query['include'] = self.include
if self.as_of is not None:
query['_httk_as_of'] = str(self.as_of)
if self.dimension_slices:
parts = []
for name, requested in self.dimension_slices.items():
start = "" if requested.start is None else str(requested.start)
stop = "" if requested.stop is None else str(requested.stop)
step = "" if requested.step is None else str(requested.step)
parts.append(f"{name}[{start}:{stop}:{step}]")
query['dimension_slices'] = ",".join(parts)
return query
@dataclass(slots=True)
[docs]
class ValidatedRequest:
"""Represent the result of validating a :class:`RawRequest`.
:param baseurl: Base URL used when generating response links.
:param representation: Original request representation.
:param endpoint: Validated endpoint name.
:param version: Validated OPTIMADE version.
:param query: Validated query parameters.
:param url_version: Version segment present in the request URL.
:param request_id: Validated entry identifier.
:param recognized_response_fields: Requested fields known to the schema.
:param unrecognized_response_fields: Requested fields not known to the schema.
:param sort_fields: Validated sort fields and directions.
:param include_paths: Validated related-entry paths.
:param property_metadata_requested: Whether property metadata was requested.
:param partial_data_parts: Entry, identifier, and property for partial data.
:param partial_data_offset: Offset into a partial-data response.
:param warnings: Warnings collected while processing the request.
"""
[docs]
query: ValidatedParameters
[docs]
url_version: str | None = None
[docs]
request_id: str | None = None
[docs]
recognized_response_fields: list[str] = field(default_factory=list)
[docs]
unrecognized_response_fields: list[str] = field(default_factory=list)
[docs]
sort_fields: list[tuple[str, bool]] = field(default_factory=list)
[docs]
include_paths: list[str] = field(default_factory=list)
[docs]
partial_data_parts: tuple[str, str, str] | None = None
[docs]
partial_data_offset: int = 0
[docs]
warnings: list[dict[str, Any]] = field(default_factory=list)
@dataclass(slots=True)
[docs]
class EndpointResponse:
"""Represent an endpoint response for serialization by the web layer.
Either ``json_response`` (a JSON:API document) or ``content`` (a raw body)
is set.
:param response_code: HTTP status code.
:param response_msg: HTTP status title.
:param content_type: Response media type.
:param encoding: Response character encoding.
:param content: Raw response body, when the response is not JSON.
:param json_response: JSON:API response document, when the response is JSON.
"""
[docs]
response_code: int = 200
[docs]
response_msg: str = 'OK'
[docs]
content_type: str = 'application/vnd.api+json'
[docs]
encoding: str = 'utf-8'
[docs]
content: str | None = None
[docs]
json_response: dict[str, Any] | None = None