"""The registry of entry types and properties served by an OPTIMADE deployment.
A :class:`ServedSchema` narrows a set of first-class
:class:`~httk.core.EntryTypeDefinition` objects (supplied from outside, e.g. by
an :class:`~httk.core.EntryProvider`) down to the entry types and properties a
backend implements, and derives the endpoint and response-field tables used
during request validation and response generation.
The full OPTIMADE property definitions live in httk-core; this module keeps, for
each served property, a small *simplified* view (its ``fulltype`` and the
implementation flags) that the filter-translation and response layers consume,
alongside the full property definitions served on the entry listing info
endpoint.
"""
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Any
from httk.core import EntryTypeDefinition, PropertyDefinition
from httk.core.property_definitions import known_definition_prefixes
[docs]
def fulltype_of(definition: PropertyDefinition) -> str:
"""Reconstruct a property's simplified ``fulltype`` string.
Maps the OPTIMADE type back to the compact spelling used by the filter
layer: ``"string"``/``"integer"``/``"float"``/``"boolean"``/``"timestamp"``,
``"dict"`` for dictionaries, and ``"list of ..."`` (nesting through the
definition's ``items``) for lists.
:param definition: Property definition to inspect.
:return: Compact fulltype spelling used by the filter and response layers.
"""
return _fulltype_from_doc(definition.as_optimade())
def _fulltype_from_doc(doc: Mapping[str, Any]) -> str:
optimade_type = doc["x-optimade-type"]
if optimade_type == "list":
return "list of " + _fulltype_from_doc(doc["items"])
if optimade_type == "dictionary":
return "dict"
return optimade_type
def _is_required_response(name: str, definition: PropertyDefinition) -> bool:
"""Whether a property must always carry a non-null value in responses.
``id`` and ``type`` are always required. Otherwise a property is required
only when its OPTIMADE requirements mark the response level ``"must"`` *and*
the value is non-nullable (so, e.g., ``files.url`` is required, but a
nullable ``last_modified`` marked ``"must"`` is not).
"""
if name in ("id", "type"):
return True
return definition.requirements.get("response-level") == "must" and not definition.nullable
def _is_default_response(name: str, definition: PropertyDefinition, default_names: set[str]) -> bool:
"""Whether a served property belongs in responses without response_fields."""
if name in ("id", "type"):
return True
return name in default_names and definition.requirements.get("response-level") not in {"should not", "must not"}
def _is_queryable(name: str, definition: PropertyDefinition) -> bool:
"""Whether a served property may be used in ``filter=``.
``id`` and ``type`` are always queryable. A query support of ``"none"``
disables filtering only for provider-specific (underscore-namespaced)
properties, where it is the provider's authoritative statement that the field
must not be filtered. On STANDARD properties an ``"none"`` query support is
merely the specification's "querying not required" requirement level, which a
deployment may (and httk does) still implement, so it must not disable them.
Only the exact value ``"none"`` disables filtering; other levels keep the
default behavior.
"""
if name in ("id", "type") or not name.startswith("_"):
return True
return definition.requirements.get("query-support") != "none"
[docs]
def simplified_property(
definition: PropertyDefinition,
*,
sortable: bool = False,
required_response: bool = False,
default_response: bool = False,
queryable: bool = True,
) -> dict[str, Any]:
"""Build a simplified property view for the filter and wrapping layers.
Carries the ``description``, reconstructed ``fulltype``, the implementation
flags (``sortable``/``required_response``/``default_response``/``queryable``),
and — when present — the property's ``unit`` and ``dimensions`` (used by the
trajectory frame-wrapping).
:param definition: Property definition to simplify.
:param sortable: Mark the property as sortable by the backend.
:param required_response: Mark the property as required in responses.
:param default_response: Mark the property as returned by default.
:param queryable: Mark the property as usable in ``filter=`` (false honors
an ``x-optimade-requirements.query-support`` of ``"none"``).
:return: Simplified property metadata.
"""
info: dict[str, Any] = {
"description": definition.description,
"fulltype": fulltype_of(definition),
"sortable": sortable,
"required_response": required_response,
"default_response": default_response,
"queryable": queryable,
}
unit = definition.unit
if unit is not None and unit not in ("dimensionless", "inapplicable"):
info["unit"] = unit
dimensions = definition.dimensions
if dimensions is not None:
info["dimensions"] = {key: list(value) for key, value in dimensions.items()}
return info
[docs]
def entry_type_definition_from_simple(name: str, info: Mapping[str, Any]) -> EntryTypeDefinition:
"""Build an :class:`~httk.core.EntryTypeDefinition` from simplified metadata.
``info`` is a ``{"description": <str>, "properties": {<name>: <simplified
property dict>}}`` mapping (as produced by, e.g.,
:func:`~httk.serve.optimade.schema.trajectories.trajectories_entry_info`); each
property is generated with
:meth:`~httk.core.PropertyDefinition.from_simple`.
:param name: Entry endpoint and definition name.
:param info: Simplified entry description and property mapping.
:return: Full entry-type definition.
"""
properties = {
prop_name: PropertyDefinition.from_simple(
prop_name,
description=prop_info.get("description", ""),
fulltype=prop_info.get("fulltype", "string"),
unit=prop_info.get("unit"),
dimensions=prop_info.get("dimensions"),
dict_properties=prop_info.get("dict_properties"),
metadata_definition=prop_info.get("metadata_definition"),
required_response=prop_info.get("required_response", False),
)
for prop_name, prop_info in info["properties"].items()
}
return EntryTypeDefinition(name, info["description"], properties)
@dataclass(frozen=True)
[docs]
class ServedSchema:
"""Describe served entry types and their derived lookup tables.
:param entry_info: Simplified entry-info documents keyed by entry type.
:param entry_definition_ids: Definition IRIs keyed by entry type.
:param recognized_prefixes: Property-definition prefixes recognized in requests.
:param all_entries: Served entry endpoint names in declaration order.
:param valid_endpoints: Fixed and entry endpoint names accepted by validation.
:param properties_by_entry: Served property names keyed by entry type.
:param default_response_fields: Default response fields keyed by entry type.
:param required_response_fields: Required response fields keyed by entry type.
:param unknown_response_fields: Defined but unserved fields keyed by entry type.
:param sortable_response_fields: Sortable response fields keyed by entry type.
:param property_definitions: Full property definitions keyed by entry type.
"""
[docs]
entry_info: dict[str, dict[str, Any]]
[docs]
entry_definition_ids: dict[str, str]
[docs]
recognized_prefixes: tuple[str, ...]
[docs]
all_entries: tuple[str, ...]
[docs]
valid_endpoints: tuple[str, ...]
[docs]
properties_by_entry: dict[str, tuple[str, ...]]
[docs]
default_response_fields: dict[str, tuple[str, ...]]
[docs]
required_response_fields: dict[str, tuple[str, ...]]
[docs]
unknown_response_fields: dict[str, tuple[str, ...]]
[docs]
sortable_response_fields: dict[str, tuple[str, ...]]
[docs]
property_definitions: dict[str, dict[str, dict[str, Any]]]
[docs]
def build_served_schema(
definitions: Mapping[str, EntryTypeDefinition],
served: Mapping[str, Sequence[str]] | None = None,
*,
default_response_overrides: Mapping[str, Sequence[str]] | None = None,
sortable: Mapping[str, Sequence[str]] | None = None,
recognized_prefixes: tuple[str, ...] | None = None,
) -> ServedSchema:
"""Build a :class:`ServedSchema` from entry-type definitions.
``definitions`` maps each served entry type name to its full
:class:`~httk.core.EntryTypeDefinition`. ``served`` maps each entry type to
the subset of property names actually served (defaulting to every property
the definition describes); every served name MUST be described by the
definition (a :class:`ValueError` names any offender). ``id`` and ``type``
are always default- and required-response; ``default_response_overrides``
marks additional served properties as default-response, and ``sortable``
marks served properties as sortable. ``recognized_prefixes`` defaults to the
prefixes currently registered via :func:`~httk.core.register_definition_prefix`
(resolved at call time so newly registered prefixes are honored).
:param definitions: Full definitions keyed by served entry type.
:param served: Optional served-property subset keyed by entry type.
:param default_response_overrides: Additional default fields keyed by entry type.
:param sortable: Sortable fields keyed by entry type.
:param recognized_prefixes: Prefixes recognized in response-field requests.
:return: Derived schema and lookup tables.
:raises ValueError: If a requested served property is not defined.
"""
if recognized_prefixes is None:
recognized_prefixes = known_definition_prefixes()
entry_info: dict[str, dict[str, Any]] = {}
entry_definition_ids: dict[str, str] = {}
property_definitions: dict[str, dict[str, dict[str, Any]]] = {}
properties_by_entry: dict[str, tuple[str, ...]] = {}
default_response_fields: dict[str, tuple[str, ...]] = {}
required_response_fields: dict[str, tuple[str, ...]] = {}
unknown_response_fields: dict[str, tuple[str, ...]] = {}
sortable_response_fields: dict[str, tuple[str, ...]] = {}
for entry, definition in definitions.items():
if definition.definition_id is not None:
entry_definition_ids[entry] = definition.definition_id
described = definition.properties
served_names = list(served[entry]) if served is not None and entry in served else list(described)
missing = [name for name in served_names if name not in described]
if missing:
raise ValueError(
"Entry type '"
+ entry
+ "' is asked to serve property(ies) not described by its definition: "
+ ", ".join(missing)
+ "."
)
sortable_names = set(sortable.get(entry, ())) if sortable is not None else set()
default_names = (
set(default_response_overrides.get(entry, ())) if default_response_overrides is not None else set()
)
simplified: dict[str, Any] = {}
prop_defs: dict[str, dict[str, Any]] = {}
defaults: list[str] = []
requireds: list[str] = []
sortables: list[str] = []
for name in served_names:
prop = described[name]
is_sortable = name in sortable_names
is_default = _is_default_response(name, prop, default_names)
is_required = _is_required_response(name, prop)
is_queryable = _is_queryable(name, prop)
simplified[name] = simplified_property(
prop,
sortable=is_sortable,
required_response=is_required,
default_response=is_default,
queryable=is_queryable,
)
prop_defs[name] = prop.with_implementation(sortable=is_sortable, response_default=is_default).as_optimade()
if is_default:
defaults.append(name)
if is_required:
requireds.append(name)
if is_sortable:
sortables.append(name)
entry_info[entry] = {"description": definition.description, "properties": simplified}
property_definitions[entry] = prop_defs
properties_by_entry[entry] = tuple(served_names)
default_response_fields[entry] = tuple(defaults)
required_response_fields[entry] = tuple(requireds)
sortable_response_fields[entry] = tuple(sortables)
unknown_response_fields[entry] = tuple(name for name in described if name not in served_names)
all_entries = tuple(definitions)
return ServedSchema(
entry_info=entry_info,
entry_definition_ids=entry_definition_ids,
recognized_prefixes=recognized_prefixes,
all_entries=all_entries,
valid_endpoints=tuple(["info", "links"] + list(all_entries) + ["info/" + x for x in all_entries] + [""]),
properties_by_entry=properties_by_entry,
default_response_fields=default_response_fields,
required_response_fields=required_response_fields,
unknown_response_fields=unknown_response_fields,
sortable_response_fields=sortable_response_fields,
property_definitions=property_definitions,
)