"""Backend-neutral save-path machinery shared by storage backends."""
import dataclasses
import functools
import re
import types
import typing
import weakref
from collections.abc import Callable, Mapping, Sequence
from typing import Annotated, Any, Protocol, TypeVar, runtime_checkable
from httk.core.storage import IdentitySkip, project_storage_record
from httk.core.storage.identity import _content_id_uncached, _trusted_content_id
from httk.store.backend.schema import FieldSpec, LinkSpec, resolve_schema
from httk.store.storage_layout import EntryFamilyLayout
__all__ = [
"EntryDispatchIntegrityError",
"EntryIdConflictError",
"EntryIdScheme",
"EntryMetadataConflictError",
"EntryReplacementError",
"EntryStore",
"IdentityCaches",
"SaveProjection",
"reject_cursor_proxy",
]
class _LinksAccessor:
"""Lazy ``.links`` namespace on a fetched, store-bound record instance.
Backend-neutral and deliberately dumb: it holds only the source record's
declared :class:`~httk.store.backend.schema.LinkSpec` set and a per-spec
resolver the store supplies (SQL and Mongo differ only in that resolver).
Each named link resolves once and is memoized; the staleness contract
mirrors reference-field memoization — no re-query per access — so a linked
tuple read here reflects the store as of first access, not later.
:param link_specs: The source record class's declared weak-link specs.
:param resolve: A callable turning one spec into its current linked-target tuple.
"""
__slots__ = ("_cache", "_resolve", "_specs")
def __init__(self, link_specs: Sequence[LinkSpec], resolve: Callable[[LinkSpec], tuple[Any, ...]]) -> None:
self._specs = {spec.name: spec for spec in link_specs}
self._resolve = resolve
self._cache: dict[str, tuple[Any, ...]] = {}
def __getattr__(self, name: str) -> tuple[Any, ...]:
if name.startswith("_"):
raise AttributeError(name)
spec = self._specs.get(name)
if spec is None:
declared = ", ".join(sorted(self._specs)) or "none"
raise AttributeError(f"no weak link named {name!r} (declared links: {declared})")
if name not in self._cache:
self._cache[name] = self._resolve(spec)
return self._cache[name]
def __dir__(self) -> list[str]:
return [*super().__dir__(), *self._specs]
@dataclasses.dataclass(frozen=True)
[docs]
class EntryIdScheme:
"""Configuration used to mint human-readable entry identifiers.
:param base: Dot-separated database namespace.
:param series: Campaign-series token.
:param type_in_base: Whether the served entry type is appended to ``base``.
"""
[docs]
type_in_base: bool = False
def __post_init__(self) -> None:
if re.fullmatch(r"[A-Za-z0-9_]+(?:\.[A-Za-z0-9_]+)*", self.base) is None:
raise ValueError("EntryIdScheme.base must be dot-separated alphanumeric/underscore tokens")
if re.fullmatch(r"[A-Za-z0-9_]+", self.series) is None:
raise ValueError("EntryIdScheme.series must be an alphanumeric/underscore token")
[docs]
class EntryIdConflictError(ValueError):
"""An entry id is already owned by a different lineage or alternative group.
:param table_name: The table containing the conflicting identifier.
:param entry_id: The conflicting entry identifier.
:param existing_logical_id: The lineage or group already owning the identifier.
:param requested_logical_id: The lineage or group requesting it, when known.
"""
def __init__(
self,
table_name: str,
entry_id: str,
existing_logical_id: int | None,
requested_logical_id: int | None,
) -> None:
[docs]
self.table_name = table_name
[docs]
self.entry_id = entry_id
[docs]
self.existing_logical_id = existing_logical_id
[docs]
self.requested_logical_id = requested_logical_id
super().__init__(
f"entry id {entry_id!r} in table {table_name!r} belongs to "
f"{existing_logical_id!r}, not {requested_logical_id!r}"
)
_StoredRecord = TypeVar("_StoredRecord")
@runtime_checkable
[docs]
class EntryStore(Protocol):
"""The store surface consumed by :mod:`httk.store.backend.sql.stored_federation`.
This protocol deliberately describes the small backend seam used by the
federation. The stored-property plan and candidate-stream objects remain
backend-specific; their SQL implementations use :meth:`searcher`.
"""
@property
[docs]
def entry_layout(self) -> tuple[EntryFamilyLayout, ...]:
"""Return the configured entry-family layouts in stable order."""
...
[docs]
def searcher(self, *, as_of: object = None) -> Any:
"""Return a backend searcher used to build candidate ID streams."""
...
[docs]
def fetch(self, cls: type[_StoredRecord], sid: int, *, eager: bool = False) -> _StoredRecord:
"""Fetch the stored record of ``cls`` identified by ``sid``.
:param cls: The storable record class.
:param sid: The stored row identifier to fetch.
:param eager: Whether to fully materialize the record instead of returning a lazy row.
:return: The reconstructed instance.
"""
...
[docs]
def fetch_many(self, cls: type[_StoredRecord], sids: Sequence[int], *, eager: bool = False) -> list[_StoredRecord]:
"""Fetch the stored records of ``cls`` identified by ``sids``.
Batched counterpart of :meth:`fetch`.
:param cls: The storable record class.
:param sids: The stored row identifiers to fetch.
:param eager: Whether to fully materialize each record instead of returning lazy rows.
:return: The reconstructed instances in ``sids`` order.
:raises KeyError: When any requested row is absent.
"""
...
[docs]
def stored_property_plan(self, family: type) -> Any:
"""Return the backend-specific stored-property plan for one family.
:param family: The logical entry-family class to plan.
:return: The validated stored-property plan consumed by federation.
"""
...
[docs]
class EntryMetadataConflictError(ValueError):
"""Stored identity-excluded metadata differs from a repeated save."""
[docs]
class EntryReplacementError(ValueError):
"""A replacement deduplicated onto a row from a different lineage.
:param table_name: The table (or collection) whose replacement failed.
:param predecessor_logical_id: The logical_id of the intended predecessor.
:param conflicting_logical_id: The logical_id of the row actually hit.
"""
def __init__(self, table_name: str, predecessor_logical_id: int, conflicting_logical_id: int) -> None:
[docs]
self.table_name = table_name
[docs]
self.predecessor_logical_id = predecessor_logical_id
[docs]
self.conflicting_logical_id = conflicting_logical_id
super().__init__(
f"replacement in table {table_name!r} deduplicated onto an existing row with logical_id "
f"{conflicting_logical_id}, but the predecessor's logical_id is {predecessor_logical_id}"
)
[docs]
class EntryDispatchIntegrityError(RuntimeError):
"""A persisted entry dispatch row does not name exactly its expected backing."""
[docs]
class SaveProjection:
"""One-save projection cache shared by core identity and SQL encoding."""
def __init__(self, *, store_timestamp: int | None = None) -> None:
[docs]
self.store_timestamp = store_timestamp
[docs]
self.values_by_source: dict[tuple[type, int], Mapping[str, object]] = {}
[docs]
self.validated: set[tuple[type, int]] = set()
[docs]
self.metadata_content_ids: dict[tuple[type, int], str] = {}
[docs]
self.active: set[tuple[type, int]] = set()
[docs]
self.inserted: list[tuple[type, int]] = []
[docs]
def projector(self, record_type: type, source: Any) -> Mapping[str, object]:
key = (record_type, id(source))
values = self.values_by_source.get(key)
if values is None:
values = project_storage_record(record_type, source)
self.values_by_source[key] = values
return values
[docs]
def content_id(self, record_type: type, source: Any, *, extras: Mapping[str, object] | None = None) -> str:
# SaveProjection only memoizes the standard deterministic projection;
# use core's trusted route so the source-owned content-id cache is
# shared across saves. Arbitrary custom projectors remain uncached via
# the public content_id path. Extras (alternative-group identity) are
# root-only and must bypass the extras-less trusted cache entirely.
if extras:
return _content_id_uncached(source, as_record=record_type, projector=self.projector, extras=extras)
return _trusted_content_id(source, as_record=record_type, projector=self.projector)
@dataclasses.dataclass(frozen=True)
class _MetadataPlan:
skipped_specs: tuple[FieldSpec, ...]
skipped_nested: tuple[FieldSpec, ...]
descend_specs: tuple[FieldSpec, ...]
_MISSING_METADATA = object()
@functools.cache
def _metadata_reachable_types(record_type: type) -> frozenset[type]:
reachable: set[type] = set()
pending = [record_type]
while pending:
current = pending.pop()
if current in reachable:
continue
reachable.add(current)
pending.extend(
spec.target for spec in resolve_schema(current).fields if not spec.derived and spec.target is not None
)
return frozenset(reachable)
def _metadata_has_plan(record_type: type) -> bool:
for reachable_type in _metadata_reachable_types(record_type):
hints = typing.get_type_hints(reachable_type, include_extras=True)
if any(
not spec.derived and _has_identity_skip(hints.get(spec.field))
for spec in resolve_schema(reachable_type).fields
):
return True
return False
@functools.cache
def _metadata_plan(record_type: type) -> _MetadataPlan | None:
schema = resolve_schema(record_type)
hints = typing.get_type_hints(record_type, include_extras=True)
skipped_specs: list[FieldSpec] = []
skipped_nested: list[FieldSpec] = []
descend_specs: list[FieldSpec] = []
for spec in schema.fields:
if spec.derived:
continue
identity_skipped = _has_identity_skip(hints.get(spec.field))
if identity_skipped:
if spec.role in {"scalar", "encoded", "fixed_array"}:
skipped_specs.append(spec)
else:
skipped_nested.append(spec)
elif spec.target is not None and _metadata_has_plan(spec.target):
descend_specs.append(spec)
if not skipped_specs and not skipped_nested and not descend_specs:
return None
return _MetadataPlan(tuple(skipped_specs), tuple(skipped_nested), tuple(descend_specs))
def _has_identity_skip(annotation: Any) -> bool:
origin = typing.get_origin(annotation)
if origin is Annotated:
arguments = typing.get_args(annotation)
return any(isinstance(marker, IdentitySkip) for marker in arguments[1:]) or _has_identity_skip(arguments[0])
if origin in (typing.Union, types.UnionType):
return any(_has_identity_skip(argument) for argument in typing.get_args(annotation))
return False
[docs]
def reject_cursor_proxy(obj: Any) -> None:
"""Reject a lazy cursor row before a backend attempts to save it."""
if getattr(obj, "__httk_cursor_proxy__", False):
raise TypeError("cursor rows cannot be saved; materialize the record first")
[docs]
class IdentityCaches:
"""Weak identity caches shared by storage backends."""
def __init__(self) -> None:
self._instances: weakref.WeakValueDictionary[tuple[type, int], Any] = weakref.WeakValueDictionary()
self._sids: weakref.WeakKeyDictionary[Any, dict[type, int]] = weakref.WeakKeyDictionary()
self._sids_by_identity: dict[tuple[type, int], int] = {}
"""Reverse cache for instances that cannot be hashed (e.g. they hold a list).
Keyed on ``id()``, with a finalizer dropping each entry when its
instance dies, so a recycled id can never resolve to a stale sid.
"""
def _clear_identity_caches(self) -> None:
self._instances.clear()
self._sids.clear()
self._sids_by_identity.clear()
def _remember(self, cls: type, sid: int, obj: Any, *, cache_instance: bool = True) -> None:
if cache_instance:
try:
self._instances[(cls, sid)] = obj
except TypeError:
return # Not weak-referenceable; identity caching is best-effort.
try:
sids = self._sids.setdefault(obj, {})
sids[cls] = sid
except TypeError:
# Unhashable (a storable class holding a list field is): key the
# reverse cache on identity instead, dropping the entry when the
# instance dies. Without this, sid_of() — and so referring() —
# would report a just-saved instance as never stored.
key = (cls, id(obj))
try:
weakref.finalize(obj, self._sids_by_identity.pop, key, None)
except TypeError:
return # Tuples and other non-weakrefable sources use database lookup.
self._sids_by_identity[key] = sid