#
# The high-throughput toolkit (httk)
# Copyright (C) 2012-2024 the httk AUTHORS
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""Stdlib-only marker vocabulary for declaring storable record classes.
A *storable* class is a plain frozen dataclass whose fields a storage layer
(such as the database layer in *httk-data*) can resolve into a relational
schema. Storability is non-intrusive: there is no base class to inherit.
This module holds only the declaration vocabulary — the markers attached to
fields via :class:`typing.Annotated`, the optional class-level
:class:`StorageInfo` declaration, and the :class:`stored_property` decorator
for derived, queryable properties — so that any httk module (or application)
can declare storable classes while depending only on httk-core. All schema
resolution and storage work happens in the storage layer.
Markers are used like this::
@dataclass(frozen=True)
class StructureRecord:
__httk_storage__: ClassVar[StorageInfo] = StorageInfo(indexes=(("spacegroup", "formula"),))
formula: Annotated[str, Indexed()]
spacegroup: int
cell_basis: Annotated[FracVector, Shape(3, 3)]
symbols: list[str]
@stored_property
def natoms(self) -> int:
return len(self.symbols)
The class-level declaration is optional; a storage layer must accept plain
frozen dataclasses with no markers at all, and may also accept an external
:class:`StorageInfo` override for classes that cannot be modified.
"""
from dataclasses import dataclass
from typing import Final, Literal
__all__ = [
"STORAGE_INFO_ATTRIBUTE",
"DedupPolicy",
"Indexed",
"Related",
"RelationshipLink",
"Shape",
"Skip",
"StorageInfo",
"Unique",
"stored_property",
]
[docs]
STORAGE_INFO_ATTRIBUTE: Final = "__httk_storage__"
"""Class attribute name where a storable class may attach its :class:`StorageInfo`."""
[docs]
type DedupPolicy = Literal["content_id", "by_value", "none"]
"""How a storage layer deduplicates saved instances of a class.
- ``"content_id"``: reuse an existing row whose stored content identity matches
(the default; suited to immutable value objects).
- ``"by_value"``: reuse an existing row whose stored columns all match (suited
to join-objects such as tags and references, whose identity is their value).
- ``"none"``: always insert a new row.
"""
_DEDUP_POLICIES: Final = ("content_id", "by_value", "none")
@dataclass(frozen=True)
[docs]
class Indexed:
"""Field marker: request a single-column index on this field's column(s)."""
@dataclass(frozen=True)
[docs]
class Unique:
"""Field marker: request a unique index on this field's column(s)."""
@dataclass(frozen=True)
[docs]
class Skip:
"""Field marker: the field exists on the dataclass but is not stored."""
@dataclass(frozen=True)
[docs]
class Shape:
"""Field marker: fixed or variable shape for a vector-valued field.
``rows >= 1`` declares a fixed-shape value stored inline (flattened
row-major into columns). ``rows == 0`` declares a variable number of rows
with ``cols`` fixed columns each, stored out-of-line (one row per entry,
in insertion order).
Args:
rows: Number of rows; ``0`` means variable-length.
cols: Number of columns per row; must be at least ``1``.
"""
def __post_init__(self) -> None:
if self.rows < 0:
raise ValueError(f"Shape rows must be >= 0 (0 means variable-length), got {self.rows}")
if self.cols < 1:
raise ValueError(f"Shape cols must be >= 1, got {self.cols}")
@dataclass(frozen=True)
@dataclass(frozen=True)
[docs]
class RelationshipLink:
"""Class-level relationship declaration: each stored row expresses one FROM→TO relationship.
Declared in :attr:`StorageInfo.links`. ``source`` and ``target`` each name
a reference field of the declaring class, or are ``None`` to mean *the
declaring class's own entry*: for every stored row, one relationship is
declared from the entry the source side resolves to, to the entry the
target side resolves to. The two canonical shapes:
- **join-object**: ``RelationshipLink("structure", "reference")`` on a
``StructureRef`` join class — every ``StructureRef`` row relates its
``structure`` to its ``reference`` (structures → references), without the
join class itself being served.
- **field-inverse**: ``RelationshipLink("structure", None, role="output")``
on a ``Calculation`` class — every ``Calculation`` row relates its
``structure`` to the calculation itself (structures → calculations),
i.e. the inverse of the ``structure`` reference field.
``role`` and ``description`` carry the same OPTIMADE per-identifier
metadata as :class:`Related` into each relationship the link declares.
Args:
source: The reference field naming the FROM-side entry, or ``None``
for the declaring class's own entry.
target: The reference field naming the TO-side entry, or ``None``
for the declaring class's own entry.
role: The machine-readable relationship role, if any.
description: The human-readable relationship description, if any.
Raises:
ValueError: If both endpoints are ``None`` (the link would relate every
entry to itself), or if ``source`` and ``target`` name the same
field.
"""
[docs]
role: str | None = None
[docs]
description: str | None = None
def __post_init__(self) -> None:
if self.source is None and self.target is None:
raise ValueError(
"RelationshipLink needs at least one named field endpoint (source and target are both None)"
)
if self.source is not None and self.source == self.target:
raise ValueError(f"RelationshipLink source and target must differ, both are {self.source!r}")
@dataclass(frozen=True)
[docs]
class StorageInfo:
"""Optional class-level storage declaration for a storable dataclass.
Attach as the class attribute named by :data:`STORAGE_INFO_ATTRIBUTE`
(``__httk_storage__``), annotated ``ClassVar[StorageInfo]`` so dataclass
processing ignores it. A storage layer may also accept an instance as an
external override for classes that cannot be modified.
Args:
storage_name: The name this class is stored under; ``None`` derives one from
the class name. A relational backend uses it as the table name, a document
store as the collection name.
indexes: Composite indexes, each a tuple of field names.
dedup: Deduplication policy applied when saving; see :data:`DedupPolicy`.
links: Class-level relationship declarations; see :class:`RelationshipLink`.
"""
[docs]
storage_name: str | None = None
[docs]
indexes: tuple[tuple[str, ...], ...] = ()
[docs]
dedup: DedupPolicy = "content_id"
[docs]
links: tuple[RelationshipLink, ...] = ()
def __post_init__(self) -> None:
if self.dedup not in _DEDUP_POLICIES:
raise ValueError(f"StorageInfo dedup must be one of {_DEDUP_POLICIES}, got {self.dedup!r}")
for index in self.indexes:
if not index:
raise ValueError("StorageInfo indexes must not contain empty field-name tuples")
[docs]
class stored_property(property):
"""A derived property that a storage layer stores and makes queryable.
Use exactly like :class:`property` (getter only). The value type is read
from the getter's return annotation. On save, the storage layer evaluates
and stores the value alongside the declared fields; on load, the value is
recomputed by the property rather than passed to ``__init__``.
"""