"""Backend-neutral save-path machinery shared by storage backends."""
import functools
import types
import typing
import weakref
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Annotated, Any, Protocol, TypeVar, runtime_checkable
from httk.core.storage import IdentitySkip, project_storage_record
from httk.core.storage.identity import _trusted_content_id
from httk.store.db.schema import FieldSpec, resolve_schema
__all__ = [
"EntryDispatchIntegrityError",
"EntryMetadataConflictError",
"EntryStore",
"IdentityCaches",
"SaveProjection",
"reject_cursor_proxy",
]
_StoredRecord = TypeVar("_StoredRecord")
@runtime_checkable
[docs]
class EntryStore(Protocol):
"""The store surface consumed by :mod:`httk.store.db.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[Any, ...]:
"""Return the configured entry-family layouts in stable order."""
...
[docs]
def searcher(self) -> Any:
"""Return a backend searcher used to build candidate ID streams."""
...
[docs]
def fetch(self, cls: type[_StoredRecord], sid: int) -> _StoredRecord:
"""Fetch the stored record of ``cls`` identified by ``sid``."""
...
[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 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) -> None:
[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) -> 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.
return _trusted_content_id(source, as_record=record_type, projector=self.projector)
@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