"""Serve standard entry types through in-memory :class:`~httk.core.EntryProvider` implementations.
These providers map ``{id: record}`` mappings of the stdlib-only record models
defined in *httk-core* (:class:`~httk.core.Reference`, :class:`~httk.core.File`,
:class:`~httk.core.Calculation`) onto the neutral httk-core entry-provider
contract, so a serving module (such as *httk-serve*) can expose them as
OPTIMADE ``references``/``files``/``calculations`` endpoints without either side
depending on the other. Each provider describes its entry type with the vendored
OPTIMADE standard definition loaded from httk-core via
:func:`~httk.core.standard_entry_type`.
The record *models* live in httk-core (contracts and models); these *providers*
live in httk-store (the capability layer built on those models), together with
property-definition validation. The database storage layer in
:mod:`httk.store.backend.sql` complements them with a database-backed provider
(:class:`~httk.store.backend.sql.entry_provider.StoreEntryProvider`) serving stored
dataclasses the same way.
"""
import datetime
from collections.abc import Iterable, Mapping
from dataclasses import fields
from functools import cache
from typing import Any, get_type_hints
from httk.core import (
Calculation,
DataRecord,
EntryProvider,
EntryTypeDefinition,
File,
ProductLink,
PropertyDefinition,
Reference,
RelatedEntry,
Run,
apply_definition_prefix,
load_entry_type_definition,
load_property_definition,
standard_entry_type,
)
from httk.core.data_records import RECORDS_DEFINITION_ID
from httk.core.provenance import RUNS_DEFINITION_ID
from httk.core.storage import (
QueryContext,
QueryExpression,
QueryLiteralError,
StoredPropertyProjection,
StrongLink,
)
from httk.store.query import ID_FIELD
def _normalized_relationships(
relationships: Mapping[str, Iterable[RelatedEntry]] | None,
) -> dict[str, tuple[RelatedEntry, ...]]:
"""Normalize a caller-supplied relationships mapping to ``{str(id): tuple(entries)}``."""
if relationships is None:
return {}
return {str(key): tuple(value) for key, value in relationships.items()}
def _provider_property_keys(record_type: type[Any]) -> dict[str, str]:
"""The served-property-name to record-key map for a standard entry type."""
property_keys = {"id": ID_FIELD, "type": "type"}
property_keys.update({field.name: field.name for field in fields(record_type)})
return property_keys
def _json_value(value: Any) -> Any:
"""A record value as one of the JSON types the provider contract promises.
The record models declare their sequence fields as tuples (immutable
records), but :meth:`~httk.core.EntryProvider.records` is contracted to
yield plain JSON-able values, and a JSON array is a ``list``. Passing a
tuple through reaches a consumer that type-checks against the property
definition — :func:`~httk.store.validation.validate_record` does — and is
rejected as "not of type 'array'", even though it serializes fine.
"""
if isinstance(value, tuple):
return [_json_value(item) for item in value]
if isinstance(value, list):
return [_json_value(item) for item in value]
if isinstance(value, Mapping):
return {key: _json_value(item) for key, item in value.items()}
if isinstance(value, datetime.datetime):
return value.isoformat()
return value
def _provider_records(entry_type: str, record_type: type[Any], entries: Mapping[str, Any]) -> list[dict[str, Any]]:
"""JSON-able records for a standard entry type, one per stored instance."""
field_names = [field.name for field in fields(record_type)]
records: list[dict[str, Any]] = []
for entry_id, record in entries.items():
row: dict[str, Any] = {ID_FIELD: entry_id, "type": entry_type}
for name in field_names:
row[name] = _json_value(getattr(record, name))
sha256 = getattr(record, "sha256", None)
if entry_type == "files" and sha256 and row["checksums"] is None:
row["checksums"] = {"sha256": sha256}
records.append(row)
return records
[docs]
class StandardEntryProvider(EntryProvider):
"""Serve one standard entry type through the neutral provider contract.
:param entries: The records keyed by their served identifiers.
:param record_type: The core record class used to construct each entry.
:param entry_type: The OPTIMADE entry type served by this provider.
:param relationships: Optional related entries keyed by served identifier.
"""
def __init__(
self,
entries: Mapping[str, Any],
*,
record_type: type[Any],
entry_type: str,
relationships: Mapping[str, Iterable[RelatedEntry]] | None,
) -> None:
self._record_type = record_type
self._entry_type = entry_type
self._entries = {str(key): record_type.from_obj(value) for key, value in entries.items()}
self._relationships = _normalized_relationships(relationships)
def __repr__(self) -> str:
return f"{type(self).__name__}(entry_type={self._entry_type!r}, entries={len(self._entries)})"
def _check_entry_type(self, entry_type: str) -> None:
if entry_type != self._entry_type:
raise KeyError(f"{type(self).__name__} serves only the '{self._entry_type}' entry type.")
[docs]
def entry_types(self) -> Mapping[str, EntryTypeDefinition]:
"""Return the one standard entry-type definition served by this provider.
:return: The served entry-type definition.
"""
return {self._entry_type: standard_entry_type(self._entry_type)}
[docs]
def property_keys(self, entry_type: str) -> Mapping[str, str]:
"""Return served property names mapped to record attribute names.
:param entry_type: The entry type to inspect.
:return: The served-property to record-key mapping.
:raises KeyError: If ``entry_type`` is not this provider's entry type.
"""
self._check_entry_type(entry_type)
return _provider_property_keys(self._record_type)
[docs]
def records(self, entry_type: str) -> Iterable[Mapping[str, Any]]:
"""Return JSON-compatible records for the requested entry type.
:param entry_type: The entry type to enumerate.
:return: The provider's records in mapping iteration order.
:raises KeyError: If ``entry_type`` is not this provider's entry type.
"""
self._check_entry_type(entry_type)
return _provider_records(self._entry_type, self._record_type, self._entries)
[docs]
def relationships(self, entry_type: str) -> Mapping[str, tuple[RelatedEntry, ...]]:
"""Return related entries keyed by served identifier.
:param entry_type: The entry type to inspect.
:return: The normalized relationship mapping.
:raises KeyError: If ``entry_type`` is not this provider's entry type.
"""
self._check_entry_type(entry_type)
return self._relationships
[docs]
class ReferenceEntryProvider(StandardEntryProvider):
"""Serves OPTIMADE ``references`` from a mapping of id to :class:`~httk.core.Reference`.
``relationships`` optionally maps a reference id to its related entries
(:class:`~httk.core.RelatedEntry` values, served flat per id).
:param entries: The references keyed by their served identifiers.
:param relationships: Optional related entries keyed by reference identifier.
"""
def __init__(
self,
entries: Mapping[str, Reference | Mapping[str, Any]],
*,
relationships: Mapping[str, Iterable[RelatedEntry]] | None = None,
) -> None:
super().__init__(entries, record_type=Reference, entry_type="references", relationships=relationships)
[docs]
class FileEntryProvider(StandardEntryProvider):
"""Serves OPTIMADE ``files`` from a mapping of id to :class:`~httk.core.File`.
``relationships`` optionally maps a file id to its related entries
(:class:`~httk.core.RelatedEntry` values, served flat per id) — e.g. the
calculations a file is ``input``/``output`` of.
:param entries: The files keyed by their served identifiers.
:param relationships: Optional related entries keyed by file identifier.
"""
def __init__(
self,
entries: Mapping[str, File | Mapping[str, Any]],
*,
relationships: Mapping[str, Iterable[RelatedEntry]] | None = None,
) -> None:
super().__init__(entries, record_type=File, entry_type="files", relationships=relationships)
[docs]
class CalculationEntryProvider(StandardEntryProvider):
"""Serves OPTIMADE ``calculations`` from a mapping of id to :class:`~httk.core.Calculation`.
``relationships`` optionally maps a calculation id to its related entries
(:class:`~httk.core.RelatedEntry` values, served flat per id) — e.g. its
``input``/``output`` files, expressed via the ``role`` metadata.
:param entries: The calculations keyed by their served identifiers.
:param relationships: Optional related entries keyed by calculation identifier.
"""
def __init__(
self,
entries: Mapping[str, Calculation | Mapping[str, Any]],
*,
relationships: Mapping[str, Iterable[RelatedEntry]] | None = None,
) -> None:
super().__init__(entries, record_type=Calculation, entry_type="calculations", relationships=relationships)
@cache
def _runs_definition() -> EntryTypeDefinition:
"""The wire (served) ``_httk_runs`` definition from the internal runs definition."""
return load_entry_type_definition(RUNS_DEFINITION_ID).served_form()
@cache
def _records_definition() -> EntryTypeDefinition:
"""The wire (served) ``_httk_records`` core definition (without per-record value properties)."""
return load_entry_type_definition(RECORDS_DEFINITION_ID).served_form()
@cache
def _wire_entry_type() -> Mapping[str, str]:
"""Internal entry-type names mapped to their served (wire) names.
Only the httk-core provider-defined types whose served name differs from
their internal name appear here; standard OPTIMADE type names (and any
unrecognized target) pass through :meth:`RunEntryProvider.relationships`
unchanged.
"""
mapping: dict[str, str] = {}
for definition_id in (RUNS_DEFINITION_ID, RECORDS_DEFINITION_ID):
internal = load_entry_type_definition(definition_id)
served = internal.served_form()
if served.name != internal.name:
mapping[internal.name] = served.name
return mapping
[docs]
def wire_relationship_key(internal_name: str, definition_id: str | None) -> str:
"""Return the wire (prefixed) relationship key for an internal marker name.
The single serving-edge transform from an internal
:class:`~httk.core.storage.StrongLink` marker name to its served OPTIMADE
relationship key. It applies the declaring family's definition prefix — the
exact registry source ``EntryTypeDefinition.served_form`` uses
— so an internal name such as ``has_input`` becomes ``_httk_has_input``. A
``definition_id`` of ``None`` (or one under no registered prefix) leaves the
name bare, and an already-prefixed name is never re-prefixed.
:param internal_name: The internal (unprefixed) marker relationship name.
:param definition_id: The declaring family's definition IRI, whose registered prefix is applied, or ``None``.
:return: The served (wire) relationship key.
"""
return apply_definition_prefix(internal_name, definition_id)
[docs]
def strong_link_markers(record_type: type[Any]) -> dict[str, StrongLink]:
"""Return a record class's :class:`~httk.core.storage.StrongLink` markers by field.
:param record_type: The storable record class to inspect.
:return: The strong-link markers keyed by field name, in dataclass field order.
"""
hints = get_type_hints(record_type, include_extras=True)
markers: dict[str, StrongLink] = {}
for field in fields(record_type):
annotation = hints.get(field.name)
for marker in getattr(annotation, "__metadata__", ()):
if isinstance(marker, StrongLink):
markers[field.name] = marker
return markers
_RUN_STRONG_LINKS: Mapping[str, StrongLink] = strong_link_markers(Run)
[docs]
class RunEntryProvider(EntryProvider):
"""Serve core :class:`~httk.core.Run` records and their provenance edges.
:param entries: The runs keyed by their served identifiers.
"""
def __init__(self, entries: Mapping[str, Run | Mapping[str, Any]]) -> None:
self._entry_type = _runs_definition().name
self._entries = {str(key): Run.from_obj(value) for key, value in entries.items()}
def __repr__(self) -> str:
return f"RunEntryProvider(entries={len(self._entries)})"
def _check_entry_type(self, entry_type: str) -> None:
if entry_type != self._entry_type:
raise KeyError(f"{type(self).__name__} serves only the '{self._entry_type}' entry type.")
[docs]
def entry_types(self) -> Mapping[str, EntryTypeDefinition]:
"""Return the served ``_httk_runs`` entry definition.
The wire naming is the served form of the internal runs definition (see
``EntryTypeDefinition.served_form()``).
:return: The served run entry-type definition.
"""
return {self._entry_type: _runs_definition()}
[docs]
def property_keys(self, entry_type: str) -> Mapping[str, str]:
"""Return the served run-property to record-key mapping.
:param entry_type: The entry type to inspect.
:return: The served-property to record-key mapping.
:raises KeyError: If ``entry_type`` is not ``_httk_runs``.
"""
self._check_entry_type(entry_type)
return {
"id": ID_FIELD,
"type": "type",
"immutable_id": "immutable_id",
"last_modified": "last_modified",
"_httk_workflow_declaration_uri": "workflow_declaration_uri",
"_httk_source_id": "source_id",
}
[docs]
def records(self, entry_type: str) -> Iterable[Mapping[str, Any]]:
"""Return JSON-compatible run records.
:param entry_type: The entry type to enumerate.
:yield: Run records in input mapping order.
:raises KeyError: If ``entry_type`` is not ``_httk_runs``.
"""
self._check_entry_type(entry_type)
for entry_id, run in self._entries.items():
yield {
ID_FIELD: entry_id,
"type": self._entry_type,
"immutable_id": run.immutable_id,
"last_modified": _json_value(run.last_modified),
"workflow_declaration_uri": run.workflow_declaration_uri,
"source_id": run.source_id,
}
[docs]
def relationships(self, entry_type: str) -> Mapping[str, tuple[RelatedEntry, ...]]:
"""Return run provenance edges as forward semantic relationships.
Each edge is grouped under its owning field's ``StrongLink`` forward
relationship key in wire form (e.g. ``_httk_has_input``, read from the
markers on :class:`~httk.core.Run`, not hardcoded). Run edges carry
internal entry-type names; this serving edge translates each target type
to its served (wire) name (via ``EntryTypeDefinition.served_form()``), so
a target such as ``records`` is served as ``_httk_records`` while
standard type names pass through unchanged.
:param entry_type: The entry type to inspect.
:return: Relationships grouped by run identifier.
:raises KeyError: If ``entry_type`` is not ``_httk_runs``.
"""
self._check_entry_type(entry_type)
wire = _wire_entry_type()
return {
entry_id: tuple(
RelatedEntry(
wire.get(edge.entry_type, edge.entry_type),
edge.entry_id,
role=marker.role,
label=edge.label,
relationship=wire_relationship_key(marker.relationship, RUNS_DEFINITION_ID),
)
for field_name, marker in _RUN_STRONG_LINKS.items()
for edge in getattr(run, field_name)
)
for entry_id, run in self._entries.items()
}
[docs]
def reverse_relationships(self) -> Mapping[str, Mapping[str, tuple[RelatedEntry, ...]]]:
"""Return the derived reverse view of the runs' provenance edges.
Each run edge ``(entry_type, entry_id)`` yields a reverse related entry
attached to the targeted entry: keyed by the target's served (wire) entry
type and its raw id, the related entry names this run under the edge
field's ``StrongLink`` reverse relationship key in wire form (e.g.
``_httk_is_input``). Fields whose marker declares no ``reverse`` key
contribute nothing.
:return: Related runs keyed by target entry type and then target entry id.
"""
wire = _wire_entry_type()
# Collect each reverse hit with its deterministic sort key
# (run id, marker/field index, edge row index) then sort per target,
# matching the stored SQL/Mongo routes.
result: dict[str, dict[str, list[tuple[tuple[str, int, int], RelatedEntry]]]] = {}
for entry_id, run in self._entries.items():
for marker_index, (field_name, marker) in enumerate(_RUN_STRONG_LINKS.items()):
if marker.reverse is None:
continue
reverse_key = wire_relationship_key(marker.reverse, RUNS_DEFINITION_ID)
for edge_index, edge in enumerate(getattr(run, field_name)):
target_type = wire.get(edge.entry_type, edge.entry_type)
result.setdefault(target_type, {}).setdefault(edge.entry_id, []).append(
(
(entry_id, marker_index, edge_index),
RelatedEntry(
self._entry_type,
entry_id,
role=marker.role,
label=edge.label,
relationship=reverse_key,
),
)
)
return {
target_type: {
target_id: tuple(entry for _key, entry in sorted(hits, key=lambda item: item[0]))
for target_id, hits in ids.items()
}
for target_type, ids in result.items()
}
[docs]
class DataRecordEntryProvider(EntryProvider):
"""Serve core :class:`~httk.core.DataRecord` values as provider properties.
Definitions are resolved eagerly at construction. Every served property name
must start with ``_``; absent record properties are emitted as JSON null.
:param entries: The data records keyed by their served identifiers.
:param definitions: Optional property definitions keyed by served property name.
:param relationships: Optional related entries keyed by record identifier.
:raises ValueError: If a property name, definition, or non-nullable property
is inconsistent with the supplied records.
"""
def __init__(
self,
entries: Mapping[str, DataRecord | Mapping[str, Any]],
*,
definitions: Mapping[str, PropertyDefinition] | None = None,
relationships: Mapping[str, Iterable[RelatedEntry]] | None = None,
) -> None:
self._entry_type = _records_definition().name
self._entries = {str(key): DataRecord.from_obj(value) for key, value in entries.items()}
self._relationships = _normalized_relationships(relationships)
resolved = dict(definitions or {})
for name in resolved:
if not name.startswith("_"):
raise ValueError(f"served property name {name!r} must start with '_'")
for key, record in self._entries.items():
definition = resolved.get(record.name)
if definition is None:
try:
definition = PropertyDefinition.from_optimade(
record.name, load_property_definition(record.definition_id).as_optimade()
)
except Exception as exc:
raise ValueError(
f"record {key!r} name {record.name!r} has no registered definition "
f"for IRI {record.definition_id!r}"
) from exc
resolved[record.name] = definition
if definition.definition_id and record.definition_id != definition.definition_id:
raise ValueError(
f"record {key!r} name {record.name!r} has definition IRI {record.definition_id!r}, "
f"but resolved definition is {definition.definition_id!r}"
)
for name, definition in resolved.items():
if not definition.nullable:
missing = next((key for key, record in self._entries.items() if record.name != name), None)
if missing is not None:
raise ValueError(
f"served property {name!r} is non-nullable, but record {missing!r} does not populate it"
)
self._definitions = resolved
base = _records_definition()
properties = dict(base.properties)
properties.update(resolved)
# A redefinition (it adds per-record value properties), so it must not
# claim the records document $id: mirror served_form() with a null
# definition_id extending the internal records IRI.
self._definition = EntryTypeDefinition(
base.name, base.description, properties, None, base.extends_id or base.definition_id
)
def __repr__(self) -> str:
return f"DataRecordEntryProvider(entries={len(self._entries)})"
def _check_entry_type(self, entry_type: str) -> None:
if entry_type != self._entry_type:
raise KeyError(f"{type(self).__name__} serves only the '{self._entry_type}' entry type.")
[docs]
def entry_types(self) -> Mapping[str, EntryTypeDefinition]:
"""Return the resolved ``_httk_records`` entry definition.
:return: The served data-record entry-type definition.
"""
return {self._entry_type: self._definition}
[docs]
def property_keys(self, entry_type: str) -> Mapping[str, str]:
"""Return served property names mapped to data-record keys.
:param entry_type: The entry type to inspect.
:return: The served-property to record-key mapping.
:raises KeyError: If ``entry_type`` is not ``_httk_records``.
"""
self._check_entry_type(entry_type)
return {
"id": ID_FIELD,
"type": "type",
"immutable_id": "immutable_id",
"last_modified": "last_modified",
**{name: name for name in self._definitions},
}
[docs]
def records(self, entry_type: str) -> Iterable[Mapping[str, Any]]:
"""Return records with union-null values for unserved properties.
:param entry_type: The entry type to enumerate.
:yield: JSON-compatible records in input mapping order.
:raises KeyError: If ``entry_type`` is not ``_httk_records``.
"""
self._check_entry_type(entry_type)
for entry_id, record in self._entries.items():
yield {
ID_FIELD: entry_id,
"type": self._entry_type,
"immutable_id": record.immutable_id,
"last_modified": _json_value(record.last_modified),
**{name: _json_value(record.value) if name == record.name else None for name in self._definitions},
}
[docs]
def relationships(self, entry_type: str) -> Mapping[str, tuple[RelatedEntry, ...]]:
"""Return normalized data-record relationships by identifier.
:param entry_type: The entry type to inspect.
:return: The relationship mapping supplied at construction.
:raises KeyError: If ``entry_type`` is not ``_httk_records``.
"""
self._check_entry_type(entry_type)
return self._relationships
[docs]
def product_relationships(links: Iterable[ProductLink]) -> dict[str, dict[str, tuple[RelatedEntry, ...]]]:
"""Build source-side relationships for a provider's ``relationships=`` argument.
Feed the inner mapping into the source-side provider's ``relationships=`` argument;
per-edge ``workflow_declaration_uri`` is deliberately not served yet (relation-object
serving is future work).
:param links: The product links to group by source type and identifier.
:return: Source-type mappings of source identifiers to related product entries.
:raises ValueError: If one source has duplicate product labels.
"""
result: dict[str, dict[str, list[RelatedEntry]]] = {}
for link in links:
source = result.setdefault(link.source_type, {}).setdefault(link.source_id, [])
if any(entry.label == link.label for entry in source):
raise ValueError(
f"duplicate product label for source {link.source_type!r}/{link.source_id!r}: {link.label!r}"
)
source.append(
RelatedEntry(
link.target_type,
link.target_id,
role="product",
label=link.label,
relationship=wire_relationship_key("has_product", RUNS_DEFINITION_ID),
)
)
return {
source_type: {source_id: tuple(entries) for source_id, entries in sources.items()}
for source_type, sources in result.items()
}
def _string_column_projection(column: str) -> StoredPropertyProjection:
"""A response/filter/sort projection for a plain nullable-string column read.
:param column: The durable string column read for response, filter, and sort.
:return: The projection reading ``column`` directly.
"""
def query(context: QueryContext, operator: str, literal: object) -> QueryExpression:
if literal is not None and not isinstance(literal, str):
raise QueryLiteralError(f"{column} is a string property; its filter needs a string or null literal")
value = context.field(column)
if operator == "IS_UNKNOWN":
return context.is_null(value)
if operator == "IS_KNOWN":
return context.not_(context.is_null(value))
right = context.null() if literal is None else context.constant(literal)
if operator == "=":
return context.equal(value, right)
if operator == "!=":
return context.not_(context.equal(value, right))
if operator in {"<", "<=", ">", ">=", "CONTAINS", "STARTS", "ENDS"}:
return context.compare(value, operator, right)
raise QueryLiteralError(f"unsupported operator for {column}: {operator}")
return StoredPropertyProjection(
response=lambda record: getattr(record, column),
query=query,
sort=lambda context: context.field(column),
)
# The capability layer owns the serving glue for core records: declare the wire
# (served) property projections for a stored ``Run`` beside its provider. The
# keys are the served names produced by ``EntryTypeDefinition.served_form``; the
# values read the plain internal columns.
Run.__httk_stored_properties__ = { # type: ignore[attr-defined]
"_httk_workflow_declaration_uri": _string_column_projection("workflow_declaration_uri"),
"_httk_source_id": _string_column_projection("source_id"),
}