"""Build an OPTIMADE adapter over lazy SQL-backed entry federation.
The durable layout, SQL translation, collision policy, and bounded global
pagination live in ``httk.store.backend.sql``. This module owns only the serving
boundary: the advertised OPTIMADE schema, request-error translation, public
response-field projection, and :class:`~httk.serve.optimade.model.ResultRow`
objects consumed by the endpoint envelope code.
Imports from ``httk.store.backend.sql`` deliberately remain inside call sites. Importing
``httk.serve.optimade`` therefore does not initialize a database backend or
load optional SQL dialects.
"""
from collections.abc import Iterator, Mapping, Sequence
from dataclasses import dataclass
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, NoReturn, cast
from httk.core import ALTERNATIVE_KIND_PATTERN, EntryTypeDefinition, RelatedEntry, load_entry_type_definition
from httk.core.optimade import FilterAst
from httk.core.storage import stored_property_projections
from httk.store import EntryStore, FilterTranslationError
from ..model.errors import OptimadeError, TranslatorError, translator_error_from
from ..model.results import QueryFunction, QueryResults, ResultRow
from ..schema.served import ServedSchema, build_served_schema
if TYPE_CHECKING:
from httk.store.backend.sql import StoredEntrySource
@dataclass(frozen=True, slots=True)
class _StoredQueryResults:
"""One already-bounded data-owned page exposed through QueryResults."""
rows: tuple[ResultRow, ...]
more_data_available: bool
total_count: int
def count(self) -> int:
return self.total_count
def __iter__(self) -> Iterator[ResultRow]:
return iter(self.rows)
@dataclass(frozen=True, slots=True)
[docs]
class StoredBackendAdapter:
"""Serve one data federation per OPTIMADE entry type.
:param federations: Durable federations keyed by entry endpoint.
:param schema: Schema describing the federations' served entries.
"""
[docs]
federations: Mapping[str, Any]
def __post_init__(self) -> None:
object.__setattr__(self, "federations", MappingProxyType(dict(self.federations)))
[docs]
def snapshot_cutoff_ns(self, entry_type: str, now_ns: int) -> int | None:
"""Return the resolution-aware snapshot cutoff for one entry type."""
return self.federations[entry_type].snapshot_cutoff_ns(now_ns)
[docs]
def query_function(self) -> QueryFunction:
"""Return the callback that queries the configured federations.
:return: Query callback consumed by the OPTIMADE request engine.
"""
def query(
entries: list[str],
response_fields: list[str],
unknown_response_fields: list[str],
page_limit: int,
page_offset: int,
filter_ast: FilterAst | None = None,
*,
as_of: int | None = None,
sort: Sequence[tuple[str, bool]] | None = None,
revisions: bool = False,
alternatives: bool = False,
immutable_id: str | None = None,
debug: bool = False,
) -> QueryResults:
del debug
if len(entries) != 1 or entries[0] not in self.federations:
raise TranslatorError(
"Stored OPTIMADE queries must target exactly one configured entry type.",
500,
"Internal server error",
)
entry_type = entries[0]
federation = self.federations[entry_type]
limit = int(page_limit)
offset = int(page_offset)
try:
public_id = _exact_id_filter(filter_ast)
if alternatives:
if immutable_id is not None:
# immutable_id carries the composite ``<id>~<kind>``; the id part
# may bear a source prefix or be a non-conforming URL-safe id, so
# split off only the kind and pass the id through untouched. A
# per-group lineage id filter, when present, is that same id.
entry_id, _sep, kind = immutable_id.rpartition("~")
if not entry_id or ALTERNATIVE_KIND_PATTERN.fullmatch(kind) is None:
found = None
else:
group_id = public_id if public_id is not None else entry_id
found = federation.fetch_alternative(group_id, kind, as_of=as_of, fields=response_fields)
page_rows, page_relationships, total_count = _single(found, offset, limit)
more_data_available = False
else:
page = federation.query(
filter_ast,
sort=tuple(sort or ()),
offset=offset,
limit=limit,
as_of=as_of,
fields=response_fields,
alternatives=True,
)
page_rows = page.rows
page_relationships = page.relationships
more_data_available = page.more_data_available
total_count = page.total_count
elif immutable_id is not None:
if public_id is None:
found = federation.fetch(immutable_id, as_of=as_of, fields=response_fields, revisions=True)
else:
found = federation.fetch_revision(public_id, immutable_id, as_of=as_of, fields=response_fields)
page_rows, page_relationships, total_count = _single(found, offset, limit)
more_data_available = False
elif public_id is not None and offset == 0 and limit > 0:
found = federation.fetch(public_id, as_of=as_of, fields=response_fields, revisions=revisions)
page_rows, page_relationships, total_count = _single(found, offset, limit)
more_data_available = False
else:
page = federation.query(
filter_ast,
sort=tuple(sort or ()),
offset=offset,
limit=limit,
as_of=as_of,
fields=response_fields,
revisions=revisions,
)
page_rows = page.rows
page_relationships = page.relationships
more_data_available = page.more_data_available
total_count = page.total_count
except FilterTranslationError as error:
raise translator_error_from(error) from error
except Exception as error:
_raise_stored_error(error)
projected = tuple(
ResultRow(
values=_project_public_row(
row,
entry_type,
response_fields,
unknown_response_fields,
),
relationships=_relationships_block(related),
)
for row, related in zip(page_rows, page_relationships, strict=True)
)
return _StoredQueryResults(projected, bool(more_data_available), int(total_count))
return query
def _single(
found: tuple[Mapping[str, Any], Mapping[str, tuple[RelatedEntry, ...]]] | None,
offset: int,
limit: int,
) -> tuple[tuple[Mapping[str, Any], ...], tuple[Mapping[str, tuple[RelatedEntry, ...]], ...], int]:
"""Shape one optional ``(row, relationships)`` fetch into row/relationship tuples.
:param found: The fetched ``(row, relationships)`` pair, or ``None`` when absent.
:param offset: The requested page offset (a positive offset skips the single row).
:param limit: The requested page limit (a zero limit yields no rows).
:return: Row-aligned ``(rows, relationships, total_count)``.
"""
if found is None:
return (), (), 0
if offset or limit == 0:
return (), (), 1
row, related = found
return (row,), (related,), 1
def _relationships_block(
related: Mapping[str, tuple[RelatedEntry, ...]],
) -> dict[str, list[dict[str, Any]]]:
"""Render a stored row's weak-link relationships as OPTIMADE identifiers.
Mirrors the provider-path extractor (``backend/providers.py``): each
:class:`~httk.core.RelatedEntry` becomes an ``{"type": ..., "id": ...,
"description"?: ..., "role"?: ..., "label"?: ...}`` identifier (``type`` being
the identifier's own target entry type), grouped under
``relationship or related_type`` (the served semantic relationship key,
falling back to the mapping key). An empty mapping renders as an empty block.
:param related: Weak-link relationships grouped by related entry type.
:return: Relationship identifiers keyed by served relationship key.
"""
block: dict[str, list[dict[str, Any]]] = {}
for related_type, entries in related.items():
for entry in entries:
identifier: dict[str, Any] = {"type": entry.entry_type, "id": entry.id}
if entry.description:
identifier["description"] = entry.description
if entry.role:
identifier["role"] = entry.role
if entry.label is not None:
identifier["label"] = entry.label
block.setdefault(entry.relationship or related_type, []).append(identifier)
return block
def _exact_id_filter(filter_ast: FilterAst | None) -> str | None:
"""Return the id from the canonical exact-id AST used by single fetches."""
if (
filter_ast is not None
and len(filter_ast) == 3
and filter_ast[0] == "="
and filter_ast[1] == ("Identifier", "id")
and isinstance(filter_ast[2], tuple)
and len(filter_ast[2]) == 2
and filter_ast[2][0] == "String"
and isinstance(filter_ast[2][1], str)
):
return cast(str, filter_ast[2][1])
return None
def _project_public_row(
row: Mapping[str, Any],
entry_type: str,
response_fields: Sequence[str],
unknown_response_fields: Sequence[str],
) -> dict[str, Any]:
"""Select only protocol-requested fields from one public federation row."""
try:
public_id = row["id"]
row_type = row["type"]
except KeyError as error:
raise OptimadeError(
"Stored entry federation returned a row without public id/type.",
500,
"Internal server error",
) from error
if not isinstance(public_id, str) or not public_id:
raise OptimadeError(
"Stored entry federation returned an invalid public entry id.",
500,
"Internal server error",
)
if row_type != entry_type:
raise OptimadeError(
"Stored entry federation returned an entry under the wrong endpoint type.",
500,
"Internal server error",
)
result: dict[str, Any] = {name: None for name in unknown_response_fields}
result.update({name: row.get(name) for name in response_fields})
# id/type are required response fields, but retain the invariant even for a
# direct query-function caller that supplies a narrower field list.
result["id"] = public_id
result["type"] = row_type
return result
def _raise_stored_error(error: Exception) -> NoReturn:
"""Translate data-owned federation failures without leaking SQL details."""
from httk.store.backend.sql import DuplicateEntryIdError
if isinstance(error, DuplicateEntryIdError):
public_id = getattr(error, "public_id", None)
origins = getattr(error, "origins", ())
origin_names = tuple(
f"{origin.source}/{origin.backing}"
for origin in origins
if isinstance(getattr(origin, "source", None), str) and isinstance(getattr(origin, "backing", None), str)
)
detail = "Duplicate public entry id"
if isinstance(public_id, str):
detail += f" {public_id!r}"
if origin_names:
detail += " was found in " + ", ".join(origin_names)
detail += "; run audit_duplicate_ids() on the stored federation."
raise OptimadeError(detail, 500, "Internal server error") from error
raise error
def _validate_sortable_backings(
plans: Sequence[Any],
entry_type: str,
sortable: Sequence[str],
) -> None:
"""Fail adapter construction when an advertised sort cannot be exact."""
for name in sortable:
if name in {"id", "type", "immutable_id", "_httk_id", "_httk_kind"}:
continue
for plan in plans:
for backing in plan.backings:
projection = stored_property_projections(backing).get(name)
if projection is None or projection.sort is None:
raise ValueError(
f"Property {name!r} is marked sortable for entry type {entry_type!r}, "
f"but {backing.__name__} has no exact stored sort mapping."
)
def _sortable_intersection(plans: Sequence[Any], property_names: Sequence[str]) -> tuple[str, ...]:
"""Return properties with an exact sort mapping on every durable backing."""
sortable: list[str] = []
for name in property_names:
if name in {"id", "type", "immutable_id", "_httk_id", "_httk_kind"}:
sortable.append(name)
continue
if all(
(projection := stored_property_projections(backing).get(name)) is not None and projection.sort is not None
for plan in plans
for backing in plan.backings
):
sortable.append(name)
return tuple(sortable)
[docs]
def adapter_from_stores(
sources: Sequence["StoredEntrySource"],
**options: Any,
) -> StoredBackendAdapter:
"""Build a lazy store-backed adapter from durable entry sources.
Sources with the same exact logical family are federated under one entry
endpoint. The data layer owns all source/backing traversal and global
pagination; this adapter advertises the family's definition and turns only
the returned page into OPTIMADE result rows.
:param sources: Durable entry sources to federate by entry type.
:param \\*\\*options: Schema options forwarded to :func:`~httk.serve.optimade.schema.served.build_served_schema`,
e.g. ``default_includes`` (per served entry type, the served entry
types to include by default on single-entry requests; always unioned
with ``references``).
:return: Lazy adapter over the supplied durable sources.
:raises ValueError: If sources conflict or expose incomplete sort mappings.
:raises TypeError: If a source is not a stored entry source.
"""
from httk.store.backend.sql import (
StoredEntryFederation,
StoredEntrySource,
related_property_resolver_factory,
stored_property_sql_plan,
)
values = tuple(sources)
if not values:
raise ValueError("adapter_from_stores requires at least one StoredEntrySource")
if not all(isinstance(source, StoredEntrySource) for source in values):
raise TypeError("adapter_from_stores sources must contain StoredEntrySource values")
grouped: dict[str, list[Any]] = {}
families: dict[str, type] = {}
definitions: dict[str, EntryTypeDefinition] = {}
plans_by_entry: dict[str, list[Any]] = {}
served_type_names: dict[str, str] = {}
for source in values:
# Resolve the internal (bare) definition exactly as the plan does, then
# serve its wire form: the served definition drives the plan's entry_type
# (now the WIRE name) and every property name, projection, filter, and
# sort. A prefixed family MUST be planned with served= set.
factory = getattr(source.entry_family, "entry_type_definition", None)
internal: EntryTypeDefinition = cast(
EntryTypeDefinition,
factory() if callable(factory) else load_entry_type_definition(source.entry_family.definition_id),
)
plan = stored_property_sql_plan(source.store, source.entry_family, served=internal.served_form())
entry_type = plan.entry_type
served_type_names[internal.name] = entry_type
existing_family = families.get(entry_type)
if existing_family is not None and existing_family is not source.entry_family:
raise ValueError(f"entry type {entry_type!r} is supplied by more than one logical entry family")
existing_definition = definitions.get(entry_type)
if existing_definition is not None and existing_definition != plan.definition:
raise ValueError(f"stored sources for entry type {entry_type!r} use different definitions")
families[entry_type] = source.entry_family
definitions[entry_type] = plan.definition
grouped.setdefault(entry_type, []).append(source)
plans_by_entry.setdefault(entry_type, []).append(plan)
served = {entry_type: tuple(definition.properties) for entry_type, definition in definitions.items()}
defaults = {
entry_type: tuple(name for name in property_names if name not in {"id", "type"})
for entry_type, property_names in served.items()
}
if "sortable" not in options:
options["sortable"] = {
entry_type: _sortable_intersection(plans_by_entry[entry_type], property_names)
for entry_type, property_names in served.items()
}
schema = build_served_schema(
definitions,
served,
default_response_overrides=defaults,
revisions=tuple(definitions),
alternatives=tuple(definitions),
**options,
)
for entry_type, plans in plans_by_entry.items():
_validate_sortable_backings(plans, entry_type, schema.sortable_response_fields[entry_type])
# Depth-1 related-property filtering (e.g. ``references.doi CONTAINS ...``)
# resolves the sibling family's own ids per backing store; the factory is
# built over every grouped plan so a filter reaches whichever served family
# the dotted type names, restricted to the filtered row's own store.
resolver_factory = related_property_resolver_factory([plan for plans in plans_by_entry.values() for plan in plans])
federations = {
entry_type: StoredEntryFederation(
tuple(entry_sources),
source_inventory=values,
served_type_names=served_type_names,
related_resolver_factory=resolver_factory,
)
for entry_type, entry_sources in grouped.items()
}
return StoredBackendAdapter(federations, schema)
[docs]
def adapter_from_store(store: EntryStore, **options: Any) -> StoredBackendAdapter:
"""Build a lazy OPTIMADE adapter from every described family in one store.
Families declared without an entry-type definition are deliberately
ignored. This lets application-specific records, such as DSP publication
declarations, coexist with OPTIMADE records in one durable layout.
:param store: Entry store whose configured layout is discovered.
:param \\*\\*options: Schema options forwarded to :func:`adapter_from_stores`.
:return: Lazy adapter over all configured OPTIMADE families.
:raises TypeError: If ``store`` does not implement :class:`EntryStore`.
:raises ValueError: If the store contains no OPTIMADE-described family.
"""
from httk.store.backend.sql import StoredEntrySource
if not isinstance(store, EntryStore):
raise TypeError("adapter_from_store requires an EntryStore")
sources = tuple(
StoredEntrySource(store, layout.family, layout.name)
for layout in store.entry_layout
if layout.definition_id is not None
)
if not sources:
raise ValueError("store has no configured entry family with an OPTIMADE definition")
return adapter_from_stores(sources, **options)