"""The SQL store: save and fetch storable frozen dataclasses through a :class:`~httk.store.backend.sql.engine.Backend`.
:class:`~httk.store.backend.sql.store.SqlStore` is the object-level storage API on top of the schema IR
(:mod:`httk.store.backend.schema`), the value codecs (:mod:`httk.store.backend.codecs`),
the content identity (:mod:`httk.core.storage`), and the SQLAlchemy table
mapping (:mod:`httk.store.backend.sql.mapping`):
- :meth:`SqlStore.save` writes an instance (recursing into referenced and
child-element storables) and returns its integer ``sid``, deduplicating per
the class's :attr:`~httk.core.storage.StorageInfo.dedup` policy;
- :meth:`SqlStore.fetch` reconstructs the instance stored under a ``sid`` —
exactly, via the ``*_exact`` companion columns for rationals — as a lazy row
by default (fields decode on first access) or, with ``eager=True``, fully
materialized; repeated live default fetches of one sid return the same
object, with a materialized instance taking precedence over a proxy;
- :meth:`SqlStore.transaction` scopes several operations into one database
transaction (commit on exit, roll back on exception); outside of it every
operation autocommits;
- :meth:`SqlStore.referring` finds join-objects (tags, references) pointing at
a stored instance, replacing v1's implicit codependent-data machinery;
- :meth:`SqlStore.searcher` starts a query through the search DSL
(:mod:`httk.store.backend.sql.searcher`), implementing the :mod:`httk.store.query`
protocols.
Deduplication semantics (ported from v1): under ``"content_id"`` an equal
instance maps to the existing row (children are not re-inserted); under
``"by_value"`` a row matching **all parent-table columns** is reused — child
table contents are *not* part of the match, mirroring v1 which matched key
columns only; under ``"none"`` every save inserts a new row.
Identity caches are best-effort; content-addressed :meth:`SqlStore.sid_of`
lookups fall back to the database.
"""
import contextlib
import datetime
import json
import threading
import time
import typing
import uuid
from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
from types import TracebackType
from typing import TYPE_CHECKING, Any, Literal, NoReturn, Self, cast
import sqlalchemy
from httk.core import (
EntryTypeDefinition,
FracVector,
load_entry_type_definition,
)
from httk.core.entry_ids import (
ALTERNATIVE_KIND_PATTERN,
check_entry_id,
check_immutable_id,
format_alternative_id,
format_entry_id,
format_immutable_id,
)
from httk.core.storage import (
Shape,
StorageProjectionCycleError,
resolve_storage_record,
)
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from httk.store.backend.codecs import (
codec_named,
decode_fracvector_exact,
encode_fracvector_exact,
encode_fracvector_floats,
)
from httk.store.backend.schema import FieldSpec, LinkSpec, SchemaError, TableSchema, resolve_schema
from httk.store.backend.sql.engine import Backend, connection_uses_autocommit
from httk.store.backend.sql.graph import LogicalEdgeGraph
from httk.store.backend.sql.layout import (
METADATA_TABLE_NAME,
STORAGE_PROTOCOL_VERSION,
BackendFacts,
EntryFamilyLayout,
StorageLayout,
StorageLayoutUpgradeRequiredError,
StoreUnderConstructionError,
actual_columns,
actual_schema_objects,
actual_table_names,
backend_facts_for_dialect,
declaration_json,
expected_metadata,
metadata_table_for,
normalize_entry_declaration,
read_store_metadata,
)
from httk.store.backend.sql.mapping import (
ALT_ID_COLUMN,
ALT_KIND_COLUMN,
CONTENT_ID_COLUMN,
DISPATCH_CONTENT_ID_COLUMN,
ENTRY_ID_OWNERS_TABLE_NAME,
IMMUTABLE_ID_OWNERS_TABLE_NAME,
LOGICAL_ID_COLUMN,
RETRACTED_COLUMN,
ROLE_COLUMN,
SID_COLUMN,
SOURCE_LID_COLUMN,
STORE_TIMESTAMP_COLUMN,
TARGET_LID_COLUMN,
added_column_ddl,
backing_dispatch_column_name,
dispatch_table_for,
entry_dispatch_table_name,
identity_owner_tables,
table_for,
)
from httk.store.backend.sql.rows import RowHydrator, StaleResultError, decode_field, is_lazy_row, lazy_row_identity
from httk.store.backend.sql.searcher import SqlSearcher
from httk.store.storage_layout import (
ADDITIVE_UPGRADE_HINT,
AdditiveUpgradePlan,
EntryFamilyDeclaration,
EntryLayoutBindingError,
classify_schema_upgrade,
schema_fingerprint_diff,
schema_fingerprint_json,
)
from httk.store.store_common import (
_MISSING_METADATA,
EntryDispatchIntegrityError,
EntryIdConflictError,
EntryIdScheme,
EntryMetadataConflictError,
EntryReplacementError,
IdentityCaches,
SaveProjection,
_metadata_plan,
_MetadataPlan,
reject_cursor_proxy,
)
from httk.store.store_timestamp import (
StoreClockRegressionError,
advance_store_timestamp_mark,
capture_store_timestamp,
encode_store_timestamp_state,
ns_operand_to_store_units,
parse_store_timestamp_state,
)
if TYPE_CHECKING:
from httk.store.backend.sql.bulk import BulkIngest
_Projection = SaveProjection
_IDENTITY_OWNERSHIP_KEY = "identity_ownership"
_IDENTITY_OWNERSHIP_VERSION = "1"
# A sid resolver assigns (by saving recursively, or by an in-memory allocator)
# an integer sid to a referenced record: ``(record_type, source, path) -> sid``.
type SidResolver = Callable[[type, Any, str], int]
__all__ = [
"EntryDispatchIntegrityError",
"EntryIdConflictError",
"EntryMetadataConflictError",
"EntryReplacementError",
"SqlStore",
"StoreClockRegressionError",
]
class _DegradedWriteCrash(BaseException):
"""Deterministic test-only hard-stop emitted after one degraded write step."""
def __init__(self, point: str) -> None:
self.point = point
super().__init__(f"injected degraded hard crash after {point}")
class _TransactionToken:
"""Marks whether the transaction that produced a batch of lazy rows rolled back.
A :class:`~httk.store.backend.sql.rows._Chunk` records the current token at birth (and
per deferred child read); the outermost :meth:`SqlStore._transaction_scope`
sets ``rolled_back`` on failure, so accessing a lazy row built inside that
transaction raises :class:`~httk.store.backend.sql.rows.ExpiredLazyRecordError`.
"""
__slots__ = ("rolled_back",)
def __init__(self) -> None:
self.rolled_back = False
def _schema_object_type(kinds: frozenset[str]) -> object:
return next(iter(kinds)) if len(kinds) == 1 else tuple(sorted(kinds))
def _served_definition(family: type) -> EntryTypeDefinition | None:
"""Return the wire (served) form of a family's internal definition, if resolvable.
Resolved exactly as :func:`stored_property_sql_plan` resolves its internal
definition (family ``entry_type_definition()`` or, failing that,
``load_entry_type_definition(definition_id)``). Returns ``None`` when the
family declares neither, leaving the plan builder to raise its own error.
:param family: The logical entry-family class to resolve.
:return: The served definition, or ``None`` when it cannot be resolved.
"""
factory = getattr(family, "entry_type_definition", None)
if callable(factory):
internal = factory()
else:
definition_id = getattr(family, "definition_id", None)
if not isinstance(definition_id, str) or not definition_id:
return None
internal = load_entry_type_definition(definition_id)
return internal.served_form() if isinstance(internal, EntryTypeDefinition) else None
[docs]
class SqlStore:
"""Object storage for storable frozen dataclasses in a relational :class:`~httk.store.backend.sql.engine.Backend`.
A store starts with an explicit, versioned entry declaration. Ordinary
unconfigured frozen-dataclass tables remain on-demand, but only after the
layout marker has been initialized on a physically empty database. Schemas
edited out-of-band fail at use time with the database's own errors.
The first open of a database requires ``entry_records`` or
``entry_families``. The store stamps
the canonical JSON declaration and protocol version, then trusts that
declaration on reopen: a supplied declaration must be byte-identical, and
mismatches raise :class:`~httk.store.backend.sql.layout.StorageLayoutUpgradeRequiredError`.
Reopening does not diff or migrate record schemas. Read paths never issue
DDL; missing ordinary tables behave as empty results or missing rows, while
table creation happens only through writes or :meth:`ensure_tables`.
:param database: The database used for storage.
:param entry_records: The required entry-family declaration when first opening a database.
:param entry_families: Application-owned declarations which bypass global registration.
:param records: Decorated application entry records; mutually exclusive with the explicit declarations.
:param entry_ids: Optional scheme used to mint ids for defined entry families.
:param store_timestamps: Whether parent rows carry store-managed timestamps.
:param store_timestamp_resolution: Nanoseconds represented by one stored unit.
:param allow_clock_regression: Whether to disable the process-local clock guard.
:param clock_regression_grace: Whether to wait briefly for sub-millisecond regressions.
:param upgrade: Whether to apply a purely additive schema-fingerprint change
(new nullable columns, new lazily created tables) on reopen instead of
raising; non-additive or non-schema differences still raise.
:raises TypeError: If the first open omits both declaration forms.
:raises httk.store.backend.sql.layout.StorageLayoutUpgradeRequiredError: If the trusted declaration or protocol does not match.
"""
# Subclasses which operate exclusively on new, offline stores may select a
# different default without changing every call site. ``"auto"`` remains
# the normal public default and always falls back to the legacy protocol
# when a store is not physically empty.
[docs]
bulk_ingest_finalize_default: Literal["auto", "parity", "deferred"] = "auto"
# Set true by the per-dialect subclasses in ``stores`` (which build and own
# the Backend), so that :meth:`~httk.store.backend.sql.store.SqlStore.close` disposes it. A store handed a Backend
# by the caller leaves this false and never disposes what it does not own.
_owns_database: bool = False
def __init__(
self,
database: Backend,
*,
entry_records: Mapping[type, type | tuple[type, ...]] | None = None,
entry_families: Sequence[EntryFamilyDeclaration] | None = None,
records: Sequence[type] | None = None,
entry_ids: EntryIdScheme | None = None,
store_timestamps: bool = True,
store_timestamp_resolution: int = 1000,
allow_clock_regression: bool = False,
clock_regression_grace: bool = True,
upgrade: bool = False,
) -> None:
if (
not isinstance(store_timestamp_resolution, int)
or isinstance(store_timestamp_resolution, bool)
or store_timestamp_resolution <= 0
):
raise ValueError("store_timestamp_resolution must be a positive integer")
self._owns_database = False
self._database = database
self._entry_ids = entry_ids
self._upgrade = upgrade
self._identity_ownership_ready = True
self._store_timestamps = store_timestamps
self._store_timestamp_resolution = store_timestamp_resolution
self._allow_clock_regression = allow_clock_regression
self._clock_regression_grace = clock_regression_grace
self._clock = time.time_ns
self._store_timestamp_mark: int | None = None
self._backend_facts: BackendFacts | None = None
self._metadata = sqlalchemy.MetaData()
self._layout: StorageLayout | None = None
self._managed_table_names: frozenset[str] = frozenset()
self._known_record_types: set[type] = set()
self._tables_present: set[str] = set()
self._candidate_names: dict[frozenset[type], frozenset[str]] = {}
self._initialized = False
self._initialization_ddl_journal: list[sqlalchemy.Table] = []
self._identity = IdentityCaches()
self._local = threading.local()
self._bulk_active = False
self._bulk_state_lock = threading.Lock()
self._write_profile: Literal["transactional", "degraded", "bulk-fenced"] = database.write_profile
self._lease_owner = uuid.uuid4().hex
self._lease_value: str | None = None
self._mutation_lock = threading.RLock()
self._lease_callback_registered = False
self._lease_lifecycle_generation: int | None = None
# A deterministic test seam. Production instances leave this unset;
# returning true simulates a process death *after* the named durable
# write, deliberately preserving any dirty marker.
self._degraded_fault_hook: Callable[[str], bool] | None = None
if self._write_profile in {"degraded", "bulk-fenced"}:
# This fence is deliberately registered before layout creation:
# initialization can create metadata, and a dispose interleaving
# must never leave a later write able to acquire an unowned lease.
self._register_degraded_lifecycle_fence()
if records is not None and (entry_records is not None or entry_families is not None):
raise TypeError("records is mutually exclusive with entry_records and entry_families")
supplied = normalize_entry_declaration(entry_records, entry_families, records)
self._initialize_layout(supplied)
self._entry_record_types: dict[type, tuple[str, int, int]] = {
record: (self._family_entry_type(family.family), len(family.records), backing_index)
for family in self.layout.families
if family.definition_id is not None
for backing_index, record in enumerate(family.records)
}
def __repr__(self) -> str:
return f"{type(self).__name__}(database={self._database!r}, write_profile={self._write_profile!r})"
[docs]
def close(self) -> None:
"""Dispose the underlying :class:`~httk.store.backend.sql.engine.Backend`, but only when this store owns it.
A store constructed from a caller-supplied ``Backend`` does not own it,
so :meth:`~httk.store.backend.sql.store.SqlStore.close` is a no-op there and the caller stays responsible for
the connection pool. The per-dialect subclasses build their own
``Backend`` and do dispose it here.
:return: None.
"""
if self._owns_database:
self._database.dispose()
def __enter__(self) -> Self:
"""Enter a context whose exit closes the store.
:return: This store.
"""
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
traceback: TracebackType | None,
) -> None:
"""Close the store when leaving its context.
:param exc_type: The exception class raised in the context, if any.
:param exc_value: The exception instance raised in the context, if any.
:param traceback: The traceback for the context exception, if any.
:return: None.
"""
self.close()
@staticmethod
def _family_entry_type(family: type) -> str:
"""Return the validated served entry type declared by ``family``."""
entry_type = getattr(family, "type", None)
if not isinstance(entry_type, str) or not entry_type or entry_type != entry_type.strip():
raise ValueError(f"{family.__name__}.type must be a non-empty stripped entry type")
return entry_type
def _entry_id_number(self, record_type: type, logical_id: int) -> int:
"""Return the family-unique numeric component for a record lineage."""
# ponytail: number = logical_id*B+index keeps family-wide uniqueness with no allocator table; switch to a per-family sequence table if backings can be added to a family after data exists
_entry_type, backing_count, backing_index = self._entry_record_types[record_type]
return logical_id * backing_count + backing_index
@property
[docs]
def layout(self) -> StorageLayout:
"""Return the immutable persisted entry declaration and resolved classes.
:return: The persisted storage layout.
"""
assert self._layout is not None
return self._layout
@property
[docs]
def backend_facts(self) -> BackendFacts:
"""Return the dialect capabilities resolved when this store was opened."""
assert self._backend_facts is not None
return self._backend_facts
@property
[docs]
def write_profile(self) -> Literal["transactional", "degraded", "bulk-fenced"]:
"""Return the persisted permanentization write profile."""
return self._write_profile
@property
[docs]
def store_timestamps(self) -> bool:
"""Whether parent rows carry store-managed timestamps."""
return self._store_timestamps
@property
[docs]
def store_timestamp_resolution(self) -> int | None:
"""Return nanoseconds per stored timestamp unit, or ``None`` when disabled."""
return self._store_timestamp_resolution if self._store_timestamps else None
@property
def _store_timestamp_state(self) -> str:
return encode_store_timestamp_state(self._store_timestamps, self._store_timestamp_resolution)
@property
def _instances(self) -> Any:
"""Compatibility view of the shared instance cache for the SQL hydrator."""
return self._identity._instances
@property
def _sids_by_identity(self) -> Any:
"""Compatibility view of the unhashable-instance reverse cache."""
return self._identity._sids_by_identity
@property
[docs]
def entry_layout(self) -> tuple[EntryFamilyLayout, ...]:
"""Return configured entry-family layouts in deterministic stable-name order.
:return: The configured entry-family layouts.
"""
return self.layout.families
@property
[docs]
def entry_records(self) -> Mapping[type, tuple[type, ...]]:
"""Return configured entry-family classes mapped to ordered concrete records.
:return: The entry-family to concrete-record mapping.
"""
return self.layout.entry_records
def _initialize_layout(self, supplied: StorageLayout | None) -> None:
try:
with self._degraded_lifecycle_guard(), self._database.engine.begin() as connection:
self._initialize_layout_on_connection(connection, supplied)
except BaseException:
created_tables = tuple(self._initialization_ddl_journal)
self._initialization_ddl_journal.clear()
# The transaction context has now unwound. This ordering matters
# for SQLite, whose DDL can survive SQLAlchemy's outer rollback;
# opening the cleanup transaction while the original one is still
# active would merely fail against its shared connection.
if created_tables:
self._cleanup_initialization_tables(created_tables)
self._metadata = sqlalchemy.MetaData()
self._layout = None
self._managed_table_names = frozenset()
self._tables_present.clear()
# The memo maps class-sets to names registered in _metadata; a hit
# skips re-registration, so it must be dropped with _metadata.
self._candidate_names.clear()
self._initialized = False
# No rollback token here: this runs during initialization, before any
# user code could hold a lazy row read on this thread-local connection.
self._clear_identity_caches()
raise
self._initialization_ddl_journal.clear()
def _initialize_layout_on_connection(
self,
connection: sqlalchemy.Connection,
supplied: StorageLayout | None,
) -> None:
self._backend_facts = backend_facts_for_dialect(connection.dialect.name)
self._validate_write_profile_connection(connection, self._write_profile)
objects_before = actual_schema_objects(connection)
names_before = frozenset(name for name, kinds in objects_before.items() if "table" in kinds)
if METADATA_TABLE_NAME in names_before:
self._open_existing_layout_on_connection(connection, supplied)
return
if not objects_before:
if supplied is None:
raise TypeError("entry_records or entry_families is required when opening an uninitialized database")
expected = expected_metadata(supplied, store_timestamps=self._store_timestamps)
metadata_table = expected.tables[METADATA_TABLE_NAME]
if self.backend_facts.metadata_backend == "keepermap":
from httk.store.backend.clickhouse.support import bootstrap_fence, keeper_database_uuid
metadata_table.info["httk_clickhouse_database_uuid"] = keeper_database_uuid(connection)
with bootstrap_fence(connection):
fenced_objects = actual_schema_objects(connection)
if METADATA_TABLE_NAME in fenced_objects:
self._open_existing_layout_on_connection(connection, supplied, retry_metadata_visibility=True)
return
try:
metadata_table.create(connection, checkfirst=False)
except BaseException as error:
try:
after_create_error = actual_schema_objects(connection)
except BaseException:
raise RuntimeError(
"ClickHouse bootstrap contention state recheck failed after metadata creation "
"error; the database UUID fence refused concurrent initialization"
) from error
if METADATA_TABLE_NAME in after_create_error:
self._open_existing_layout_on_connection(
connection, supplied, retry_metadata_visibility=True
)
return
raise RuntimeError(
"ClickHouse bootstrap contention during first metadata-table creation; "
"the database UUID fence refused concurrent initialization"
) from error
self._initialization_ddl_journal.append(metadata_table)
self._stamp_layout(connection, supplied, fence_held=True)
else:
metadata_table.create(connection, checkfirst=False)
self._initialization_ddl_journal.append(metadata_table)
for owner_table in identity_owner_tables(expected):
owner_table.create(connection, checkfirst=False)
self._initialization_ddl_journal.append(owner_table)
self._stamp_layout(connection, supplied)
self._install_layout(supplied, expected, names_before | {METADATA_TABLE_NAME})
self._initialize_store_timestamp_mark(connection)
return
schema: dict[str, object] = {METADATA_TABLE_NAME: {"missing": True}}
for name, kinds in sorted(objects_before.items()):
object_type = _schema_object_type(kinds)
if name.startswith("_httk_"):
schema[name] = {
"reserved": True,
"object_type": object_type,
"message": "unexpected schema object uses the SqlStore-reserved _httk_ prefix",
}
else:
schema[name] = {
"unversioned": True,
"object_type": object_type,
"message": "a nonempty database without SqlStore metadata cannot be adopted",
}
raise StorageLayoutUpgradeRequiredError(
{
"protocol": {"expected": STORAGE_PROTOCOL_VERSION, "actual": None},
"declaration": {
"expected": declaration_json(supplied) if supplied is not None else "explicit entry_records",
"actual": None,
},
"schema": schema,
}
)
def _open_existing_layout_on_connection(
self,
connection: sqlalchemy.Connection,
supplied: StorageLayout | None,
*,
retry_metadata_visibility: bool = False,
) -> None:
"""Validate and open an existing marked layout on this connection."""
if self.backend_facts.metadata_backend == "keepermap":
from httk.store.backend.clickhouse.support import validate_metadata_table
validate_metadata_table(connection)
stored = None
read_error: ValueError | SQLAlchemyError | None = None
visible_metadata_keys = {"protocol", "entry_declaration", "entry_schemas", "store_timestamps"}
if self.backend_facts.metadata_backend == "keepermap":
visible_metadata_keys.add("write_profile")
for attempt in range(20 if retry_metadata_visibility else 1):
try:
stored = read_store_metadata(connection)
read_error = None
except (ValueError, SQLAlchemyError) as error:
read_error = error
if read_error is None and stored is not None and visible_metadata_keys <= set(stored):
break
if attempt + 1 < (20 if retry_metadata_visibility else 1):
time.sleep(0.05)
if read_error is not None:
raise StorageLayoutUpgradeRequiredError(
{"declaration": {"metadata": "malformed", "error": str(read_error)}}
) from read_error
if stored is None:
raise StorageLayoutUpgradeRequiredError({"declaration": {"metadata": "missing"}})
if "ingest_state" in stored:
raise StoreUnderConstructionError(
"ingest_state marker from an interrupted bulk ingest is present; "
"the database must be dropped and re-ingested rather than clearing the marker"
)
self._open_marked_layout(connection, stored, supplied)
def _validate_write_profile_connection(
self, connection: sqlalchemy.Connection, profile: Literal["transactional", "degraded", "bulk-fenced"]
) -> None:
"""Require the requested persisted profile to match the live connection.
``Backend.degraded`` selects a requested profile, but custom engines
can disagree with it. Permanentization's safety properties depend on
the DBAPI connection actually being SQLite autocommit, so every open
validates both dialect and live isolation state before inspecting or
creating store metadata.
"""
autocommit = connection_uses_autocommit(connection)
if profile == "degraded":
if connection.dialect.name != "sqlite" or not autocommit:
raise StorageLayoutUpgradeRequiredError(
{
"declaration": {
"write_profile": {
"expected": "SQLite autocommit connection for degraded profile",
"actual": {
"dialect": connection.dialect.name,
"autocommit": autocommit,
},
}
}
}
)
return
if profile == "bulk-fenced":
if connection.dialect.name != "clickhousedb":
raise StorageLayoutUpgradeRequiredError(
{
"declaration": {
"write_profile": {
"expected": "ClickHouse clickhousedb connection for bulk-fenced profile",
"actual": connection.dialect.name,
}
}
}
)
return
if autocommit:
raise StorageLayoutUpgradeRequiredError(
{"declaration": {"write_profile": "transactional profile rejects an SQLite autocommit engine"}}
)
def _open_marked_layout(
self,
connection: sqlalchemy.Connection,
stored: Mapping[str, str],
supplied: StorageLayout | None,
) -> None:
required_keys = {
"protocol",
"entry_declaration",
"entry_schemas",
"store_timestamps",
_IDENTITY_OWNERSHIP_KEY,
}
persistent_optional_keys = {"write_profile"}
recognized_runtime_keys = {"ingest_state", "lease"}
allowed_keys = required_keys | persistent_optional_keys | recognized_runtime_keys
diff: dict[str, object] = {}
def declaration_diff() -> dict[str, object]:
# Each check contributes one named aspect; independent mismatches
# accumulate instead of overwriting a single "declaration" payload.
return cast("dict[str, object]", diff.setdefault("declaration", {}))
unknown_keys = {
key
for key in stored
if key not in allowed_keys and not (key.startswith("dirty:") and len(key) > len("dirty:"))
}
missing_keys = required_keys - set(stored)
ownership_upgrade_pending = missing_keys == {_IDENTITY_OWNERSHIP_KEY}
if unknown_keys or (missing_keys and not ownership_upgrade_pending):
declaration_diff()["metadata_keys"] = {
"expected": tuple(sorted(required_keys)),
"recognized_runtime": tuple(sorted(recognized_runtime_keys)),
"actual": tuple(sorted(stored)),
}
if stored.get("protocol") != STORAGE_PROTOCOL_VERSION:
diff["protocol"] = {"expected": STORAGE_PROTOCOL_VERSION, "actual": stored.get("protocol")}
if not ownership_upgrade_pending and stored.get(_IDENTITY_OWNERSHIP_KEY) != _IDENTITY_OWNERSHIP_VERSION:
declaration_diff()[_IDENTITY_OWNERSHIP_KEY] = {
"expected": _IDENTITY_OWNERSHIP_VERSION,
"actual": stored.get(_IDENTITY_OWNERSHIP_KEY),
}
persisted_timestamps = stored.get("store_timestamps")
parsed_timestamps = parse_store_timestamp_state(persisted_timestamps)
effective_timestamps = None if parsed_timestamps is None else parsed_timestamps[0]
effective_resolution = None if parsed_timestamps is None else parsed_timestamps[1]
if persisted_timestamps not in (None, "off") and parsed_timestamps is None:
declaration_diff()["store_timestamps"] = {
"expected": self._store_timestamp_state,
"actual": persisted_timestamps,
}
elif persisted_timestamps is None:
declaration_diff()["store_timestamps"] = {"expected": self._store_timestamp_state, "actual": None}
elif effective_timestamps != self._store_timestamps or (
effective_timestamps and effective_resolution != self._store_timestamp_resolution
):
declaration_diff()["store_timestamps"] = {
"expected": self._store_timestamp_state,
"actual": persisted_timestamps,
}
persisted_profile = stored.get("write_profile", "transactional")
if persisted_profile not in {"transactional", "degraded", "bulk-fenced"}:
declaration_diff()["write_profile"] = {"actual": persisted_profile}
elif persisted_profile != self._write_profile:
declaration_diff()["write_profile"] = {
"expected": self._write_profile,
"actual": persisted_profile,
"message": "open the store with a Backend selecting the persisted write profile",
}
persisted: StorageLayout | None = None
stored_declaration = stored.get("entry_declaration")
if supplied is not None and isinstance(stored_declaration, str):
if stored_declaration == declaration_json(supplied):
persisted = supplied
else:
declaration_diff()["entry_declaration"] = {
"expected": stored_declaration,
"actual": declaration_json(supplied),
}
else:
try:
persisted = self._layout_from_stored_declaration(stored_declaration)
except EntryLayoutBindingError:
raise
except (TypeError, ValueError) as error:
declaration_diff()["entry_declaration"] = {
"expected": "canonical registered declaration or explicit entry_families binding",
"actual": stored_declaration,
"error": str(error),
}
if persisted is not None and "entry_schemas" in stored:
# Absence of the key is already reported by required_keys above.
schema_diff = schema_fingerprint_diff(stored["entry_schemas"], schema_fingerprint_json(persisted))
if schema_diff:
diff["schema"] = schema_diff
upgrade_plan: AdditiveUpgradePlan | None = None
if set(diff) == {"schema"}:
assert persisted is not None
plan = classify_schema_upgrade(stored["entry_schemas"], schema_fingerprint_json(persisted))
if isinstance(plan, AdditiveUpgradePlan):
if self.backend_facts.metadata_backend == "keepermap":
raise StorageLayoutUpgradeRequiredError(
diff, hint="additive schema upgrade is not supported on the ClickHouse bulk-fenced backend"
)
if not self._upgrade:
raise StorageLayoutUpgradeRequiredError(diff, hint=ADDITIVE_UPGRADE_HINT)
# The apply is deferred until every other verification below has
# passed: SQLite DDL escapes the open transaction, so a later
# failing check must not be able to strand a half-applied upgrade.
upgrade_plan = plan
diff = {}
if diff:
raise StorageLayoutUpgradeRequiredError(diff)
assert persisted is not None
assert persisted_profile in {"transactional", "degraded", "bulk-fenced"}
self._write_profile = cast(Literal["transactional", "degraded", "bulk-fenced"], persisted_profile)
self._validate_write_profile_connection(connection, self._write_profile)
objects_before = actual_schema_objects(connection)
names_before = frozenset(name for name, kinds in objects_before.items() if "table" in kinds)
invalid_dirty = sorted(
key for key in stored if key.startswith("dirty:") and key.removeprefix("dirty:") not in names_before
)
if invalid_dirty:
raise StorageLayoutUpgradeRequiredError(
{"declaration": {"metadata_keys": {"invalid_dirty": tuple(invalid_dirty)}}}
)
declaration_owned = {
METADATA_TABLE_NAME,
"_httk_sid_counters",
ENTRY_ID_OWNERS_TABLE_NAME,
IMMUTABLE_ID_OWNERS_TABLE_NAME,
*(entry_dispatch_table_name(family.name) for family in persisted.families if len(family.records) > 1),
}
object_problems: dict[str, object] = {}
for name, kinds in objects_before.items():
if not name.startswith("_httk_"):
continue
# A weak-link table (and its backing sid sequence on dialects such as
# DuckDB/Postgres) is store-owned: the reserved _httk_link_ prefix is
# produced only by a WeakLink declaration, never by a record's storage
# name. It may belong to an ad-hoc (undeclared) record and so is not
# in declaration_owned, and at reopen it is not yet in metadata — so
# accept it structurally from the live schema: a real link table
# carries the source_lid/target_lid endpoint columns, its sid sequence
# is a column-less sequence, and an alien table/view merely wearing
# the prefix has neither and is flagged.
if name.startswith("_httk_link_"):
columns = actual_columns(connection, name) if "table" in kinds else frozenset()
if kinds == {"sequence"} or (SOURCE_LID_COLUMN in columns and TARGET_LID_COLUMN in columns):
continue
if name not in declaration_owned or kinds != {"table"}:
object_problems[name] = {
"reserved": True,
"object_type": _schema_object_type(kinds),
"message": "unexpected schema object uses the SqlStore-reserved _httk_ prefix",
}
if object_problems:
raise StorageLayoutUpgradeRequiredError({"schema": object_problems})
if not ownership_upgrade_pending and self.backend_facts.metadata_backend != "keepermap":
self._validate_identity_owner_tables(connection)
if ownership_upgrade_pending:
self._identity_ownership_ready = False
if self._upgrade:
self._upgrade_identity_ownership(connection, persisted)
self._identity_ownership_ready = True
names_before = actual_table_names(connection)
if upgrade_plan is not None:
self._apply_additive_upgrade(connection, persisted, upgrade_plan)
names_before = actual_table_names(connection)
self._install_layout(
persisted,
expected_metadata(persisted, store_timestamps=self._store_timestamps),
names_before,
)
self._initialize_store_timestamp_mark(connection)
@staticmethod
def _layout_from_stored_declaration(value: str | None) -> StorageLayout:
# The canonical JSON parser is intentionally private to layout.py; a
# no-op explicit declaration round-trip uses the public normalizer.
from httk.store.backend.sql.layout import _layout_from_declaration
if value is None:
raise ValueError("metadata is missing entry_declaration")
return _layout_from_declaration(value)
def _stamp_layout(
self,
connection: sqlalchemy.Connection,
layout: StorageLayout,
*,
fence_held: bool = False,
) -> None:
table = metadata_table_for(sqlalchemy.MetaData())
rows = {
"protocol": STORAGE_PROTOCOL_VERSION,
"entry_declaration": declaration_json(layout),
"entry_schemas": schema_fingerprint_json(layout),
"store_timestamps": self._store_timestamp_state,
_IDENTITY_OWNERSHIP_KEY: _IDENTITY_OWNERSHIP_VERSION,
}
if self._write_profile != "transactional":
rows["write_profile"] = self._write_profile
if self.backend_facts.metadata_backend == "keepermap":
from httk.store.backend.clickhouse.support import stamp_store_metadata
stamp_store_metadata(connection, table, rows, fence_held=fence_held)
return
connection.execute(
sqlalchemy.insert(table),
tuple({"key": key, "value": value} for key, value in rows.items()),
)
def _apply_additive_upgrade(
self, connection: sqlalchemy.Connection, layout: StorageLayout, plan: AdditiveUpgradePlan
) -> None:
"""Create newly declared tables, add the plan's nullable columns, and re-stamp the fingerprint.
New tables are created whole (checkfirst) so a pre-existing row whose
closure references them stops reading as absent; pre-existing tables gain
their new nullable columns. Every step is idempotent — live columns are
reflected and already-present ones skipped, and the index create is
``IF NOT EXISTS`` — so a crash mid-upgrade (SQLite DDL escapes the
transaction) self-heals on retry and a concurrent ``upgrade=True`` loser
is benign. The re-stamp is last so a failure leaves the store repeatable.
:param connection: The open initialization connection/transaction.
:param layout: The persisted layout whose current fingerprint is re-stamped.
:param plan: The additive upgrade plan of per-table added parent columns.
:return: None.
"""
expected_metadata(layout, store_timestamps=self._store_timestamps).create_all(connection, checkfirst=True)
for table_name, columns in plan.added_columns.items():
present_columns = actual_columns(connection, table_name)
for spec in columns:
statements = added_column_ddl(table_name, spec, connection)
if spec.name in present_columns:
# Column already added by a prior (possibly crashed) run; only
# the idempotent IF NOT EXISTS index statement remains to run.
statements = statements[1:]
for statement in statements:
connection.execute(sqlalchemy.text(statement))
table = metadata_table_for(sqlalchemy.MetaData())
connection.execute(
sqlalchemy.update(table).where(table.c.key == "entry_schemas").values(value=schema_fingerprint_json(layout))
)
def _upgrade_identity_ownership(self, connection: sqlalchemy.Connection, layout: StorageLayout) -> None:
"""Validate existing entry ids, create claims, then mark the capability complete."""
if self.backend_facts.metadata_backend == "keepermap":
self._ensure_degraded_lease(connection)
self._sync_identity_ownership(connection, layout)
from httk.store.backend.clickhouse.support import stamp_store_metadata
stamp_store_metadata(
connection,
metadata_table_for(sqlalchemy.MetaData()),
{_IDENTITY_OWNERSHIP_KEY: _IDENTITY_OWNERSHIP_VERSION},
)
return
self._ensure_degraded_lease(connection)
owner_metadata = sqlalchemy.MetaData()
for owner_table in identity_owner_tables(owner_metadata):
owner_table.create(connection, checkfirst=True)
self._validate_identity_owner_tables(connection)
self._sync_identity_ownership(connection, layout)
metadata = metadata_table_for(sqlalchemy.MetaData())
connection.execute(
sqlalchemy.insert(metadata).values(key=_IDENTITY_OWNERSHIP_KEY, value=_IDENTITY_OWNERSHIP_VERSION)
)
def _sync_identity_ownership(
self,
connection: sqlalchemy.Connection,
layout: StorageLayout,
*,
sid_ranges: Mapping[str, tuple[int, int]] | None = None,
) -> None:
"""Validate and claim identities, optionally for appended rows only.
``sid_ranges`` maps physical backing tables to half-open allocated sid
ranges. Omitted tables are skipped; None scans every backing for upgrade
and initial-build validation. Even scoped claims are checked against all
durable owners by the ownership tables' native uniqueness constraints.
"""
if sid_ranges is not None and not sid_ranges:
return
present = actual_table_names(connection)
for family in layout.families:
if family.definition_id is None:
continue
selections: dict[str, list[Any]] = {"id": [], "immutable_id": []}
for backing_name, record_type in zip(family.record_names, family.records, strict=True):
table_name = resolve_schema(record_type).table_name
if table_name not in present:
continue
if sid_ranges is not None and table_name not in sid_ranges:
continue
table = table_for(
resolve_schema(record_type),
sqlalchemy.MetaData(),
store_timestamps=self._store_timestamps,
)
for field, owner_column in (
("id", ALT_ID_COLUMN),
("immutable_id", SID_COLUMN),
):
selection = sqlalchemy.select(
table.c[field].label("value"),
sqlalchemy.literal(backing_name).label("backing"),
table.c[owner_column].label("owner"),
).distinct()
if sid_ranges is not None:
start, stop = sid_ranges[table_name]
selection = selection.where(table.c[SID_COLUMN] >= start, table.c[SID_COLUMN] < stop)
selections[field].append(selection)
for field, selects in selections.items():
if not selects:
continue
rows = sqlalchemy.union_all(*selects).subquery()
missing = connection.execute(
sqlalchemy.select(rows.c.backing, rows.c.owner).where(rows.c.value.is_(None)).limit(1)
).first()
if missing is not None:
raise RuntimeError(
f"entry family {family.name!r} backing {missing[0]!r} owner {missing[1]} is missing {field!r}"
)
for columns in ((rows.c.value,), (rows.c.backing, rows.c.owner)):
conflict = connection.execute(
sqlalchemy.select(sqlalchemy.func.min(rows.c.value))
.group_by(*columns)
.having(sqlalchemy.func.count() > 1)
.limit(1)
).first()
if conflict is not None:
raise EntryIdConflictError(family.name, str(conflict[0]), None, None)
if self.backend_facts.metadata_backend == "keepermap":
# KeeperMap's exclusive lease and construction marker fence the
# only supported write: one empty-store ingest. Final rows are
# immutable afterwards; MergeTree is never treated as unique.
continue
entry, immutable = identity_owner_tables(sqlalchemy.MetaData())
target = entry if field == "id" else immutable
id_column = "entry_id" if field == "id" else "immutable_id"
owner_column = "logical_id" if field == "id" else "sid"
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
insert = sqlite_insert if connection.dialect.name == "sqlite" else pg_insert
connection.execute(
insert(target)
.from_select(
["family", id_column, "backing", owner_column],
sqlalchemy.select(
sqlalchemy.literal(family.name),
rows.c.value,
rows.c.backing,
rows.c.owner,
).where(sqlalchemy.true()),
)
.on_conflict_do_nothing()
)
conflict = connection.execute(
sqlalchemy.select(rows.c.value)
.select_from(
rows.outerjoin(
target,
sqlalchemy.and_(
target.c.family == family.name,
target.c[id_column] == rows.c.value,
target.c.backing == rows.c.backing,
target.c[owner_column] == rows.c.owner,
),
)
)
.where(target.c.family.is_(None))
.limit(1)
).first()
if conflict is not None:
raise EntryIdConflictError(family.name, str(conflict[0]), None, None)
@staticmethod
def _validate_identity_owner_tables(connection: sqlalchemy.Connection) -> None:
present = actual_table_names(connection)
for table in identity_owner_tables(sqlalchemy.MetaData()):
expected = {
tuple(column.name for column in constraint.columns)
for constraint in table.constraints
if isinstance(
constraint,
(sqlalchemy.PrimaryKeyConstraint, sqlalchemy.UniqueConstraint),
)
}
if table.name not in present or actual_columns(connection, table.name) != frozenset(table.c.keys()):
raise StorageLayoutUpgradeRequiredError(
{"schema": {table.name: {"identity_ownership": "missing or malformed table"}}}
)
if connection.dialect.name == "duckdb":
constraints = connection.execute(
sqlalchemy.text(
"SELECT constraint_column_names FROM duckdb_constraints() WHERE database_name=current_database() "
"AND schema_name=current_schema() AND table_name=:name AND constraint_type IN ('PRIMARY KEY', 'UNIQUE')"
),
{"name": table.name},
)
actual = {tuple(row[0]) for row in constraints}
else:
inspector = sqlalchemy.inspect(connection)
actual = {tuple(inspector.get_pk_constraint(table.name)["constrained_columns"])}
actual.update(tuple(item["column_names"]) for item in inspector.get_unique_constraints(table.name))
if not expected <= actual:
raise StorageLayoutUpgradeRequiredError(
{"schema": {table.name: {"identity_ownership": "missing unique constraint"}}}
)
def _claim_identity_values(
self,
connection: sqlalchemy.Connection,
family_name: str,
backing_name: str,
entry_id: str,
group_id: int,
immutable_id: str,
sid: int,
) -> None:
"""Atomically claim one logical and immutable identity through native unique constraints."""
entry_owners, immutable_owners = identity_owner_tables(sqlalchemy.MetaData())
claims = (
(
entry_owners,
{
"family": family_name,
"entry_id": entry_id,
"backing": backing_name,
"logical_id": group_id,
},
),
(
immutable_owners,
{
"family": family_name,
"immutable_id": immutable_id,
"backing": backing_name,
"sid": sid,
},
),
)
dialect = connection.dialect.name
for table, values in claims:
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
insert = sqlite_insert if dialect == "sqlite" else pg_insert
try:
connection.execute(insert(table).values(values).on_conflict_do_nothing())
except IntegrityError as error:
raise EntryIdConflictError(family_name, entry_id, None, group_id) from error
id_column = "entry_id" if table is entry_owners else "immutable_id"
owner_column = LOGICAL_ID_COLUMN if table is entry_owners else SID_COLUMN
expected_owner = group_id if table is entry_owners else sid
existing = connection.execute(
sqlalchemy.select(table.c.backing, table.c[owner_column]).where(
table.c.family == family_name,
table.c[id_column] == values[id_column],
)
).one_or_none()
if existing != (backing_name, expected_owner):
raise EntryIdConflictError(
family_name,
str(values[id_column]),
None if existing is None else int(existing[1]),
expected_owner,
)
def _claim_entry_identity(
self,
connection: sqlalchemy.Connection,
record_type: type,
values: Mapping[str, Any],
*,
group_id: int,
sid: int,
) -> None:
family = self._family_for_backing(record_type)
if family is None or family.definition_id is None or self.backend_facts.metadata_backend == "keepermap":
return
backing_name = family.record_names[family.records.index(record_type)]
self._claim_identity_values(
connection,
family.name,
backing_name,
str(values["id"]),
group_id,
str(values["immutable_id"]),
sid,
)
def _release_unused_identity_claims(self, connection: sqlalchemy.Connection, table_name: str) -> None:
"""Remove claims whose owner row is absent, under a transaction or degraded lease."""
for family in self.layout.families:
if family.definition_id is None:
continue
for backing, record in zip(family.record_names, family.records, strict=True):
if resolve_schema(record).table_name != table_name:
continue
parent = self._table(table_name)
entry, immutable = identity_owner_tables(sqlalchemy.MetaData())
for owners, owner_column, parent_column in (
(entry, "logical_id", ALT_ID_COLUMN),
(immutable, "sid", SID_COLUMN),
):
connection.execute(
sqlalchemy.delete(owners).where(
owners.c.family == family.name,
owners.c.backing == backing,
~sqlalchemy.exists(
sqlalchemy.select(1).where(parent.c[parent_column] == owners.c[owner_column])
),
)
)
def _initialize_store_timestamp_mark(self, connection: sqlalchemy.Connection) -> None:
"""Derive the writable process-local timestamp mark from present parent tables."""
if not self._store_timestamps or self._allow_clock_regression:
self._store_timestamp_mark = None
return
maximum: int | None = None
tables: dict[str, sqlalchemy.Table] = dict(self._metadata.tables)
reflection_metadata = sqlalchemy.MetaData()
durable_tables = actual_table_names(connection)
for name in durable_tables - tables.keys():
if name.startswith("_httk_"):
continue
try:
tables[name] = sqlalchemy.Table(name, reflection_metadata, autoload_with=connection)
except SQLAlchemyError:
continue
for name, table in tables.items():
# Link tables carry store_timestamp but no _httk_role column; their
# rows must still advance the clock-regression mark on reopen.
is_link = name.startswith("_httk_link_")
if (
name not in durable_tables
or STORE_TIMESTAMP_COLUMN not in table.c
or (ROLE_COLUMN not in table.c and not is_link)
):
continue
value = connection.execute(
sqlalchemy.select(sqlalchemy.func.max(table.c[STORE_TIMESTAMP_COLUMN]))
).scalar_one()
if value is not None:
maximum = int(value) if maximum is None else max(maximum, int(value))
self._store_timestamp_mark = maximum
def _capture_store_timestamp(self, connection: sqlalchemy.Connection) -> int | None:
"""Capture one guarded store-unit timestamp for a save or ingest batch."""
if not self._store_timestamps:
return None
return capture_store_timestamp(
self._clock,
self._store_timestamp_resolution,
self._store_timestamp_mark,
allow_clock_regression=self._allow_clock_regression,
clock_regression_grace=self._clock_regression_grace,
)
def _advance_store_timestamp_mark(self, captured: int | None) -> None:
if captured is not None and not self._allow_clock_regression:
self._store_timestamp_mark = advance_store_timestamp_mark(
self._store_timestamp_mark, captured, allow_clock_regression=self._allow_clock_regression
)
def _install_layout(
self,
layout: StorageLayout,
metadata: sqlalchemy.MetaData,
table_names: Iterable[str],
) -> None:
self._layout = layout
self._metadata = metadata
self._managed_table_names = frozenset(metadata.tables)
self._tables_present = set(table_names)
self._initialized = True
def _cleanup_initialization_tables(self, created_tables: tuple[sqlalchemy.Table, ...]) -> None:
# SQLite DDL can escape SQLAlchemy's outer rollback helper. A fresh
# explicit cleanup transaction means a failed first initialization is
# never left with a usable partial marker/layout. DuckDB has already
# rolled it back at this point; IF EXISTS makes the same path harmless.
try:
with self._database.engine.begin() as cleanup:
for table in reversed(created_tables):
cleanup.execute(sqlalchemy.schema.DropTable(table, if_exists=True))
except BaseException:
# Preserve the original initialization error. A remaining table
# still has no marker and will be refused as unversioned.
return
# ------------------------------------------------------------------ tables and transactions
def _reject_during_bulk(self) -> None:
"""Refuse ordinary writes while a :meth:`bulk_ingest` context owns the store.
:raises RuntimeError: If a bulk-ingest context is currently open.
"""
with self._bulk_state_lock:
active = self._bulk_active
if active:
raise RuntimeError(
"this SqlStore has an open bulk_ingest context; ordinary save/ensure_tables/transaction "
"operations are refused until it exits"
)
def _claim_bulk_context(self) -> None:
"""Atomically reserve this store for one bulk context."""
with self._bulk_state_lock:
if self._bulk_active:
raise RuntimeError("this SqlStore already has an open bulk_ingest context")
self._bulk_active = True
def _release_bulk_context(self) -> None:
"""Release the short-lived in-memory bulk admission state."""
with self._bulk_state_lock:
self._bulk_active = False
def _check_mutation_policy(self, operation: str, *, empty_deferred_bulk: bool = False) -> None:
"""Apply the single public mutation policy for backend capability gates."""
if not self._identity_ownership_ready:
raise StorageLayoutUpgradeRequiredError(
{
"declaration": {
_IDENTITY_OWNERSHIP_KEY: {
"expected": _IDENTITY_OWNERSHIP_VERSION,
"actual": None,
}
}
},
hint="reopen with upgrade=True to validate and install durable entry-id ownership before writing",
)
if self.backend_facts.supports_incremental_save:
return
if operation == "bulk_ingest" and empty_deferred_bulk:
return
raise RuntimeError(
f"{operation} is refused for the clickhousedb bulk-fenced profile in P1; "
"ClickHouse incremental mutations are not supported"
)
[docs]
def bulk_ingest(
self,
*,
chunk_size: int = 100_000,
verify_metadata: bool = True,
index_strategy: Literal["auto", "keep", "rebuild"] = "auto",
on_progress: Callable[[int, int], None] | None = None,
workers: int = 1,
finalize: Literal["auto", "parity", "deferred"] = "auto",
track_sids: bool = True,
id_series: str | None = None,
) -> "BulkIngest":
"""Return a context manager that appends a stream of objects into this store.
The returned :class:`~httk.store.backend.sql.bulk.BulkIngest` exposes
``save(obj, *, as_record=None, promote=None) -> int`` mirroring :meth:`save`, but
buffers encoded rows with pre-assigned sids and appends them in
executemany batches. On a physically empty store the record tables are
created index-less and their separable indexes are built once the stream
completes; on a populated store each flushed chunk is staged and
resolved set-wise against the existing rows (content-id anti-join with
metadata verification, ``by_value`` whole-column anti-join, and a sid
remap of the surviving references) before it is appended. While the
context is open the store's ordinary write path is exclusively owned:
:meth:`save`, :meth:`ensure_tables` and :meth:`transaction` raise
:class:`RuntimeError`.
A sid returned by ``save`` inside the context is provisional: a record
that deduplicates against a pre-existing row is remapped at flush, so its
durable sid is obtained from :meth:`~httk.store.backend.sql.bulk.BulkIngest.resolved_sid`
once the context has exited cleanly.
``save(..., promote=RecordClass)`` additionally makes every nested
occurrence of that record class a top-level entry without a second
projection or worker transfer. An iterable promotes several classes.
:param chunk_size: The number of top-level saves buffered before a flush.
:param verify_metadata: Whether content-id hits compare identity-excluded metadata.
:param index_strategy: How an existing table's separable indexes are handled during the append —
``"keep"`` appends through them, ``"rebuild"`` drops and recreates them, and ``"auto"`` picks
per table by the staged-to-existing row ratio. On DuckDB, which reserves a dropped index's name
until commit, ``"rebuild"`` instead keeps the indexes and verifies content-id uniqueness with a
duplicate scan; the final indexes are the same either way.
:param on_progress: An optional ``(records_buffered_total, rows_flushed_total)`` callback invoked after each flush.
:param workers: The number of worker processes. ``1`` (the default) is the serial path with byte-for-byte
unchanged semantics; ``>1`` encodes the stream in forked worker processes and merges their per-table
shards set-wise. Parallel mode requires a physically empty target store (the offline-build use case) and,
on DuckDB, the ``httk-store[parallel]`` extra (pyarrow); incremental appends stay on the serial path.
:param finalize: ``"parity"`` selects the historical in-database ingest; ``"deferred"`` stages a physically
empty ingest outside the store and finalizes it at context exit; ``"auto"`` selects deferred only for a
physically empty, supported serial ingest and otherwise selects parity (including ``workers>1``). A
subclass may override :attr:`bulk_ingest_finalize_default` for ``"auto"`` calls.
:param track_sids: Whether to retain provisional-to-durable sid mappings.
:param id_series: Override the configured entry-id series for minted bulk rows.
:return: A bulk-ingest context manager bound to this store.
"""
from httk.store.backend.sql.bulk import BulkIngest
return BulkIngest(
self,
chunk_size=chunk_size,
verify_metadata=verify_metadata,
index_strategy=index_strategy,
on_progress=on_progress,
workers=workers,
finalize=finalize,
track_sids=track_sids,
id_series=id_series,
)
[docs]
def ensure_tables(self, *classes: type) -> None:
r"""Create the requested tables as an explicit write operation.
:param \*classes: The storable classes whose tables should exist.
:return: None.
:raises RuntimeError: If a :meth:`bulk_ingest` context is currently open.
"""
self._check_mutation_policy("ensure_tables")
self._reject_during_bulk()
with self._write_connection() as connection:
self._create_tables_for_write(connection, classes)
[docs]
def transaction(self) -> contextlib.AbstractContextManager[None]:
"""Return a context manager for one database transaction.
:return: A transaction context manager that commits on normal exit and rolls back on failure.
"""
return self._transaction_scope()
@contextlib.contextmanager
def _transaction_scope(self) -> Iterator[None]:
self._check_mutation_policy("transaction")
self._reject_during_bulk()
stack = self._connection_stack()
if stack:
# Nested scopes are bare passthroughs: they share the outermost
# scope's connection and its single rollback token, so they neither
# allocate a token nor clear it.
yield
return
pending = self._pending_table_names()
timestamp_state = {"initialized": False, "captured": None}
# One token per outermost transaction; lazy rows built inside it record
# it and expire if this scope rolls back.
token = _TransactionToken()
self._local.transaction_token = token
try:
with self._mutation_lock:
with self._database.engine.begin() as connection:
self._ensure_degraded_lease(connection)
stack.append(connection)
try:
self._local.store_timestamp_transaction = timestamp_state
try:
yield
if token.rolled_back:
raise RuntimeError("transaction was rolled back after an entry-id conflict")
except BaseException:
if self._write_profile == "degraded":
self._initialize_store_timestamp_mark(connection)
raise
finally:
self._local.store_timestamp_transaction = None
stack.pop()
self._advance_store_timestamp_mark(timestamp_state["captured"])
self._tables_present.update(pending)
except BaseException:
token.rolled_back = True
pending.clear()
self._tables_present.clear()
self._clear_identity_caches()
raise
finally:
self._local.transaction_token = None
self._local.store_timestamp_transaction = None
pending.clear()
def _connection_stack(self) -> list[sqlalchemy.Connection]:
stack = getattr(self._local, "stack", None)
if stack is None:
stack = []
self._local.stack = stack
return cast(list[sqlalchemy.Connection], stack)
def _current_connection(self) -> sqlalchemy.Connection | None:
stack = self._connection_stack()
return stack[-1] if stack else None
def _current_transaction_token(self) -> "_TransactionToken | None":
"""Return the outermost transaction's rollback token, or None outside one."""
return getattr(self._local, "transaction_token", None)
@contextlib.contextmanager
def _write_connection(
self,
*,
_publish_after_commit: Callable[[], None] | None = None,
) -> Iterator[sqlalchemy.Connection]:
current = self._current_connection()
if current is not None:
with self._mutation_lock, self._degraded_lifecycle_guard():
self._ensure_degraded_lease(current)
started = self._begin_degraded_operation(current)
try:
yield current
except EntryIdConflictError:
if self._write_profile == "transactional":
token = self._current_transaction_token()
if token is not None:
token.rolled_back = True
current.rollback()
self._pending_table_names().clear()
self._tables_present.clear()
self._clear_identity_caches()
raise
finally:
self._end_degraded_operation(current, started)
return
pending = self._pending_table_names()
try:
with self._mutation_lock, self._degraded_lifecycle_guard():
with self._database.engine.begin() as connection:
self._ensure_degraded_lease(connection)
started = self._begin_degraded_operation(connection)
stack = self._connection_stack()
stack.append(connection)
try:
yield connection
finally:
stack.pop()
self._end_degraded_operation(connection, started)
# Local callback, invoked only after Engine.begin() committed.
if _publish_after_commit is not None:
_publish_after_commit()
self._tables_present.update(pending)
except BaseException:
# No rollback token here: a failed write runs inside save()/ensure_tables,
# where no user code can perform a deferred lazy read on this
# thread-local connection before the cache clear completes.
pending.clear()
self._tables_present.clear()
self._clear_identity_caches()
raise
finally:
pending.clear()
@contextlib.contextmanager
def _fsck_connection(self) -> Iterator[sqlalchemy.Connection]:
"""Open fsck's mutation scope with SQLite's real exclusive write lock."""
if self._current_connection() is not None:
raise RuntimeError("fsck cannot run inside an active SqlStore transaction")
if self._write_profile == "degraded" or self._database.engine.dialect.name != "sqlite":
with self._write_connection() as connection:
yield connection
return
with self._mutation_lock, self._database.engine.connect() as connection:
# ``engine.begin()`` is deferred on SQLite. fsck must block a
# concurrent writer before its first inspection, not merely at
# its first DELETE.
connection.exec_driver_sql("BEGIN IMMEDIATE")
stack = self._connection_stack()
stack.append(connection)
try:
yield connection
except BaseException:
connection.rollback()
raise
else:
connection.commit()
finally:
stack.pop()
def _ensure_degraded_lease(self, connection: sqlalchemy.Connection) -> None:
"""Acquire once and verify on every degraded mutation operation.
The lease intentionally remains held by this ``Backend`` owner until
disposal (or an explicit conditional steal). Transactional stores do
not even query the metadata table here, preserving their save hot path.
"""
if self._write_profile == "bulk-fenced":
from httk.store.backend.clickhouse.support import acquire_lease, verify_lease
if self._lease_value is None:
self._lease_value = acquire_lease(connection, self._lease_owner)
else:
verify_lease(connection, self._lease_value)
return
if self._write_profile != "degraded":
return
table = metadata_table_for(sqlalchemy.MetaData())
if self._lease_value is None:
payload = json.dumps(
{"owner": self._lease_owner, "acquired_at": datetime.datetime.now(datetime.UTC).isoformat()},
sort_keys=True,
separators=(",", ":"),
)
if connection.dialect.name == "sqlite":
connection.execute(
sqlalchemy.text(
'INSERT OR IGNORE INTO "_httk_store_metadata" (key, value) VALUES (\'lease\', :value)'
),
{"value": payload},
)
else: # defensive: degraded is SQLite-only, but keep the primitive explicit.
connection.execute(
sqlalchemy.text(
'INSERT INTO "_httk_store_metadata" (key, value) '
'SELECT \'lease\', :value WHERE NOT EXISTS '
'(SELECT 1 FROM "_httk_store_metadata" WHERE key = \'lease\')'
),
{"value": payload},
)
current = connection.execute(sqlalchemy.select(table.c.value).where(table.c.key == "lease")).scalar_one()
if current != payload:
raise RuntimeError(f"degraded SqlStore lease is held by {self._lease_description(str(current))}")
self._lease_value = payload
return
current = connection.execute(
sqlalchemy.select(table.c.value).where(table.c.key == "lease")
).scalar_one_or_none()
if current != self._lease_value:
holder = "missing lease" if current is None else self._lease_description(str(current))
raise RuntimeError(f"degraded SqlStore lease ownership was lost ({holder})")
@staticmethod
def _lease_description(value: str) -> str:
try:
parsed = json.loads(value)
acquired = datetime.datetime.fromisoformat(str(parsed["acquired_at"]))
age = datetime.datetime.now(datetime.UTC) - acquired.astimezone(datetime.UTC)
return f"{parsed['owner']!r}, age {age}"
except (KeyError, TypeError, ValueError):
return repr(value)
def _register_degraded_lifecycle_fence(self) -> None:
"""Register this store's disposal release before any mutation can start."""
generation = self._database.lifecycle_generation
self._database.add_dispose_callback(lambda: self._release_degraded_lease(generation), generation=generation)
self._lease_lifecycle_generation = generation
self._lease_callback_registered = True
@contextlib.contextmanager
def _degraded_lifecycle_guard(self) -> Iterator[None]:
if self._write_profile not in {"degraded", "bulk-fenced"}:
yield
return
assert self._lease_lifecycle_generation is not None
with self._database.lifecycle_guard(
self._lease_lifecycle_generation,
holder=f"{type(self).__name__} {self._write_profile} mutation",
):
yield
def _release_degraded_lease(self, generation: int) -> None:
# Backend.dispose may run on a different thread. Taking this same
# lock makes release wait for an in-flight mutation rather than delete
# the lease underneath its remaining ordered writes.
with self._mutation_lock:
if generation != self._lease_lifecycle_generation:
return
value = self._lease_value
try:
if value is not None:
with self._database.engine.begin() as connection:
if self._write_profile == "bulk-fenced":
from httk.store.backend.clickhouse.support import release_lease
release_lease(connection, value)
else:
connection.execute(
sqlalchemy.text(
'DELETE FROM "_httk_store_metadata" WHERE key = \'lease\' AND value = :value'
),
{"value": value},
)
finally:
self._lease_value = None
# Backend consumes callbacks for this lifecycle. A disposed
# Backend refuses late registration, so this store cannot
# mutate again; callers must construct a fresh Backend.
self._lease_callback_registered = False
def _operation_dirty_state(self) -> tuple[str, list[str]] | None:
return cast(tuple[str, list[str]] | None, getattr(self._local, "dirty_state", None))
def _begin_degraded_operation(self, connection: sqlalchemy.Connection) -> bool:
if self._write_profile != "degraded" or self._operation_dirty_state() is not None:
return False
metadata = metadata_table_for(sqlalchemy.MetaData())
for key in connection.execute(sqlalchemy.select(metadata.c.key).where(metadata.c.key.like("dirty:%"))):
self._release_unused_identity_claims(connection, str(key[0]).removeprefix("dirty:"))
self._local.dirty_state = (f"{self._lease_owner}:{uuid.uuid4().hex}", [])
return True
def _end_degraded_operation(self, connection: sqlalchemy.Connection, started: bool) -> None:
if not started:
return
value, touched = cast(tuple[str, list[str]], self._local.dirty_state)
try:
if getattr(self._local, "degraded_crashed", False):
return
for table_name in touched:
self._release_unused_identity_claims(connection, table_name)
connection.execute(
sqlalchemy.text('DELETE FROM "_httk_store_metadata" WHERE key = :key AND value = :value'),
{"key": f"dirty:{table_name}", "value": value},
)
self._after_degraded_write(f"dirty-delete:{table_name}")
finally:
del self._local.dirty_state
if hasattr(self._local, "degraded_crashed"):
del self._local.degraded_crashed
def _after_degraded_write(self, point: str) -> None:
"""Run the deterministic degraded crash hook after a durable step."""
hook = self._degraded_fault_hook
if hook is not None and hook(point):
self._local.degraded_crashed = True
raise _DegradedWriteCrash(point)
def _touch_dirty_table(self, connection: sqlalchemy.Connection, table: sqlalchemy.Table) -> None:
"""Mark a degraded operation's table before its first physical write."""
state = self._operation_dirty_state()
if state is None:
return
value, touched = state
if table.name in touched:
return
metadata = metadata_table_for(sqlalchemy.MetaData())
dirty_key = f"dirty:{table.name}"
leftover = connection.execute(
sqlalchemy.select(metadata.c.value).where(metadata.c.key == dirty_key)
).scalar_one_or_none()
if leftover is not None and leftover != value:
self._targeted_dirty_sweep(connection, table)
connection.execute(
sqlalchemy.delete(metadata).where(metadata.c.key == dirty_key, metadata.c.value == leftover)
)
connection.execute(
sqlalchemy.text(
'INSERT INTO "_httk_store_metadata" (key, value) VALUES (:key, :value) '
'ON CONFLICT(key) DO UPDATE SET value = excluded.value'
),
{"key": dirty_key, "value": value},
)
self._after_degraded_write(f"dirty-upsert:{table.name}")
touched.append(table.name)
def _targeted_dirty_sweep(self, connection: sqlalchemy.Connection, table: sqlalchemy.Table) -> None:
"""Delete only child-element residue attributable to one dirty table."""
self._release_unused_identity_claims(connection, table.name)
schemas = tuple(resolve_schema(record) for record in self._known_record_types)
graph = LogicalEdgeGraph.from_store(self, schemas)
for edge in graph.ownership():
# A dirty parent owns its declared child-element tables; a dirty
# child is itself sweepable. Reference columns never participate,
# so a nullable ``*_sid`` reference cannot be mistaken for
# ownerless child residue.
if edge.source_table == table.name or edge.target_table == table.name:
parent_name, candidate_name = edge.source_table, edge.target_table
else:
continue
if parent_name not in self._metadata.tables or candidate_name not in self._metadata.tables:
continue
parent = self._metadata.tables[parent_name]
candidate = self._metadata.tables[candidate_name]
assert edge.target_column is not None
connection.execute(
sqlalchemy.delete(candidate).where(
~sqlalchemy.exists(
sqlalchemy.select(1).where(parent.c[SID_COLUMN] == candidate.c[edge.target_column])
)
)
)
[docs]
def steal_lease(self) -> None:
"""Conditionally replace the current degraded-store writer lease.
The compare-and-swap includes the complete observed value so a stale
caller can never overwrite a newer owner.
"""
self._check_mutation_policy("steal_lease")
if self._write_profile != "degraded":
raise RuntimeError("steal_lease is available only for a degraded-profile store")
with self._mutation_lock, self._degraded_lifecycle_guard(), self._database.engine.begin() as connection:
table = metadata_table_for(sqlalchemy.MetaData())
prior = connection.execute(
sqlalchemy.select(table.c.value).where(table.c.key == "lease")
).scalar_one_or_none()
if prior is None:
self._ensure_degraded_lease(connection)
return
mine = json.dumps(
{"owner": self._lease_owner, "acquired_at": datetime.datetime.now(datetime.UTC).isoformat()},
sort_keys=True,
separators=(",", ":"),
)
changed = connection.execute(
sqlalchemy.update(table).where(table.c.key == "lease", table.c.value == prior).values(value=mine)
).rowcount
if changed != 1:
raise RuntimeError(
f"could not steal degraded SqlStore lease from {self._lease_description(str(prior))}; retry"
)
self._lease_value = mine
def _pending_table_names(self) -> set[str]:
pending = getattr(self._local, "pending_tables", None)
if pending is None:
pending = set()
self._local.pending_tables = pending
return cast(set[str], pending)
@contextlib.contextmanager
def _read_connection(self) -> Iterator[sqlalchemy.Connection]:
current = self._current_connection()
if current is not None:
yield current
return
with self._database.engine.connect() as connection:
stack = self._connection_stack()
stack.append(connection)
try:
yield connection
finally:
stack.pop()
def _candidate_metadata(self, classes: Iterable[type]) -> sqlalchemy.MetaData:
candidate = sqlalchemy.MetaData()
requested = tuple(classes)
for cls in requested:
table_for(resolve_schema(cls), candidate, store_timestamps=self._store_timestamps)
for family in self.layout.families:
if not any(record in family.records for record in requested):
continue
schemas = tuple(resolve_schema(record) for record in family.records)
for schema in schemas:
table_for(schema, candidate, store_timestamps=self._store_timestamps)
if len(schemas) > 1:
dispatch_table_for(family.name, tuple(zip(family.record_names, schemas, strict=True)), candidate)
return candidate
def _register_tables(self, classes: Iterable[type]) -> sqlalchemy.MetaData:
requested = tuple(classes)
self._known_record_types.update(requested)
candidate = self._candidate_metadata(requested)
for cls in requested:
table_for(resolve_schema(cls), self._metadata, store_timestamps=self._store_timestamps)
for family in self.layout.families:
if not any(record in family.records for record in requested):
continue
schemas = tuple(resolve_schema(record) for record in family.records)
for schema in schemas:
table_for(schema, self._metadata, store_timestamps=self._store_timestamps)
if len(schemas) > 1:
dispatch_table_for(family.name, tuple(zip(family.record_names, schemas, strict=True)), self._metadata)
return candidate
def _validate_table_names(self, names: Iterable[str]) -> None:
forbidden = sorted(
name
for name in names
if name.startswith("_httk_") and name not in self._managed_table_names and not self._is_link_table(name)
)
if forbidden:
raise ValueError(f"ordinary records may not claim reserved SqlStore table names: {', '.join(forbidden)}")
def _is_link_table(self, name: str) -> bool:
"""Whether ``name`` is a registered weak-link table (structurally, by its endpoint columns).
A weak-link table legitimately owns the reserved ``_httk_`` prefix; it is
built only from a declared ``WeakLink``, never
from a record's storage name (an ordinary ``_httk_`` storage name has no
``source_lid``/``target_lid`` columns and stays forbidden).
"""
table = self._metadata.tables.get(name)
return table is not None and SOURCE_LID_COLUMN in table.c and TARGET_LID_COLUMN in table.c
def _create_tables_for_write(self, connection: sqlalchemy.Connection, classes: Iterable[type]) -> None:
"""Register and create missing record tables for the caller's write operation.
SQLite's legacy transaction mode may commit DDL eagerly, so a failed
save can leave empty or partial declaration-shaped tables. Stamp trust
accepts that residue; the next write's ``checkfirst`` completes it.
"""
candidate = self._register_tables(classes)
candidate_names = frozenset(candidate.tables)
self._validate_table_names(candidate_names)
pending = self._pending_table_names()
missing = candidate_names - self._tables_present - pending
if missing:
pending.update(actual_table_names(connection))
missing = candidate_names - self._tables_present - pending
if missing:
candidate.create_all(connection, checkfirst=True)
# Publish only after the owning transaction commits. SQLite may
# retain empty or partial declaration-shaped tables after rollback;
# stamp trust accepts that residue and the next write completes it.
self._pending_table_names().update(missing)
def _allocate_degraded_sid(self, connection: sqlalchemy.Connection, table_name: str, count: int = 1) -> int:
"""Reserve a never-reused SQLite sid block while the writer lease is held."""
assert self._write_profile == "degraded"
if connection.dialect.name != "sqlite": # pragma: no cover - opener validation protects this
raise RuntimeError("degraded SqlStore sid allocation is supported only on SQLite")
connection.execute(
sqlalchemy.text(
'CREATE TABLE IF NOT EXISTS "_httk_sid_counters" '
'(table_name TEXT PRIMARY KEY, next_sid INTEGER NOT NULL)'
)
)
self._after_degraded_write(f"counter-table-create:{table_name}")
def initialize() -> None:
quoted = table_name.replace('"', '""')
connection.execute(
sqlalchemy.text(
'INSERT INTO "_httk_sid_counters" (table_name, next_sid) '
f'SELECT :table_name, COALESCE((SELECT MAX(sid) + 1 FROM "{quoted}"), 1) '
'WHERE NOT EXISTS (SELECT 1 FROM "_httk_sid_counters" WHERE table_name = :table_name)'
),
{"table_name": table_name},
)
self._after_degraded_write(f"counter-init:{table_name}")
for attempt in range(2):
result = connection.execute(
sqlalchemy.text(
'UPDATE "_httk_sid_counters" SET next_sid = next_sid + :count '
'WHERE table_name = :table_name RETURNING next_sid'
),
{"table_name": table_name, "count": count},
).scalar_one_or_none()
if result is not None:
self._after_degraded_write(f"counter-allocation:{table_name}")
return int(result) - count
if attempt == 0:
initialize()
raise RuntimeError(f"could not initialize degraded sid counter for table {table_name!r}")
def _missing_tables_for_read(self, classes: Iterable[type]) -> bool:
"""Register tables and report absence without issuing DDL."""
key = frozenset(classes)
candidate_names = self._candidate_names.get(key)
if candidate_names is None:
# The name set is a pure function of the class-set given the fixed
# layout and _store_timestamps; _register_tables also idempotently
# populates _metadata so later _table() lookups resolve on a hit.
candidate_names = frozenset(self._register_tables(key).tables)
self._candidate_names[key] = candidate_names
self._validate_table_names(candidate_names)
pending = self._pending_table_names()
missing = candidate_names - self._tables_present - pending
if missing:
current = self._current_connection()
if current is not None:
# Keep transaction-local catalog observations in the overlay;
# publishing them before commit would make rollback unsafe.
pending.update(actual_table_names(current))
else:
self._refresh_committed_table_names()
missing = candidate_names - self._tables_present - pending
return bool(missing)
def _refresh_committed_table_names(self) -> None:
"""Refresh the shared table cache from a connection outside this transaction."""
with self._database.engine.connect() as connection:
self._tables_present.update(actual_table_names(connection))
def _table(self, name: str) -> sqlalchemy.Table:
return self._metadata.tables[name]
# ------------------------------------------------------------------ saving
[docs]
def save(
self,
obj: Any,
*,
as_record: type | None = None,
id_series: str | None = None,
alternative_of: str | None = None,
alternative_kind: str | None = None,
links: Mapping[str, object] | None = None,
) -> int:
"""Store ``obj`` (deduplicating per its class's policy) and return its integer sid.
An opted-in domain object is projected through its exact
``__httk_storage_record__``; ``as_record`` selects an alternate record
representation explicitly. Referenced records and record-valued child
elements are saved recursively without constructing intermediate
record instances.
A content-id deduplication hit compares metadata marked with
:class:`~httk.core.storage.markers.IdentitySkip` in schema order.
Nested plans are cached per record type, and a mismatch raises
:class:`~httk.store.backend.sql.store.EntryMetadataConflictError` without replacing the row.
Passing ``alternative_of`` (a stored main entry's id) with
``alternative_kind`` saves ``obj`` as a named ALTERNATIVE representation
of that main: it copies the main's public ``id``, joins the main's
alternative group, and hashes with the group identity folded in so its
content never dedups onto the main. The main must live in ``obj``'s own
backing table and must itself be a main (not another alternative).
:param obj: The object to store.
:param as_record: The alternate record representation to use, if any.
:param id_series: Override the configured entry-id series for minted ids.
:param alternative_of: The stored main entry's id this record is an alternative of, if any.
:param alternative_kind: The alternative kind name (grammar ``[a-z][a-z0-9_]*``); required with ``alternative_of``.
:param links: Weak links to add after saving, mapping each declared link name to a target or iterable of targets; the save and every link are applied in one atomic transaction.
:return: The stored row's sid.
:raises TypeError: If ``obj`` is a cursor row that must be materialized first.
:raises ValueError: If exactly one of ``alternative_of``/``alternative_kind`` is given, the kind is malformed, or the named main is missing, in another backing table, or itself an alternative.
:raises httk.store.backend.sql.store.EntryMetadataConflictError: If a deduplication hit has conflicting metadata.
:raises httk.core.storage.identity.StorageProjectionCycleError: If projection reaches a reference cycle.
:raises RuntimeError: If a :meth:`bulk_ingest` context is currently open.
"""
if (alternative_of is None) != (alternative_kind is None):
raise ValueError("alternative_of and alternative_kind must be given together, or neither")
if alternative_kind is not None and ALTERNATIVE_KIND_PATTERN.fullmatch(alternative_kind) is None:
raise ValueError(
f"invalid alternative_kind {alternative_kind!r}; expected {ALTERNATIVE_KIND_PATTERN.pattern}"
)
if not links:
return self._save_top(
obj,
as_record=as_record,
replace_logical_id=None,
id_series=id_series,
alternative_of=alternative_of,
alternative_kind=alternative_kind,
)
self._refuse_degraded_links("save(links=...)")
with self.transaction():
sid = self._save_top(
obj,
as_record=as_record,
replace_logical_id=None,
id_series=id_series,
alternative_of=alternative_of,
alternative_kind=alternative_kind,
)
self._apply_save_links(resolve_storage_record(obj, as_record=as_record), sid, links)
return sid
def _save_top(
self,
obj: Any,
*,
as_record: type | None,
replace_logical_id: int | None,
id_series: str | None = None,
replacement_entry_id: str | None = None,
alternative_of: str | None = None,
alternative_kind: str | None = None,
replace_alt_group: int | None = None,
replace_alt_kind: str | None = None,
replace_alt_main_id: str | None = None,
) -> int:
"""Run one top-level save, optionally as a replacement carrying ``replace_logical_id``.
Alternatives enter one of two ways: a fresh :meth:`save` resolves
``alternative_of``/``alternative_kind`` against the record's own backing
table; a :meth:`replace` of an alternative carries the predecessor's
already-resolved ``replace_alt_*`` values so the replacement hashes with
the same group extras as revision 1.
"""
self._check_mutation_policy("save")
self._reject_during_bulk()
reject_cursor_proxy(obj)
record_type = resolve_storage_record(obj, as_record=as_record)
projection = SaveProjection()
timestamp_state = getattr(self._local, "store_timestamp_transaction", None)
_publish_after_commit = (
None
if timestamp_state is not None
else lambda: self._advance_store_timestamp_mark(projection.store_timestamp)
)
with self._write_connection(_publish_after_commit=_publish_after_commit) as connection:
self._create_tables_for_write(connection, (record_type,))
alt_group: int | None
alt_kind: str | None
alt_main_id: str | None
if alternative_of is not None:
alt_group, alt_main_id = self._resolve_alternative_main(connection, record_type, alternative_of)
alt_kind = alternative_kind
else:
alt_group = replace_alt_group
alt_kind = replace_alt_kind
alt_main_id = replace_alt_main_id
alt_extras: Mapping[str, object] | None = (
{"alternative_of": alt_main_id, "alternative_kind": alt_kind} if alt_kind is not None else None
)
if timestamp_state is None:
projection.store_timestamp = self._capture_store_timestamp(connection)
elif not timestamp_state["initialized"]:
projection.store_timestamp = self._capture_store_timestamp(connection)
timestamp_state["captured"] = projection.store_timestamp
timestamp_state["initialized"] = True
else:
projection.store_timestamp = timestamp_state["captured"]
sid = self._save(
connection,
record_type,
obj,
projection,
"",
top_level=True,
replace_logical_id=replace_logical_id,
id_series=id_series,
replacement_entry_id=replacement_entry_id,
alt_group=alt_group,
alt_kind=alt_kind,
alt_main_id=alt_main_id,
alt_extras=alt_extras,
)
family = self._family_for_backing(record_type)
if family is not None:
# The dispatch hash is a top-level content id, so it carries the
# same alternative-group extras as the dedup key below.
self._save_entry_dispatch(
connection, family, record_type, sid, projection.content_id(record_type, obj, extras=alt_extras)
)
# A saved lazy row of this store's own base type is registered directly
# (not via _remember, whose _sids write would hash it) so a subsequent
# default fetch of its sid returns the same proxy. Only when save()
# returned the proxy's OWN sid: a dedup="none" re-save mints a new sid
# whose row is a fresh copy, and caching the proxy (which reads its
# original row) under it would make fetch() of the new sid read the old.
identity = lazy_row_identity(obj)
if (
identity is not None
and identity[0] is self
and identity[1] == sid
and getattr(type(obj), "__httk_row_base__", None) is record_type
):
self._identity._instances[(record_type, sid)] = obj
return sid
def _save(
self,
connection: sqlalchemy.Connection,
record_type: type,
source: Any,
projection: _Projection,
path: str,
*,
top_level: bool = False,
replace_logical_id: int | None = None,
id_series: str | None = None,
replacement_entry_id: str | None = None,
alt_group: int | None = None,
alt_kind: str | None = None,
alt_main_id: str | None = None,
alt_extras: Mapping[str, object] | None = None,
) -> int:
active_key = (record_type, id(source))
if active_key in projection.active:
raise StorageProjectionCycleError(path, record_type)
projection.active.add(active_key)
try:
return self._save_active(
connection,
record_type,
source,
projection,
path,
top_level=top_level,
replace_logical_id=replace_logical_id,
id_series=id_series,
replacement_entry_id=replacement_entry_id,
alt_group=alt_group,
alt_kind=alt_kind,
alt_main_id=alt_main_id,
alt_extras=alt_extras,
)
finally:
projection.active.remove(active_key)
def _save_active(
self,
connection: sqlalchemy.Connection,
record_type: type,
source: Any,
projection: _Projection,
path: str,
*,
top_level: bool,
replace_logical_id: int | None = None,
id_series: str | None = None,
replacement_entry_id: str | None = None,
alt_group: int | None = None,
alt_kind: str | None = None,
alt_main_id: str | None = None,
alt_extras: Mapping[str, object] | None = None,
) -> int:
# The replacement lineage applies ONLY to the top-level parent row; a
# nested record reached through references/ownership keeps its own-sid
# lineage. A fresh row's logical_id is its own sid. Alternative-group
# identity is likewise a top-level concept only: nested records never
# carry a group id, kind, or extras.
replacement = replace_logical_id if top_level else None
alt_group = alt_group if top_level else None
alt_kind = alt_kind if top_level else None
alt_extras = alt_extras if top_level else None
schema = resolve_schema(record_type)
table = self._table(schema.table_name)
projected = projection.projector(record_type, source)
if record_type in self._entry_record_types:
self._validate_projected_entry_ids(record_type, projected)
if top_level and replacement_entry_id is not None:
entry_id = projected.get("id")
if entry_id is not None and entry_id != replacement_entry_id:
raise EntryIdConflictError(table.name, str(entry_id), replacement, replacement)
validation_key = (record_type, id(source))
if type(source) is record_type and validation_key not in projection.validated:
validator = vars(record_type).get("__httk_validate__")
if validator is not None:
# Bind the descriptor fetched from the class's own dict; the
# own-dict lookup (not getattr) keeps inherited validators out.
validator.__get__(None, record_type)(source)
projection.validated.add(validation_key)
key: str | None = None
if schema.dedup == "content_id":
key = projection.content_id(record_type, source, extras=alt_extras)
found = connection.execute(
sqlalchemy.select(table.c[SID_COLUMN], table.c[ROLE_COLUMN]).where(table.c[CONTENT_ID_COLUMN] == key)
).first()
if found is not None:
if self._write_profile == "degraded":
self._after_degraded_write(f"content-dedup-select:{table.name}")
sid = int(found[0])
if replacement is not None:
self._apply_replacement_collision(connection, table, sid, replacement)
# Match Mongo's order: a rejected metadata comparison is
# observational only and must never promote a dependency.
self._check_metadata(connection, record_type, sid, source, projection)
if top_level and int(found[1]) == 0:
connection.execute(
sqlalchemy.update(table).where(table.c[SID_COLUMN] == sid).values({ROLE_COLUMN: 1})
)
if self._write_profile == "degraded":
self._after_degraded_write(f"content-promotion-update:{table.name}")
self._remember(
record_type,
sid,
source,
cache_instance=type(source) is record_type and record_type not in self._entry_record_types,
)
return sid
checkpoint = len(projection.inserted)
values = self._parent_row(connection, schema, source, projected, projection, path, id_series=id_series)
enforced_entry = record_type in self._entry_record_types
if enforced_entry:
self._prepare_entry_ids(
connection,
table,
record_type,
values,
lineage=replacement,
sid=None,
id_series=id_series,
replacement_entry_id=replacement_entry_id if top_level else None,
alt_group=alt_group,
alt_kind=alt_kind,
alt_main_id=alt_main_id,
)
if schema.dedup == "by_value":
# v1 semantics: a by_value match compares the parent table's stored
# columns only; child-table contents are not part of the match.
conditions = [
table.c[name].is_(None) if value is None else table.c[name] == value for name, value in values.items()
]
statement = sqlalchemy.select(table.c[SID_COLUMN], table.c[ROLE_COLUMN])
if conditions:
statement = statement.where(*conditions)
found = connection.execute(statement.limit(1)).first()
if found is not None:
sid = int(found[0])
self._discard_inserted(connection, projection, checkpoint)
if replacement is not None:
self._apply_replacement_collision(connection, table, sid, replacement)
if top_level and int(found[1]) == 0:
connection.execute(
sqlalchemy.update(table).where(table.c[SID_COLUMN] == sid).values({ROLE_COLUMN: 1})
)
self._remember(
record_type,
sid,
source,
cache_instance=type(source) is record_type and record_type not in self._entry_record_types,
)
return sid
if self._write_profile == "degraded":
# Permanentization is deliberately sid-write-last: dependent
# records are encoded first by _parent_row, element rows are made
# durable under a reserved sid, and only then is the parent row
# written. No error path deletes this residue.
self._touch_dirty_table(connection, table)
sid = self._allocate_degraded_sid(connection, table.name)
values[SID_COLUMN] = sid
values[ROLE_COLUMN] = int(top_level)
# The sid is client-allocated pre-insert, so the lineage is known
# directly: a replacement copies its predecessor's, a fresh row uses
# its own sid.
values[LOGICAL_ID_COLUMN] = replacement if replacement is not None else sid
# alt_id is a replacement's/alternative's group (both carried in
# alt_group), else the fresh main's own sid; alt_kind is NULL for a
# main. Both are final pre-insert here (no write-after-insert).
values[ALT_ID_COLUMN] = alt_group if alt_group is not None else sid
values[ALT_KIND_COLUMN] = alt_kind
if enforced_entry:
self._prepare_entry_ids(
connection,
table,
record_type,
values,
lineage=values[LOGICAL_ID_COLUMN],
sid=sid,
id_series=id_series,
replacement_entry_id=replacement_entry_id if top_level else None,
alt_group=alt_group,
alt_kind=alt_kind,
alt_main_id=alt_main_id,
)
if projection.store_timestamp is not None:
values[STORE_TIMESTAMP_COLUMN] = projection.store_timestamp
if key is not None:
values[CONTENT_ID_COLUMN] = key
for spec in schema.fields:
if spec.role == "child":
self._insert_child_rows(
connection,
schema,
spec,
sid,
self._projected_value(record_type, source, projected, spec),
projection,
_field_path(path, spec.field),
id_series=id_series,
)
if enforced_entry:
self._claim_entry_identity(
connection,
record_type,
values,
group_id=int(values[ALT_ID_COLUMN]),
sid=sid,
)
self._after_degraded_write(f"identity-claim:{table.name}")
try:
connection.execute(sqlalchemy.insert(table).values(values))
except IntegrityError as error:
self._raise_entry_id_integrity(table, values, error)
self._after_degraded_write(f"parent-row-write:{table.name}")
projection.inserted.append((record_type, sid))
self._remember(
record_type,
sid,
source,
cache_instance=type(source) is record_type and record_type not in self._entry_record_types,
)
return sid
if key is not None:
values[CONTENT_ID_COLUMN] = key
values[ROLE_COLUMN] = int(top_level)
if projection.store_timestamp is not None:
values[STORE_TIMESTAMP_COLUMN] = projection.store_timestamp
# A replacement knows its lineage pre-insert; a fresh row's sid is
# DB-allocated by the ON CONFLICT ... RETURNING below, so it inserts
# a placeholder and is filled in the same transaction just after.
values[LOGICAL_ID_COLUMN] = replacement if replacement is not None else 0
# alt_group carries an alternative's/replacement's group id; a fresh
# main leaves a placeholder here and is filled with its own sid just
# after insert. alt_kind is final pre-insert and never updated.
values[ALT_ID_COLUMN] = alt_group if alt_group is not None else 0
values[ALT_KIND_COLUMN] = alt_kind
try:
sid, inserted = self._insert_content_row(connection, table, values, key)
except IntegrityError as error:
self._raise_entry_id_integrity(table, values, error)
if not inserted:
self._discard_inserted(connection, projection, checkpoint)
if replacement is not None:
self._apply_replacement_collision(connection, table, sid, replacement)
self._check_metadata(connection, record_type, sid, source, projection)
if top_level:
connection.execute(
sqlalchemy.update(table)
.where(table.c[SID_COLUMN] == sid, table.c[ROLE_COLUMN] == 0)
.values({ROLE_COLUMN: 1})
)
self._remember(
record_type,
sid,
source,
cache_instance=type(source) is record_type and record_type not in self._entry_record_types,
)
return sid
if replacement is None:
# The only sanctioned write-after-insert: fill the fresh row's
# own-sid lineage inside the same transaction as its insert.
update_values: dict[str, Any] = {LOGICAL_ID_COLUMN: sid}
if alt_group is None:
# A fresh main self-references; a fresh alternative keeps the
# group id inserted above (its own sid is only its lineage).
update_values[ALT_ID_COLUMN] = sid
if enforced_entry:
self._prepare_entry_ids(
connection,
table,
record_type,
values,
lineage=sid,
sid=sid,
id_series=id_series,
replacement_entry_id=None,
alt_group=alt_group,
alt_kind=alt_kind,
alt_main_id=alt_main_id,
)
update_values.update({"id": values["id"], "immutable_id": values["immutable_id"]})
try:
connection.execute(sqlalchemy.update(table).where(table.c[SID_COLUMN] == sid).values(update_values))
except IntegrityError as error:
self._raise_entry_id_integrity(table, values, error)
elif enforced_entry:
self._prepare_entry_ids(
connection,
table,
record_type,
values,
lineage=replacement,
sid=sid,
id_series=id_series,
replacement_entry_id=replacement_entry_id if top_level else None,
alt_group=alt_group,
alt_kind=alt_kind,
alt_main_id=alt_main_id,
)
try:
connection.execute(
sqlalchemy.update(table)
.where(table.c[SID_COLUMN] == sid)
.values({"id": values["id"], "immutable_id": values["immutable_id"]})
)
except IntegrityError as error:
self._raise_entry_id_integrity(table, values, error)
else:
values[ROLE_COLUMN] = int(top_level)
if projection.store_timestamp is not None:
values[STORE_TIMESTAMP_COLUMN] = projection.store_timestamp
# A replacement knows its lineage pre-insert; a fresh row's sid is
# DB-allocated, so it inserts a placeholder and is filled in the same
# transaction just after.
values[LOGICAL_ID_COLUMN] = replacement if replacement is not None else 0
# alt_group carries an alternative's/replacement's group id; a fresh
# main leaves a placeholder here and is filled with its own sid just
# after insert. alt_kind is final pre-insert and never updated.
values[ALT_ID_COLUMN] = alt_group if alt_group is not None else 0
values[ALT_KIND_COLUMN] = alt_kind
try:
result = connection.execute(sqlalchemy.insert(table).values(values))
except IntegrityError as error:
self._raise_entry_id_integrity(table, values, error)
sid = int(cast(Any, result.inserted_primary_key)[0])
if replacement is None:
# The only sanctioned write-after-insert: fill the fresh row's
# own-sid lineage inside the same transaction as its insert.
update_values = {LOGICAL_ID_COLUMN: sid}
if alt_group is None:
# A fresh main self-references; a fresh alternative keeps the
# group id inserted above (its own sid is only its lineage).
update_values[ALT_ID_COLUMN] = sid
if enforced_entry:
self._prepare_entry_ids(
connection,
table,
record_type,
values,
lineage=sid,
sid=sid,
id_series=id_series,
replacement_entry_id=None,
alt_group=alt_group,
alt_kind=alt_kind,
alt_main_id=alt_main_id,
)
update_values.update({"id": values["id"], "immutable_id": values["immutable_id"]})
try:
connection.execute(sqlalchemy.update(table).where(table.c[SID_COLUMN] == sid).values(update_values))
except IntegrityError as error:
self._raise_entry_id_integrity(table, values, error)
elif enforced_entry:
self._prepare_entry_ids(
connection,
table,
record_type,
values,
lineage=replacement,
sid=sid,
id_series=id_series,
replacement_entry_id=replacement_entry_id if top_level else None,
alt_group=alt_group,
alt_kind=alt_kind,
alt_main_id=alt_main_id,
)
try:
connection.execute(
sqlalchemy.update(table)
.where(table.c[SID_COLUMN] == sid)
.values({"id": values["id"], "immutable_id": values["immutable_id"]})
)
except IntegrityError as error:
self._raise_entry_id_integrity(table, values, error)
if enforced_entry:
group_id = alt_group if alt_group is not None else (replacement if replacement is not None else sid)
self._claim_entry_identity(connection, record_type, values, group_id=group_id, sid=sid)
projection.inserted.append((record_type, sid))
for spec in schema.fields:
if spec.role == "child":
self._insert_child_rows(
connection,
schema,
spec,
sid,
self._projected_value(record_type, source, projected, spec),
projection,
_field_path(path, spec.field),
id_series=id_series,
)
self._remember(
record_type,
sid,
source,
cache_instance=type(source) is record_type and record_type not in self._entry_record_types,
)
return sid
def _resolve_alternative_main(
self, connection: sqlalchemy.Connection, record_type: type, alternative_of: str
) -> tuple[int, str]:
"""Resolve ``alternative_of`` to its ``(group alt_id, main entry id)``.
The main must live in ``record_type``'s own backing table (alternatives
share their main's backing) and must itself be a main, not another
alternative.
:param connection: The live write connection.
:param record_type: The alternative's record class, fixing the backing table to search.
:param alternative_of: The main entry's id this record is an alternative of.
:return: The main's ``logical_id`` (the alternative group id) and its entry id.
:raises ValueError: If the main is missing, lives in another family backing table, or is itself an alternative.
"""
table = self._table(resolve_schema(record_type).table_name)
# Alternatives copy their main's id, so an id names a whole group; the
# group's MAIN is the one row with alt_kind NULL.
main = connection.execute(
sqlalchemy.select(table.c[LOGICAL_ID_COLUMN], table.c.id)
.where(table.c.id == alternative_of, table.c[ALT_KIND_COLUMN].is_(None))
.limit(1)
).one_or_none()
if main is not None:
return int(main[0]), str(main[1])
# No main here: the id may exist only as an alternative (defensive; an
# orphan without its main cannot arise through the public API), in a
# sibling backing table, or nowhere.
if (
connection.execute(
sqlalchemy.select(table.c[SID_COLUMN]).where(table.c.id == alternative_of).limit(1)
).one_or_none()
is not None
):
raise ValueError(
f"alternative_of {alternative_of!r} names an alternative, not a main; "
"alternatives of alternatives are not allowed"
)
for sibling in self._entry_family_tables(record_type):
if sibling is table:
continue
if (
connection.execute(
sqlalchemy.select(sibling.c[SID_COLUMN]).where(sibling.c.id == alternative_of).limit(1)
).one_or_none()
is not None
):
raise ValueError(
f"alternative_of {alternative_of!r} is stored in table {sibling.name!r}, but an "
f"alternative must share its main's backing table {table.name!r}"
)
raise ValueError(f"alternative_of {alternative_of!r} names no entry in table {table.name!r}")
def _prepare_entry_ids(
self,
connection: sqlalchemy.Connection,
table: sqlalchemy.Table,
record_type: type,
values: dict[str, Any],
*,
lineage: int | None,
sid: int | None,
id_series: str | None,
replacement_entry_id: str | None,
alt_group: int | None = None,
alt_kind: str | None = None,
alt_main_id: str | None = None,
) -> None:
"""Validate or mint the entry-id columns for one parent-row write.
An alternative copies its group main's id (``alt_main_id``) rather than
minting, owns id-space per its alternative group (``alt_group``) instead
of its own lineage, and stamps an alternative immutable id under
``alt_kind``. Mains keep the original per-lineage behaviour.
"""
entry_id = values.get("id")
immutable_id = values.get("immutable_id")
# An alternative copies its main's id; a replacement copies its
# predecessor's. Both forbid an explicit-but-different id on the record.
forced_entry_id = replacement_entry_id if replacement_entry_id is not None else alt_main_id
# Ownership is per alternative group, not per lineage: a fresh main's
# group is its own (yet-unknown) sid, so it falls back to the lineage
# exactly as before; an alternative/replacement carries its group id.
group_id = alt_group if alt_group is not None else lineage
if forced_entry_id is not None:
if entry_id is None:
entry_id = forced_entry_id
values["id"] = entry_id
elif entry_id != forced_entry_id:
raise EntryIdConflictError(table.name, str(entry_id), group_id, group_id)
if entry_id is not None:
# ponytail: these checks are advisory under concurrent transactions; a
# family-owned id-ownership table with unique constraints is the upgrade
# path if cross-transaction family-wide ownership must be serialized.
for sibling in self._entry_family_tables(record_type):
existing = connection.execute(
sqlalchemy.select(sibling.c[ALT_ID_COLUMN], sibling.c[SID_COLUMN])
.where(sibling.c.id == entry_id)
.limit(1)
).one_or_none()
if existing is None:
continue
if sibling is not table or (
int(existing[1]) != sid and (group_id is None or int(existing[0]) != group_id)
):
raise EntryIdConflictError(sibling.name, entry_id, int(existing[0]), group_id)
else:
if self._entry_ids is None:
raise ValueError(
f"{record_type.__name__} has no id and SqlStore(entry_ids=EntryIdScheme(...)) was not declared; "
"pass an explicit id or declare a scheme"
)
if lineage is not None:
base = self._entry_ids.base
if self._entry_ids.type_in_base:
base = f"{base}.{self._entry_record_types[record_type][0]}"
entry_id = format_entry_id(
base, id_series or self._entry_ids.series, self._entry_id_number(record_type, lineage)
)
values["id"] = entry_id
# One alternative lineage per (alt_id, alt_kind): a fresh alternative
# (no lineage of its own yet) conflicts with ANY existing row of that
# (group, kind); a replacement excludes its own predecessor lineage.
# This runs only at the pre-hash pass (sid is None), before this save's
# row exists, so it never matches itself. It closes the hole an
# explicit ``<id>~<kind>~N`` immutable id would otherwise punch through
# the incidental revision-1 immutable-id collision.
if alt_kind is not None and sid is None:
group_kind = sqlalchemy.select(table.c[LOGICAL_ID_COLUMN]).where(
table.c[ALT_ID_COLUMN] == alt_group, table.c[ALT_KIND_COLUMN] == alt_kind
)
if lineage is not None:
group_kind = group_kind.where(table.c[LOGICAL_ID_COLUMN] != lineage)
existing_lineage = connection.execute(group_kind.limit(1)).scalar_one_or_none()
if existing_lineage is not None:
raise EntryIdConflictError(table.name, str(entry_id), int(existing_lineage), lineage)
if immutable_id is not None:
for sibling in self._entry_family_tables(record_type):
existing = connection.execute(
sqlalchemy.select(sibling.c[SID_COLUMN]).where(sibling.c.immutable_id == immutable_id).limit(1)
).one_or_none()
if existing is not None and (sibling is not table or int(existing[0]) != sid):
raise EntryIdConflictError(sibling.name, immutable_id, int(existing[0]), sid)
return
if lineage is None or sid is None:
return
count_query = (
sqlalchemy.select(sqlalchemy.func.count()).select_from(table).where(table.c[LOGICAL_ID_COLUMN] == lineage)
)
if sid is not None:
count_query = count_query.where(table.c[SID_COLUMN] != sid)
revision = 1 + int(connection.execute(count_query).scalar_one())
if alt_kind is not None:
# Each alternative is its own lineage, so the per-logical_id revision
# counter is correct here; the id namespace is the kind-qualified one.
values["immutable_id"] = format_alternative_id(str(entry_id), alt_kind, revision)
else:
values["immutable_id"] = format_immutable_id(str(entry_id), revision)
@staticmethod
def _validate_projected_entry_ids(record_type: type, projected: Mapping[str, object]) -> None:
"""Validate supplied entry ids before deduplication can write dependencies."""
del record_type
entry_id = projected.get("id")
immutable_id = projected.get("immutable_id")
if entry_id is not None:
check_entry_id(cast(str, entry_id))
if immutable_id is not None:
check_immutable_id(cast(str, immutable_id))
def _entry_family_tables(self, record_type: type) -> tuple[sqlalchemy.Table, ...]:
"""Return every backing table sharing an enforced entry-id namespace."""
family = self._family_for_backing(record_type)
if family is None:
return (self._table(resolve_schema(record_type).table_name),)
return tuple(self._table(resolve_schema(backing).table_name) for backing in family.records)
@staticmethod
def _raise_entry_id_integrity(
table: sqlalchemy.Table, values: Mapping[str, Any], error: IntegrityError
) -> NoReturn:
"""Translate entry-id uniqueness failures without hiding their identifier."""
immutable_id = values.get("immutable_id")
if immutable_id is not None:
raise EntryIdConflictError(table.name, str(immutable_id), None, None) from error
raise error
def _insert_content_row(
self,
connection: sqlalchemy.Connection,
table: sqlalchemy.Table,
values: dict[str, Any],
key: str,
) -> tuple[int, bool]:
"""Insert one content-addressed row, returning the race winner safely."""
dialect = connection.dialect.name
if dialect == "sqlite":
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
statement: Any = (
sqlite_insert(table).values(values).on_conflict_do_nothing(index_elements=[CONTENT_ID_COLUMN])
)
elif dialect in {"duckdb", "postgresql"}:
from sqlalchemy.dialects.postgresql import insert as postgresql_insert
statement = (
postgresql_insert(table).values(values).on_conflict_do_nothing(index_elements=[CONTENT_ID_COLUMN])
)
else:
result = connection.execute(sqlalchemy.insert(table).values(values))
return int(cast(Any, result.inserted_primary_key)[0]), True
result = connection.execute(statement.returning(table.c[SID_COLUMN]))
inserted_sid = result.scalar_one_or_none()
if inserted_sid is not None:
return int(inserted_sid), True
found = connection.execute(
sqlalchemy.select(table.c[SID_COLUMN]).where(table.c[CONTENT_ID_COLUMN] == key)
).scalar_one()
return int(found), False
def _apply_replacement_collision(
self, connection: sqlalchemy.Connection, table: sqlalchemy.Table, hit_sid: int, predecessor_logical_id: int
) -> None:
"""Enforce :meth:`replace`'s dedup collision policy against a deduplicated hit row.
A replacement whose content deduplicates onto an existing row is an
idempotent no-op when that row shares the predecessor's lineage, and a
:class:`EntryReplacementError` when it belongs to a different one.
"""
existing = int(
connection.execute(
sqlalchemy.select(table.c[LOGICAL_ID_COLUMN]).where(table.c[SID_COLUMN] == hit_sid)
).scalar_one()
)
if existing != predecessor_logical_id:
raise EntryReplacementError(table.name, predecessor_logical_id, existing)
def _parent_row(
self,
connection: sqlalchemy.Connection,
schema: TableSchema,
source: Any,
projected: Mapping[str, object],
projection: _Projection,
path: str,
*,
id_series: str | None,
) -> dict[str, Any]:
"""Encode projected parent columns, saving referenced records recursively."""
def resolve_sid(record_type: type, value: Any, field_path: str) -> int:
return self._save(
connection, record_type, value, projection, field_path, top_level=False, id_series=id_series
)
return _encode_parent_row(schema, source, projected, path, resolve_sid)
@staticmethod
def _projected_value(record_type: type, source: Any, projected: Mapping[str, object], spec: FieldSpec) -> Any:
if spec.field in projected:
return projected[spec.field]
if spec.derived:
try:
return getattr(source, spec.field)
except AttributeError:
raise TypeError(
f"projecting {type(source).__name__} as {record_type.__name__} requires the source "
f"to expose derived stored property {spec.field!r}"
) from None
raise ValueError(f"projection for {type(source).__name__} omitted stored field {spec.field!r}")
def _insert_child_rows(
self,
connection: sqlalchemy.Connection,
schema: TableSchema,
spec: FieldSpec,
sid: int,
value: Any,
projection: _Projection,
path: str,
*,
id_series: str | None,
) -> None:
assert spec.child is not None
def resolve_sid(record_type: type, element: Any, element_path: str) -> int:
return self._save(
connection, record_type, element, projection, element_path, top_level=False, id_series=id_series
)
rows = _encode_child_rows(schema, spec, sid, value, path, resolve_sid)
if rows:
table = self._table(spec.child.table_name)
self._touch_dirty_table(connection, table)
connection.execute(sqlalchemy.insert(table), rows)
self._after_degraded_write(f"child-row-write:{table.name}")
# ------------------------------------------------------------------ fetching
[docs]
def fetch[T](self, cls: type[T], sid: int, *, eager: bool = False) -> T:
"""Reconstruct the ``cls`` instance stored under ``sid``.
By default a lazy row is returned: the parent row is loaded now, but
every child, reference and derived field decodes only when first
accessed (recursively, so a lazy record's children are lazy too). Pass
``eager=True`` to fully materialize the base dataclass up front — the
behaviour required for records that must outlive the fetching
transaction, connection or engine.
Repeated default fetches of a live ``(class, sid)`` return the same
object; a live materialized instance takes precedence over creating a
new proxy. Mixing eager and lazy access may hand out two distinct but
equal objects when a caller still holds the older one, and internal
cache maintenance (a failed write, dedup compensation) may
re-materialize a later fetch — strict ``is`` identity across arbitrary
call sequences is not promised.
Raises :class:`KeyError` (carrying the class and sid) when no such
parent row exists. A missing table therefore has the same result as a
missing row. Under the lazy default, abnormal external deletion of a
*referenced* row surfaces at attribute access as
:class:`~httk.store.backend.sql.rows.StaleResultError`; abnormally deleted
*child* rows are indistinguishable from an empty sequence.
:param cls: The storable class to reconstruct.
:param sid: The stored row identifier.
:param eager: Whether to fully materialize the record instead of returning a lazy row.
:return: The reconstructed instance.
:raises KeyError: If no row exists for ``cls`` and ``sid``.
"""
if eager:
with self._read_connection() as connection:
return cast(T, self._fetch(connection, cls, sid))
return cast(T, self._fetch_lazy(cls, sid))
[docs]
def fetch_many[T](self, cls: type[T], sids: Sequence[int], *, eager: bool = False) -> list[T]:
"""Reconstruct every ``cls`` instance stored under ``sids`` in one batch.
The batched counterpart of :meth:`fetch`: child-element and reference
reads are shared across the requested rows instead of re-queried per
sid. By default lazy rows are returned; they share one
:class:`~httk.store.backend.sql.rows.RowHydrator`, so a deferred child or
reference read stays chunk-batched (one SELECT per child table per
500-row chunk on first touch) exactly as the eager path batches it,
merely deferred. Pass ``eager=True`` to fully materialize every record
up front.
Mirroring :meth:`fetch`, a live cached object (proxy or materialized)
is returned for any ``(class, sid)`` still alive without touching the
database (so a fully cached call issues no SQL); the remaining rows
share one connection. Memory is O(``len(sids)``) — every chunk stays
pinned for the batch — so callers pass bounded pages.
:param cls: The storable class to reconstruct.
:param sids: The stored row identifiers to reconstruct.
:param eager: Whether to fully materialize each record instead of returning lazy rows.
:return: The reconstructed instances in ``sids`` order.
:raises KeyError: If any requested row does not exist.
"""
if not eager:
return cast(list[T], self._fetch_many_lazy(cls, sids))
resolved = [int(sid) for sid in sids]
instances: dict[int, Any] = {}
missing: list[int] = []
for sid in resolved:
cached = self._identity._instances.get((cls, sid))
# A lazy proxy hit is not a materialized instance; re-materialize it.
if cached is None or type(cached) is not cls:
missing.append(sid)
else:
instances[sid] = cached
if missing:
with self._read_connection():
try:
# materialize() populates the identity map via _remember.
hydrated = RowHydrator(self, cls, missing).materialize_many()
except StaleResultError as error:
raise KeyError(cls, tuple(missing)) from error
instances.update(zip(missing, hydrated, strict=True))
return cast(list[T], [instances[sid] for sid in resolved])
def _fetch_lazy(self, cls: type, sid: int) -> Any:
sid = int(sid)
cached = self._identity._instances.get((cls, sid))
if cached is not None:
return cached
with self._read_connection():
try:
# row() loads the parent chunk (the KeyError-bearing SELECT);
# child/reference fields decode lazily on first access.
proxy = RowHydrator(self, cls, (sid,)).row(sid)
except StaleResultError as error:
raise KeyError(cls, sid) from error
# Register the proxy directly (never via _remember, whose _sids write
# would hash the row and force-decode its hash fields).
self._identity._instances[(cls, sid)] = proxy
return proxy
def _fetch_many_lazy(self, cls: type, sids: Sequence[int]) -> list[Any]:
resolved = [int(sid) for sid in sids]
instances: dict[int, Any] = {}
missing: list[int] = []
for sid in resolved:
cached = self._identity._instances.get((cls, sid))
if cached is None:
missing.append(sid)
else:
instances[sid] = cached
if missing:
# One hydrator over every miss keeps deferred child/reference reads
# chunk-batched; one-hydrator-per-sid would reintroduce N+1.
with self._read_connection():
hydrator = RowHydrator(self, cls, missing)
try:
for sid in missing:
proxy = hydrator.row(sid)
self._identity._instances[(cls, sid)] = proxy
instances[sid] = proxy
except StaleResultError as error:
raise KeyError(cls, tuple(missing)) from error
return [instances[sid] for sid in resolved]
def _fetch_result(self, connection: sqlalchemy.Connection, cls: type, sid: int, *, eager: bool) -> Any:
"""Hydrate one sid eagerly or lazily, reusing the current stacked connection."""
if eager:
return self._fetch(connection, cls, sid)
return self._fetch_lazy(cls, sid)
[docs]
def fetch_by_content_id[T](self, cls: type[T], key: str, *, eager: bool = False) -> T | None:
"""Return the ``cls`` instance whose content identity is ``key``, or None if not stored.
Only classes with the ``"content_id"`` dedup policy carry a content
identity column; :class:`~httk.store.backend.schema.SchemaError` is raised
for any other class. A lazy row is returned by default; pass
``eager=True`` to fully materialize it.
:param cls: The storable class to search.
:param key: The content identity to find.
:param eager: Whether to fully materialize the record instead of returning a lazy row.
:return: The stored instance, or ``None`` when no row matches.
:raises httk.store.backend.schema.SchemaError: If the class does not use content-id deduplication.
"""
schema = resolve_schema(cls)
if schema.dedup != "content_id":
raise SchemaError(
f"{cls.__name__} has dedup policy {schema.dedup!r}; only classes with the "
f"'content_id' policy have a content identity column"
)
with self._read_connection() as connection:
if self._missing_tables_for_read((cls,)):
return None
table = self._table(schema.table_name)
found = connection.execute(
sqlalchemy.select(table.c[SID_COLUMN]).where(table.c[CONTENT_ID_COLUMN] == key)
).first()
if found is None:
return None
return cast(T, self._fetch_result(connection, cls, int(found[0]), eager=eager))
[docs]
def fetch_entry(self, family_cls: type, content_id: str, *, eager: bool = False) -> object | None:
"""Return the concrete configured record for an entry-family content identity.
The result is the actual frozen record class, not the family protocol.
A single-record family can query that record directly; only
multi-record families use their reserved one-of-many dispatch table,
whose constraint permits exactly one backing sid per content identity.
A lazy row is returned by default; pass ``eager=True`` to fully
materialize it.
:param family_cls: The configured entry-family class.
:param content_id: The entry content identity to find.
:param eager: Whether to fully materialize the record instead of returning a lazy row.
:return: The concrete stored record, or ``None`` when no row matches.
:raises ValueError: If ``family_cls`` is not configured for this store.
:raises EntryDispatchIntegrityError: If a dispatch row is inconsistent with its backing row.
"""
family = next((item for item in self.layout.families if item.family is family_cls), None)
if family is None:
raise ValueError(f"{family_cls.__name__} is not a configured entry family in this SqlStore")
with self._read_connection() as connection:
if self._missing_tables_for_read(family.records):
return None
if len(family.records) == 1:
backing = family.records[0]
schema = resolve_schema(backing)
table = self._table(schema.table_name)
sid = connection.execute(
sqlalchemy.select(table.c[SID_COLUMN]).where(table.c[CONTENT_ID_COLUMN] == content_id)
).scalar_one_or_none()
return None if sid is None else self._fetch_result(connection, backing, int(sid), eager=eager)
table = self._table(entry_dispatch_table_name(family.name))
row = (
connection.execute(sqlalchemy.select(table).where(table.c[DISPATCH_CONTENT_ID_COLUMN] == content_id))
.mappings()
.one_or_none()
)
if row is None:
for backing in family.records:
backing_table = self._table(resolve_schema(backing).table_name)
found = connection.execute(
sqlalchemy.select(backing_table.c[SID_COLUMN])
.where(backing_table.c[CONTENT_ID_COLUMN] == content_id)
.limit(1)
).first()
if found is not None:
raise EntryDispatchIntegrityError(
f"entry dispatch {family.name!r} is missing for stored content_id {content_id!r}"
)
return None
backing, sid = self._dispatch_target(family, row, content_id)
backing_table = self._table(resolve_schema(backing).table_name)
backing_content_id = connection.execute(
sqlalchemy.select(backing_table.c[CONTENT_ID_COLUMN]).where(backing_table.c[SID_COLUMN] == sid)
).scalar_one_or_none()
if backing_content_id != content_id:
raise EntryDispatchIntegrityError(
f"entry dispatch {family.name!r} maps content_id {content_id!r} to backing sid {sid} "
f"whose content_id is {backing_content_id!r}"
)
return self._fetch_result(connection, backing, sid, eager=eager)
[docs]
def sid_of(self, obj: Any, *, as_record: type | None = None) -> int | None:
"""Return this store's sid for ``obj``'s record identity, if present.
:param obj: The object whose stored identity should be looked up.
:param as_record: The alternate record representation to use, if any.
:return: The stored sid, or ``None`` when no matching row is known.
"""
record_type = resolve_storage_record(obj, as_record=as_record)
lazy_identity = lazy_row_identity(obj)
if is_lazy_row(obj) and record_type is resolve_storage_record(obj):
return lazy_identity[1] if lazy_identity is not None and lazy_identity[0] is self else None
try:
cached = self._identity._sids.get(obj, {}).get(record_type)
except TypeError:
cached = None
if cached is None:
cached = self._identity._sids_by_identity.get((record_type, id(obj)))
if cached is not None:
return cached
schema = resolve_schema(record_type)
if schema.dedup != "content_id":
return cached
projection = _Projection()
key = projection.content_id(record_type, obj)
with self._read_connection() as connection:
if self._missing_tables_for_read((record_type,)):
return None
table = self._table(schema.table_name)
found = connection.execute(
sqlalchemy.select(table.c[SID_COLUMN]).where(table.c[CONTENT_ID_COLUMN] == key)
).scalar_one_or_none()
if found is None:
return None
sid = int(found)
self._remember(
record_type,
sid,
obj,
cache_instance=type(obj) is record_type and record_type not in self._entry_record_types,
)
return sid
[docs]
def searcher(self, *, as_of: object = None, only_latest: bool = False, only_main_alt: bool = True) -> SqlSearcher:
"""Return a new :class:`~httk.store.backend.sql.searcher.SqlSearcher` querying this store.
The searcher runs on this store's read path — inside an open
:meth:`transaction` block it sees uncommitted writes — and
reconstructs matched objects as lazy rows, decoding each field on first
access exactly as the lazy default of :meth:`fetch` does.
:param as_of: Optional historic cutoff in canonical timestamp form.
:param only_latest: Whether root variables are restricted to the latest row of each
``logical_id`` lineage by sid (bounded by ``as_of`` when given). Reference/child
variables stay unfiltered. Does not require ``store_timestamps=True``.
:param only_main_alt: Whether root variables are restricted to mains (``alt_kind IS NULL``),
hiding named alternatives. Defaults to ``True``; pass ``False`` to reveal alternatives.
:return: A new SQL searcher bound to this store.
"""
if as_of is not None:
if not self._store_timestamps:
raise ValueError("as_of queries require SqlStore(store_timestamps=True)")
ns_operand_to_store_units(as_of, self._store_timestamp_resolution)
return SqlSearcher(self, as_of=as_of, only_latest=only_latest, only_main_alt=only_main_alt)
[docs]
def fsck(
self,
*,
repair: bool = True,
collect_garbage: bool = True,
repair_conflicts: bool = False,
clamp_future_timestamps: bool = False,
known_types: tuple[type, ...] = (),
exclusive: bool = False,
) -> Any:
"""Repair dispatches and reclaim permanentization residue.
Only tables attributable to the persisted layout or ``known_types``
are swept; unrelated application tables make collection refuse.
"""
self._check_mutation_policy("fsck")
self._reject_during_bulk()
from httk.store.backend.sql.fsck import run_fsck
return run_fsck(
self,
repair=repair,
collect_garbage=collect_garbage,
repair_conflicts=repair_conflicts,
clamp_future_timestamps=clamp_future_timestamps,
known_types=known_types,
exclusive=exclusive,
)
[docs]
def stored_property_plan(self, family: type) -> Any:
"""Return the wire-form SQL stored-property plan for one configured entry family.
This method is a serving edge: like the entry providers, it serves the
family's definition in OPTIMADE wire form (``served_form()``), so a
prefixed family whose ``__httk_stored_properties__`` keys are served
names plans correctly. ``served_form()`` is idempotent for standard
(unprefixed) families, so their plans are byte-identical.
:param family: The logical entry-family class to plan.
:return: The validated SQL stored-property plan.
"""
from httk.store.backend.sql.stored_properties import stored_property_sql_plan
return stored_property_sql_plan(self, family, served=_served_definition(family))
[docs]
def referring(self, cls: type, *, field: str, to: Any, eager: bool = False) -> list[Any]:
"""Return all stored ``cls`` instances whose reference field ``field`` points at ``to``.
``field`` must be a reference field of ``cls`` targeting ``to``'s class
(:class:`~httk.store.backend.schema.SchemaError` otherwise), and ``to`` must
be known to this store — saved or fetched through it — else
:class:`ValueError` is raised. Results are ordered by sid. Lazy rows are
returned by default (batched over the matched sids); pass ``eager=True``
to fully materialize them.
:param cls: The storable class whose references should be searched.
:param field: The reference field to match.
:param to: The stored target instance.
:param eager: Whether to fully materialize the records instead of returning lazy rows.
:return: The referring stored instances ordered by sid.
:raises httk.store.backend.schema.SchemaError: If ``field`` is not a compatible reference field.
:raises ValueError: If ``to`` is not known to this store.
"""
schema = resolve_schema(cls)
spec = schema.field(field)
if spec.role != "reference":
raise SchemaError(f"{cls.__name__}.{field} is not a reference field (its role is {spec.role!r})")
assert spec.target is not None
if not isinstance(to, spec.target):
raise SchemaError(f"{cls.__name__}.{field} references {spec.target.__name__}, not {type(to).__name__}")
sid = self.sid_of(to)
if sid is None:
raise ValueError(f"the {type(to).__name__} instance has not been stored or fetched through this store")
with self._read_connection() as connection:
if self._missing_tables_for_read((cls,)):
return []
table = self._table(schema.table_name)
found = connection.execute(
sqlalchemy.select(table.c[SID_COLUMN])
.where(table.c[spec.columns[0].name] == sid)
.order_by(table.c[SID_COLUMN])
).all()
found_sids = [int(row[0]) for row in found]
if eager:
return [self._fetch(connection, cls, referring_sid) for referring_sid in found_sids]
return self._fetch_many_lazy(cls, found_sids)
[docs]
def replace(
self,
predecessor: Any,
obj: Any,
*,
id_series: str | None = None,
links: Mapping[str, object] | None = None,
) -> int:
"""Store ``obj`` as a logical replacement of ``predecessor`` and return its sid.
The saved row copies ``predecessor``'s ``logical_id`` (its lineage
identity) instead of starting a fresh one, so both rows share the
lineage :meth:`history` walks. Nothing is updated or deleted: plain
:meth:`fetch` and :meth:`searcher` queries keep returning both rows, and
the lineage's latest row is simply the one with the highest sid.
``predecessor`` need not itself be the latest row of its lineage —
replacing an already-replaced row is allowed and extends the same
lineage. ``obj`` is saved through the ordinary :meth:`save` path, so its
dedup policy, timestamp capture, identity caching and entry-family
dispatch all behave exactly as they do there.
If ``obj``'s content deduplicates onto an existing row (under the
``"content_id"`` or ``"by_value"`` policies), that row's lineage is
compared with ``predecessor``'s: an equal lineage (including ``obj``
equalling ``predecessor`` itself) is an idempotent no-op returning the
existing sid, while a different lineage raises
:class:`EntryReplacementError`.
``links`` optionally adds weak links to the replacement's lineage, as in
:meth:`save`; the replace and every link are applied in one atomic
transaction.
:param predecessor: The stored instance or lazy proxy being replaced; it must have been stored or fetched through this store.
:param obj: The replacement object to store.
:param id_series: Override the configured entry-id series when an id must be minted.
:param links: Weak links to add, mapping each declared link name to a target or iterable of targets.
:return: The stored replacement row's sid.
:raises ValueError: If ``predecessor`` is not known to this store, or ``obj``'s record table differs from ``predecessor``'s.
:raises EntryReplacementError: If ``obj`` deduplicates onto a row from a different lineage.
"""
if not links:
sid, _obj_type = self._replace_core(predecessor, obj, id_series)
return sid
self._refuse_degraded_links("replace(links=...)")
with self.transaction():
sid, obj_type = self._replace_core(predecessor, obj, id_series)
self._apply_save_links(obj_type, sid, links)
return sid
def _replace_core(self, predecessor: Any, obj: Any, id_series: str | None) -> tuple[int, type]:
self._check_mutation_policy("replace")
self._reject_during_bulk()
predecessor_sid = self.sid_of(predecessor)
if predecessor_sid is None:
raise ValueError(
f"the {type(predecessor).__name__} instance has not been stored or fetched through this store"
)
predecessor_type = resolve_storage_record(predecessor)
obj_type = resolve_storage_record(obj)
predecessor_table = resolve_schema(predecessor_type).table_name
obj_table = resolve_schema(obj_type).table_name
if obj_table != predecessor_table:
raise ValueError(
f"cannot replace a record stored in table {predecessor_table!r} with a "
f"{obj_type.__name__} record stored in table {obj_table!r}"
)
with self._read_connection() as connection:
self._missing_tables_for_read((predecessor_type,))
table = self._table(predecessor_table)
predecessor_alt_kind: str | None = None
alt_main_id: str | None = None
if obj_type in self._entry_record_types:
predecessor_row = connection.execute(
sqlalchemy.select(
table.c[LOGICAL_ID_COLUMN], table.c.id, table.c[ALT_ID_COLUMN], table.c[ALT_KIND_COLUMN]
).where(table.c[SID_COLUMN] == predecessor_sid)
).one()
predecessor_logical_id = int(predecessor_row[0])
predecessor_entry_id = str(predecessor_row[1])
predecessor_alt_id = int(predecessor_row[2])
predecessor_alt_kind = None if predecessor_row[3] is None else str(predecessor_row[3])
if predecessor_alt_kind is not None:
# An alternative's replacement must hash with the same group
# extras as revision 1: recover the group main's id, which is
# lineage-constant, so any group-main row's id serves.
alt_main_id = str(
connection.execute(
sqlalchemy.select(table.c.id)
.where(
table.c[LOGICAL_ID_COLUMN] == predecessor_alt_id,
table.c[ALT_KIND_COLUMN].is_(None),
)
.limit(1)
).scalar_one()
)
else:
predecessor_plain = connection.execute(
sqlalchemy.select(table.c[LOGICAL_ID_COLUMN], table.c[ALT_ID_COLUMN]).where(
table.c[SID_COLUMN] == predecessor_sid
)
).one()
predecessor_logical_id = int(predecessor_plain[0])
predecessor_alt_id = int(predecessor_plain[1])
predecessor_entry_id = None
sid = self._save_top(
obj,
as_record=None,
replace_logical_id=predecessor_logical_id,
id_series=id_series,
replacement_entry_id=predecessor_entry_id,
replace_alt_group=predecessor_alt_id,
replace_alt_kind=predecessor_alt_kind,
replace_alt_main_id=alt_main_id,
)
return sid, obj_type
# ------------------------------------------------------------------ weak links
def _link_spec(self, source_cls: type, name: str) -> LinkSpec:
"""Resolve the source class's declared weak link named ``name``."""
schema = resolve_schema(source_cls)
for spec in schema.links:
if spec.name == name:
return spec
declared = ", ".join(link.name for link in schema.links) or "none"
raise SchemaError(f"{source_cls.__name__} declares no weak link named {name!r} (declared links: {declared})")
def _refuse_degraded_links(self, operation: str) -> None:
if self._write_profile == "degraded":
raise RuntimeError(
f"{operation} is refused for the degraded SqlStore write profile; weak links require the "
"transactional write-after-insert path"
)
def _capture_link_timestamp(self, connection: sqlalchemy.Connection, timestamp_state: Any) -> int | None:
"""Capture a store timestamp, sharing one value across an open transaction (as save does)."""
if timestamp_state is None:
return self._capture_store_timestamp(connection)
if not timestamp_state["initialized"]:
captured = self._capture_store_timestamp(connection)
timestamp_state["captured"] = captured
timestamp_state["initialized"] = True
return captured
return cast(int | None, timestamp_state["captured"])
def _lid_from_sid(self, connection: sqlalchemy.Connection, schema: TableSchema, sid: int) -> int:
"""The lineage id (``logical_id``) of the row ``sid`` in ``schema``'s parent table."""
table = self._table(schema.table_name)
found = connection.execute(
sqlalchemy.select(table.c[LOGICAL_ID_COLUMN]).where(table.c[SID_COLUMN] == sid)
).scalar_one_or_none()
if found is None:
raise ValueError(f"no {schema.cls.__name__} row exists for sid {sid}")
return int(found)
def _lid_of(self, connection: sqlalchemy.Connection, cls: type, obj: Any) -> int:
"""The lineage id of ``obj`` in ``cls``'s table; raises if ``obj`` is not stored."""
sid = self.sid_of(obj, as_record=cls)
if sid is None:
raise ValueError(f"the {type(obj).__name__} instance has not been stored or fetched through this store")
return self._lid_from_sid(connection, resolve_schema(cls), sid)
def _link_target_lid(
self, connection: sqlalchemy.Connection, spec: LinkSpec, target: Any, source_cls: type, name: str
) -> int:
"""Validate ``target``'s type against ``spec`` and resolve its lineage id."""
target_cls = resolve_storage_record(target)
if resolve_schema(target_cls).table_name != resolve_schema(spec.target).table_name:
raise TypeError(
f"weak link {name!r} on {source_cls.__name__} expects a {spec.target.__name__} target, "
f"got {target_cls.__name__}"
)
return self._lid_of(connection, target_cls, target)
def _latest_link_by_lineage(
self, connection: sqlalchemy.Connection, link_table: sqlalchemy.Table, source_lid: int, target_lid: int
) -> dict[int, tuple[int, int]]:
"""Map each pair lineage (``logical_id``) to its latest ``(sid, retracted)``."""
rows = connection.execute(
sqlalchemy.select(
link_table.c[LOGICAL_ID_COLUMN], link_table.c[SID_COLUMN], link_table.c[RETRACTED_COLUMN]
).where(link_table.c[SOURCE_LID_COLUMN] == source_lid, link_table.c[TARGET_LID_COLUMN] == target_lid)
).all()
latest: dict[int, tuple[int, int]] = {}
for logical_id, sid, retracted in rows:
lineage = int(logical_id)
previous = latest.get(lineage)
if previous is None or int(sid) > previous[0]:
latest[lineage] = (int(sid), int(retracted))
return latest
def _insert_link_row(
self,
connection: sqlalchemy.Connection,
link_table: sqlalchemy.Table,
*,
logical_id: int | None,
source_lid: int,
target_lid: int,
retracted: int,
store_timestamp: int | None,
) -> None:
"""Append one link revision. ``logical_id=None`` founds a fresh own-sid lineage."""
values: dict[str, Any] = {
SOURCE_LID_COLUMN: source_lid,
TARGET_LID_COLUMN: target_lid,
RETRACTED_COLUMN: retracted,
LOGICAL_ID_COLUMN: logical_id if logical_id is not None else 0,
}
if store_timestamp is not None:
values[STORE_TIMESTAMP_COLUMN] = store_timestamp
result = connection.execute(sqlalchemy.insert(link_table).values(values))
if logical_id is None:
# The sanctioned write-after-insert: a fresh lineage's logical_id is
# its own DB-allocated sid, filled inside the same transaction.
sid = int(cast(Any, result.inserted_primary_key)[0])
connection.execute(
sqlalchemy.update(link_table).where(link_table.c[SID_COLUMN] == sid).values({LOGICAL_ID_COLUMN: sid})
)
def _do_link(
self,
connection: sqlalchemy.Connection,
spec: LinkSpec,
source_lid: int,
target_lid: int,
store_timestamp: int | None,
) -> None:
"""Idempotently assert the pair ``(source_lid, target_lid)`` is linked."""
link_table = self._table(spec.table_name)
latest = self._latest_link_by_lineage(connection, link_table, source_lid, target_lid)
if any(retracted == 0 for _sid, retracted in latest.values()):
return # some lineage is already live: no-op
self._insert_link_row(
connection,
link_table,
# All lineages retracted: revive the max-logical_id one. None founds a fresh lineage.
logical_id=max(latest) if latest else None,
source_lid=source_lid,
target_lid=target_lid,
retracted=0,
store_timestamp=store_timestamp,
)
def _do_unlink(
self,
connection: sqlalchemy.Connection,
spec: LinkSpec,
source_lid: int,
target_lid: int,
store_timestamp: int | None,
) -> None:
"""Retract every live lineage of the pair ``(source_lid, target_lid)``."""
link_table = self._table(spec.table_name)
latest = self._latest_link_by_lineage(connection, link_table, source_lid, target_lid)
for lineage, (_sid, retracted) in latest.items():
if retracted == 0:
self._insert_link_row(
connection,
link_table,
logical_id=lineage,
source_lid=source_lid,
target_lid=target_lid,
retracted=1,
store_timestamp=store_timestamp,
)
def _apply_save_links(self, source_cls: type, source_sid: int, links: Mapping[str, object]) -> None:
"""Add each declared link for the just-saved row, inside the open save transaction."""
connection = self._current_connection()
assert connection is not None, "_apply_save_links runs inside an open transaction"
timestamp_state = getattr(self._local, "store_timestamp_transaction", None)
store_timestamp = timestamp_state["captured"] if timestamp_state is not None else None
source_lid = self._lid_from_sid(connection, resolve_schema(source_cls), source_sid)
for name, targets in links.items():
spec = self._link_spec(source_cls, name)
for target in _iter_link_targets(targets):
target_lid = self._link_target_lid(connection, spec, target, source_cls, name)
self._do_link(connection, spec, source_lid, target_lid, store_timestamp)
[docs]
def link(self, source: Any, name: str, target: Any) -> None:
"""Assert a weak link named ``name`` from ``source``'s lineage to ``target``'s lineage.
Weak links live in a dedicated append-only link table and bind
*lineages*: both endpoints always resolve to their latest revision. The
operation is idempotent per the pair ``(source lineage, target
lineage)`` — a live pair is a no-op, a retracted pair is revived, an
absent pair founds a fresh link lineage — and duplicate pair lineages
(from concurrent writers) are tolerated: the pair is live if any lineage
is live.
:param source: The stored source instance or lazy proxy declaring the link.
:param name: The declared weak-link name.
:param target: The stored target instance whose type matches the link declaration.
:return: None.
:raises httk.store.backend.schema.SchemaError: If ``source``'s class declares no link named ``name``.
:raises TypeError: If ``target``'s type does not match the link's declared target.
:raises ValueError: If ``source`` or ``target`` is not stored in this store.
:raises RuntimeError: If the store uses the degraded write profile or a bulk-ingest context is open.
"""
self._check_mutation_policy("link")
self._reject_during_bulk()
self._refuse_degraded_links("link")
source_cls = resolve_storage_record(source)
spec = self._link_spec(source_cls, name)
timestamp_state = getattr(self._local, "store_timestamp_transaction", None)
captured: list[int | None] = [None]
publish = None if timestamp_state is not None else (lambda: self._advance_store_timestamp_mark(captured[0]))
with self._write_connection(_publish_after_commit=publish) as connection:
self._create_tables_for_write(connection, (source_cls,))
store_timestamp = self._capture_link_timestamp(connection, timestamp_state)
captured[0] = store_timestamp
source_lid = self._lid_of(connection, source_cls, source)
target_lid = self._link_target_lid(connection, spec, target, source_cls, name)
self._do_link(connection, spec, source_lid, target_lid, store_timestamp)
[docs]
def unlink(self, source: Any, name: str, target: Any) -> None:
"""Retract the weak link named ``name`` from ``source``'s lineage to ``target``'s lineage.
Retracts every live lineage of the pair (duplicate-tolerant); an absent
or already-retracted pair is a no-op. Retraction appends a revision — no
row is deleted, and history/``as_of`` still see the earlier live rows.
:param source: The stored source instance or lazy proxy declaring the link.
:param name: The declared weak-link name.
:param target: The stored target instance whose type matches the link declaration.
:return: None.
:raises httk.store.backend.schema.SchemaError: If ``source``'s class declares no link named ``name``.
:raises TypeError: If ``target``'s type does not match the link's declared target.
:raises ValueError: If ``source`` or ``target`` is not stored in this store.
:raises RuntimeError: If the store uses the degraded write profile or a bulk-ingest context is open.
"""
self._check_mutation_policy("unlink")
self._reject_during_bulk()
self._refuse_degraded_links("unlink")
source_cls = resolve_storage_record(source)
spec = self._link_spec(source_cls, name)
timestamp_state = getattr(self._local, "store_timestamp_transaction", None)
captured: list[int | None] = [None]
publish = None if timestamp_state is not None else (lambda: self._advance_store_timestamp_mark(captured[0]))
with self._write_connection(_publish_after_commit=publish) as connection:
self._create_tables_for_write(connection, (source_cls,))
store_timestamp = self._capture_link_timestamp(connection, timestamp_state)
captured[0] = store_timestamp
source_lid = self._lid_of(connection, source_cls, source)
target_lid = self._link_target_lid(connection, spec, target, source_cls, name)
self._do_unlink(connection, spec, source_lid, target_lid, store_timestamp)
[docs]
def linked(self, source: Any, name: str, *, eager: bool = False) -> tuple[Any, ...]:
"""Return the latest revisions of the targets currently linked from ``source`` under ``name``.
Targets are deduplicated by lineage and ordered by first-link order (the
link lineage's root ``logical_id``, ascending; stable across
retract+relink). Each returned object is the latest revision of its
target lineage, hydrated through the ordinary fetch machinery (a lazy row
by default, fully materialized when ``eager`` is true).
:param source: The stored source instance or lazy proxy declaring the link.
:param name: The declared weak-link name.
:param eager: Whether to fully materialize each target instead of returning a lazy row.
:return: The linked targets' latest revisions, deduplicated and ordered by first-link order.
:raises httk.store.backend.schema.SchemaError: If ``source``'s class declares no link named ``name``.
:raises ValueError: If ``source`` is not stored in this store.
"""
source_cls = resolve_storage_record(source)
spec = self._link_spec(source_cls, name)
with self._read_connection() as connection:
if self._missing_tables_for_read((source_cls,)):
raise ValueError(
f"the {type(source).__name__} instance has not been stored or fetched through this store"
)
source_lid = self._lid_of(connection, source_cls, source)
return self._linked_by_lid(spec, source_lid, eager=eager)
def _linked_by_lid(
self, spec: LinkSpec, source_lid: int, *, eager: bool = False, as_of: object = None
) -> tuple[Any, ...]:
"""Latest live-linked targets of ``source_lid`` under ``spec`` (the query core of :meth:`linked`).
Shared by :meth:`linked` (which resolves the source lineage id first),
the fetched-row ``.links`` accessor (which already holds the lineage id
from the row's chunk), and a link-set search output (which passes its
searcher's ``as_of``). Targets are deduplicated by lineage and ordered by
first-link order; each is the latest revision of its target lineage. When
``as_of`` is given, both the link rows and each target's latest revision
are bounded by that cutoff — the same semantics the predicate path
applies to link rows and targets.
"""
target_cls = spec.target
as_of_units = (
None if as_of is None else ns_operand_to_store_units(as_of, cast(int, self._store_timestamp_resolution))
)
with self._read_connection() as connection:
link_table = self._table(spec.table_name)
link_query = sqlalchemy.select(
link_table.c[TARGET_LID_COLUMN],
link_table.c[LOGICAL_ID_COLUMN],
link_table.c[SID_COLUMN],
link_table.c[RETRACTED_COLUMN],
).where(link_table.c[SOURCE_LID_COLUMN] == source_lid)
if as_of_units is not None:
link_query = link_query.where(link_table.c[STORE_TIMESTAMP_COLUMN] <= as_of_units)
rows = connection.execute(link_query).all()
# Latest revision per link lineage, then per target keep the smallest
# live lineage root (first-link order); a target is live if any of
# its lineages is live.
latest: dict[int, tuple[int, int, int]] = {} # logical_id -> (sid, retracted, target_lid)
for target_lid, logical_id, sid, retracted in rows:
lineage = int(logical_id)
previous = latest.get(lineage)
if previous is None or int(sid) > previous[0]:
latest[lineage] = (int(sid), int(retracted), int(target_lid))
live_root: dict[int, int] = {} # target_lid -> smallest live lineage root
for lineage, (_sid, retracted, target_lid) in latest.items():
if retracted != 0:
continue
existing = live_root.get(target_lid)
if existing is None or lineage < existing:
live_root[target_lid] = lineage
ordered_target_lids = [
target_lid for target_lid, _root in sorted(live_root.items(), key=lambda item: (item[1], item[0]))
]
if not ordered_target_lids or self._missing_tables_for_read((target_cls,)):
return ()
target_table = self._table(resolve_schema(target_cls).table_name)
target_sids: list[int] = []
for target_lid in ordered_target_lids:
max_sid_query = sqlalchemy.select(sqlalchemy.func.max(target_table.c[SID_COLUMN])).where(
target_table.c[LOGICAL_ID_COLUMN] == target_lid
)
if as_of_units is not None:
max_sid_query = max_sid_query.where(target_table.c[STORE_TIMESTAMP_COLUMN] <= as_of_units)
max_sid = connection.execute(max_sid_query).scalar_one_or_none()
if max_sid is not None: # a None max is a dangling link; fsck reports it
target_sids.append(int(max_sid))
return tuple(self.fetch_many(target_cls, target_sids, eager=eager))
[docs]
def history(self, obj: Any) -> tuple[Any, ...]:
"""Return every record in ``obj``'s replacement lineage, oldest first.
The lineage is the set of rows sharing ``obj``'s ``logical_id`` — the
fresh record that started it and every :meth:`replace` of it — ordered
by sid ascending (the fresh record first, the latest replacement last).
Records are reconstructed through the same machinery as :meth:`fetch`,
lazily by default.
:param obj: A stored instance or lazy proxy whose lineage to walk; it must have been stored or fetched through this store.
:return: The lineage's records ordered by ascending sid.
:raises ValueError: If ``obj`` is not known to this store.
"""
obj_sid = self.sid_of(obj)
if obj_sid is None:
raise ValueError(f"the {type(obj).__name__} instance has not been stored or fetched through this store")
record_type = resolve_storage_record(obj)
schema = resolve_schema(record_type)
with self._read_connection() as connection:
if self._missing_tables_for_read((record_type,)):
return ()
table = self._table(schema.table_name)
# A lineage is one logical_id; alternatives own distinct logical_ids,
# so a main and its alternatives never share a history() walk.
logical_id = int(
connection.execute(
sqlalchemy.select(table.c[LOGICAL_ID_COLUMN]).where(table.c[SID_COLUMN] == obj_sid)
).scalar_one()
)
found = connection.execute(
sqlalchemy.select(table.c[SID_COLUMN])
.where(table.c[LOGICAL_ID_COLUMN] == logical_id)
.order_by(table.c[SID_COLUMN])
).all()
return tuple(self._fetch_many_lazy(record_type, [int(row[0]) for row in found]))
def _fetch(self, connection: sqlalchemy.Connection, cls: type, sid: int) -> Any:
sid = int(sid)
cached = self._identity._instances.get((cls, sid))
# A lazy proxy registered under this key must never satisfy an eager
# fetch: treat a proxy hit as a miss and re-materialize the base type.
# Materialization then _remember()s the base instance, overwriting the
# proxy's cache slot (materialized-wins precedence).
if cached is not None and type(cached) is cls:
return cached
# The hydrator owns exact decoding and child/reference batching; this
# path still materializes and validates the real base dataclass.
try:
instance = RowHydrator(self, cls, (sid,)).materialize(sid)
except StaleResultError:
raise KeyError(cls, sid) from None
self._remember(cls, sid, instance)
return instance
def _family_for_backing(self, record_type: type) -> EntryFamilyLayout | None:
for family in self.layout.families:
if any(backing is record_type for backing in family.records):
return family
return None
def _save_entry_dispatch(
self,
connection: sqlalchemy.Connection,
family: EntryFamilyLayout,
backing: type,
sid: int,
key: str,
) -> None:
if len(family.records) == 1:
return
table = self._table(entry_dispatch_table_name(family.name))
column_name = backing_dispatch_column_name(family.record_names[family.records.index(backing)])
existing = (
connection.execute(sqlalchemy.select(table).where(table.c[DISPATCH_CONTENT_ID_COLUMN] == key))
.mappings()
.one_or_none()
)
if existing is not None:
found_backing, found_sid = self._dispatch_target(family, existing, key)
if found_backing is not backing or found_sid != sid:
raise EntryDispatchIntegrityError(
f"entry dispatch {family.name!r} maps content_id {key!r} to a conflicting backing row"
)
return
values = {DISPATCH_CONTENT_ID_COLUMN: key, column_name: sid}
inserted = self._insert_dispatch_row(connection, table, values)
if inserted:
if self._write_profile == "degraded":
self._after_degraded_write(f"dispatch-row-write:{table.name}")
return
# ``ON CONFLICT DO NOTHING`` keeps SQLite, DuckDB and PostgreSQL
# transactions usable after either uniqueness conflict. Diagnose the
# content-id path first, then the per-backing UNIQUE sid path.
existing = (
connection.execute(sqlalchemy.select(table).where(table.c[DISPATCH_CONTENT_ID_COLUMN] == key))
.mappings()
.one_or_none()
)
if existing is not None:
found_backing, found_sid = self._dispatch_target(family, existing, key)
if found_backing is backing and found_sid == sid:
return
raise EntryDispatchIntegrityError(
f"entry dispatch {family.name!r} maps content_id {key!r} to a conflicting backing row"
)
sid_owner = (
connection.execute(sqlalchemy.select(table).where(table.c[column_name] == sid)).mappings().one_or_none()
)
if sid_owner is not None:
owner_content_id = str(sid_owner[DISPATCH_CONTENT_ID_COLUMN])
raise EntryDispatchIntegrityError(
f"entry dispatch {family.name!r} already maps backing sid {sid} to content_id {owner_content_id!r}, "
f"not {key!r}"
)
raise EntryDispatchIntegrityError(
f"entry dispatch {family.name!r} declined content_id {key!r} without a discoverable conflicting row"
)
@staticmethod
def _insert_dispatch_row(
connection: sqlalchemy.Connection,
table: sqlalchemy.Table,
values: Mapping[str, object],
) -> bool:
"""Insert one dispatch association, safely detecting every uniqueness conflict."""
dialect = connection.dialect.name
if dialect == "sqlite":
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
statement: Any = sqlite_insert(table).values(values).on_conflict_do_nothing()
elif dialect in {"duckdb", "postgresql"}:
from sqlalchemy.dialects.postgresql import insert as postgresql_insert
statement = postgresql_insert(table).values(values).on_conflict_do_nothing()
else:
result = connection.execute(sqlalchemy.insert(table).values(values))
return result.rowcount == 1
result = connection.execute(statement.returning(table.c[DISPATCH_CONTENT_ID_COLUMN]))
return result.scalar_one_or_none() is not None
@staticmethod
def _dispatch_target(
family: EntryFamilyLayout,
row: Any,
content_id: str,
) -> tuple[type, int]:
populated: list[tuple[type, int]] = []
for backing_name, backing in zip(family.record_names, family.records, strict=True):
value = row[backing_dispatch_column_name(backing_name)]
if value is not None:
populated.append((backing, int(value)))
if len(populated) != 1:
raise EntryDispatchIntegrityError(
f"entry dispatch {family.name!r} has {len(populated)} backing rows for content_id {content_id!r}"
)
return populated[0]
def _check_metadata(
self,
connection: sqlalchemy.Connection,
record_type: type,
sid: int,
source: Any,
projection: _Projection,
) -> None:
plan = _metadata_plan(record_type)
if plan is None:
return
self._check_metadata_at(connection, record_type, sid, source, projection, record_type.__name__, plan)
if record_type in self._entry_record_types:
table = self._table(resolve_schema(record_type).table_name)
row = connection.execute(sqlalchemy.select(table).where(table.c[SID_COLUMN] == sid)).mappings().one()
self._claim_entry_identity(
connection,
record_type,
dict(row),
group_id=int(row[ALT_ID_COLUMN]),
sid=sid,
)
def _check_metadata_at(
self,
connection: sqlalchemy.Connection,
record_type: type,
sid: int,
source: Any,
projection: _Projection,
path: str,
plan: _MetadataPlan | None = None,
) -> None:
plan = _metadata_plan(record_type) if plan is None else plan
if plan is None:
return
schema = resolve_schema(record_type)
row = self._metadata_parent_row(connection, record_type, sid, plan, projection)
values = projection.projector(record_type, source)
skipped_specs = {spec.field for spec in plan.skipped_specs}
skipped_nested = {spec.field for spec in plan.skipped_nested}
descend_specs = {spec.field for spec in plan.descend_specs}
for spec in schema.fields:
if spec.derived:
continue
field_path = _field_path(path, spec.field)
if spec.field in skipped_specs:
incoming = values[spec.field]
existing = decode_field(self, schema, spec, sid, row)
if spec.field in {"id", "immutable_id"} and incoming is None:
continue
if not _metadata_scalar_equal(incoming, existing):
raise EntryMetadataConflictError(
f"metadata conflict for {field_path}: stored {existing!r}, received {incoming!r}"
)
elif spec.field in skipped_nested:
self._check_metadata_nested(
connection, schema, row, sid, spec, values[spec.field], projection, field_path, True
)
elif spec.field in descend_specs:
self._check_metadata_nested(
connection, schema, row, sid, spec, values[spec.field], projection, field_path, False
)
def _metadata_parent_row(
self,
connection: sqlalchemy.Connection,
record_type: type,
sid: int,
plan: _MetadataPlan,
projection: _Projection,
) -> Mapping[str, Any]:
key = (record_type, int(sid))
cached = projection.metadata_rows.get(key)
if cached is not None:
return cached
schema = resolve_schema(record_type)
table = self._table(schema.table_name)
specs = (*plan.skipped_specs, *plan.skipped_nested, *plan.descend_specs)
columns: list[sqlalchemy.Column[Any]] = []
seen: set[str] = set()
for spec in specs:
if spec.role == "child":
if spec.optional:
name = f"{spec.field}_present"
if name not in seen:
columns.append(table.c[name])
seen.add(name)
continue
for column in spec.columns:
if column.name not in seen:
columns.append(table.c[column.name])
seen.add(column.name)
if not columns:
row: Mapping[str, Any] = {}
projection.metadata_rows[key] = row
return row
result = connection.execute(sqlalchemy.select(*columns).where(table.c[SID_COLUMN] == sid)).mappings().first()
if result is None:
raise KeyError(record_type, sid)
row = cast(Mapping[str, Any], result)
projection.metadata_rows[key] = row
return row
def _metadata_child_value(
self,
connection: sqlalchemy.Connection,
schema: TableSchema,
sid: int,
spec: FieldSpec,
parent_row: Mapping[str, Any],
projection: _Projection,
) -> Any:
key = (schema.cls, int(sid), spec.field)
cached = projection.metadata_children.get(key, _MISSING_METADATA)
if cached is not _MISSING_METADATA:
return cached
if spec.optional and not parent_row[f"{spec.field}_present"]:
projection.metadata_children[key] = None
return None
assert spec.child is not None
table = self._table(spec.child.table_name)
parent_column = f"{schema.table_name}_sid"
index_column = f"{spec.field}_index"
columns = tuple(table.c[column.name] for column in spec.child.element_columns)
rows = connection.execute(
sqlalchemy.select(*columns).where(table.c[parent_column] == sid).order_by(table.c[index_column])
).mappings()
decoded = [self._metadata_child_element(spec, cast(Mapping[str, Any], row)) for row in rows]
if spec.shape is not None:
value: Any = FracVector(decoded)
elif typing.get_origin(spec.python_type) is tuple:
value = tuple(decoded)
else:
value = decoded
projection.metadata_children[key] = value
return value
def _metadata_child_records(
self,
connection: sqlalchemy.Connection,
spec: FieldSpec,
stored: Any,
) -> Any:
if stored is None:
return None
assert spec.target is not None
records = [self._fetch(connection, spec.target, int(stored_sid)) for stored_sid in stored]
return tuple(records) if typing.get_origin(spec.python_type) is tuple else records
@staticmethod
def _metadata_child_element(spec: FieldSpec, row: Mapping[str, Any]) -> Any:
assert spec.child is not None
if spec.target is not None:
return int(row[spec.child.element_columns[0].name])
if spec.shape is not None:
assert spec.shape is not None
return decode_fracvector_exact(row[f"{spec.field}_exact"], 1, spec.shape.cols).to_fractions()[0]
if spec.codec_name is not None:
return codec_named(spec.codec_name).decode(tuple(row[column.name] for column in spec.child.element_columns))
return row[spec.child.element_columns[0].name]
def _check_metadata_nested(
self,
connection: sqlalchemy.Connection,
schema: TableSchema,
parent_row: Mapping[str, Any],
sid: int,
spec: FieldSpec,
incoming: Any,
projection: _Projection,
path: str,
compare_content: bool,
) -> None:
if spec.role == "reference":
assert spec.target is not None
stored_sid = parent_row[spec.columns[0].name]
if incoming is None or stored_sid is None:
if incoming is not None or stored_sid is not None:
if compare_content:
stored = None if stored_sid is None else self._fetch(connection, spec.target, int(stored_sid))
raise EntryMetadataConflictError(
f"metadata conflict for {path}: stored {stored!r}, received {incoming!r}"
)
raise EntryMetadataConflictError(f"metadata conflict for {path}")
return
self._check_metadata_target(
connection, spec.target, int(stored_sid), incoming, projection, path, compare_content
)
return
stored = self._metadata_child_value(connection, schema, sid, spec, parent_row, projection)
if spec.target is None:
if not _metadata_scalar_equal(incoming, stored):
raise EntryMetadataConflictError(
f"metadata conflict for {path}: stored {stored!r}, received {incoming!r}"
)
return
if incoming is None or stored is None:
if incoming is not stored:
if compare_content:
existing = self._metadata_child_records(connection, spec, stored)
raise EntryMetadataConflictError(
f"metadata conflict for {path}: stored {existing!r}, received {incoming!r}"
)
raise EntryMetadataConflictError(f"metadata conflict for {path}")
return
if len(incoming) != len(stored):
if compare_content:
existing = self._metadata_child_records(connection, spec, stored)
raise EntryMetadataConflictError(
f"metadata conflict for {path}: stored {existing!r}, received {incoming!r}"
)
raise EntryMetadataConflictError(f"metadata conflict for {path}")
for index, (incoming_item, stored_sid) in enumerate(zip(incoming, stored, strict=True)):
item_path = f"{path}[{index}]"
self._check_metadata_target(
connection, spec.target, int(stored_sid), incoming_item, projection, item_path, compare_content
)
def _check_metadata_target(
self,
connection: sqlalchemy.Connection,
record_type: type,
sid: int,
source: Any,
projection: _Projection,
path: str,
compare_content: bool,
) -> None:
if compare_content:
stored_content_id = self._metadata_content_id(connection, record_type, sid, projection)
incoming_content_id = projection.content_id(record_type, source)
if incoming_content_id != stored_content_id:
stored = self._fetch(connection, record_type, sid)
raise EntryMetadataConflictError(
f"metadata conflict for {path}: stored {stored!r}, received {source!r}"
)
plan = _metadata_plan(record_type)
if plan is not None:
self._check_metadata_at(connection, record_type, sid, source, projection, path, plan)
def _metadata_content_id(
self,
connection: sqlalchemy.Connection,
record_type: type,
sid: int,
projection: _Projection,
) -> str:
key = (record_type, int(sid))
cached = projection.metadata_content_ids.get(key)
if cached is not None:
return cached
schema = resolve_schema(record_type)
table = self._table(schema.table_name)
if schema.dedup == "content_id":
stored_content_id = connection.execute(
sqlalchemy.select(table.c[CONTENT_ID_COLUMN]).where(table.c[SID_COLUMN] == sid)
).scalar_one()
result = str(stored_content_id)
else:
result = projection.content_id(record_type, self._fetch(connection, record_type, sid))
projection.metadata_content_ids[key] = result
return result
# ------------------------------------------------------------------ identity caches
def _discard_inserted(
self,
connection: sqlalchemy.Connection,
projection: _Projection,
checkpoint: int,
) -> None:
# In the autocommit permanentization profile, pre-parent residue is the
# deliberate crash-recovery input for fsck; only transactional saves
# compensate dependency inserts after a dedup hit.
if self._write_profile == "degraded":
return
if checkpoint == len(projection.inserted):
return
for record_type, sid in reversed(projection.inserted[checkpoint:]):
schema = resolve_schema(record_type)
for spec in schema.fields:
if spec.role == "child":
assert spec.child is not None
table = self._table(spec.child.table_name)
connection.execute(sqlalchemy.delete(table).where(table.c[f"{schema.table_name}_sid"] == sid))
table = self._table(schema.table_name)
connection.execute(sqlalchemy.delete(table).where(table.c[SID_COLUMN] == sid))
self._release_unused_identity_claims(connection, table.name)
del projection.inserted[checkpoint:]
# No rollback token here: dedup compensation runs inside save(), where no
# user code can perform a deferred lazy read on this thread-local
# connection before the cache clear completes.
self._clear_identity_caches()
def _clear_identity_caches(self) -> None:
self._identity._clear_identity_caches()
def _remember(self, cls: type, sid: int, obj: Any, *, cache_instance: bool = True) -> None:
self._identity._remember(cls, sid, obj, cache_instance=cache_instance)
def _iter_link_targets(targets: object) -> Iterator[Any]:
"""Yield one target, or each target of an iterable-of-targets (list/tuple/set)."""
if isinstance(targets, (list, tuple, set, frozenset)):
yield from targets
else:
yield targets
def _as_fixed_tensor(schema: TableSchema, spec: FieldSpec, shape: Shape, value: Any) -> FracVector:
"""Normalize a fixed-shape field value to a ``(rows, cols)`` FracVector, validating its shape."""
tensor = FracVector(value)
dim = tensor.dim
if dim == (shape.rows, shape.cols):
return tensor
if shape.rows == 1 and dim == (shape.cols,):
return FracVector._of((tensor.noms,), tensor.denom)
raise ValueError(
f"{schema.cls.__name__}.{spec.field}: expected a FracVector of shape ({shape.rows}, {shape.cols}), got {dim}"
)
def _tensor_rows(schema: TableSchema, spec: FieldSpec, shape: Shape, value: Any) -> list[FracVector]:
"""The rows of a variable-rows (``Shape(0, c)``) field value, each as a ``(c,)`` FracVector."""
if value is None:
return []
tensor = FracVector(value)
dim = tensor.dim
if dim == () or dim == (0,):
return []
if len(dim) != 2 or dim[1] != shape.cols:
raise ValueError(
f"{schema.cls.__name__}.{spec.field}: expected a FracVector with {shape.cols} columns per row, "
f"got shape {dim}"
)
rows = cast(tuple[tuple[int, ...], ...], tensor.noms) # dim was validated two-dimensional above
return [FracVector._of(noms_row, tensor.denom) for noms_row in rows]
def _field_path(path: str, field: str) -> str:
return f"{path}.{field}" if path else field
def _encode_promoted_descendants(
schema: TableSchema,
source: Any,
projected: Mapping[str, object],
path: str,
sid: int,
promoted: frozenset[type],
resolve_sid: SidResolver,
*,
references: bool,
) -> None:
"""Resolve only branches which can contain a requested bulk-promoted record."""
def reaches(candidate: type) -> bool:
pending = [candidate]
visited: set[type] = set()
while pending:
current = pending.pop()
if current in promoted:
return True
if current not in visited:
visited.add(current)
pending.extend(resolve_schema(current).referenced_classes())
return False
for spec in schema.fields:
if spec.target is None or not reaches(spec.target):
continue
value = SqlStore._projected_value(schema.cls, source, projected, spec)
if spec.role == "child":
assert spec.child is not None
_encode_child_rows(schema, spec, sid, value, _field_path(path, spec.field), resolve_sid)
elif references and value is not None:
resolve_sid(spec.target, value, _field_path(path, spec.field))
def _encode_parent_row(
schema: TableSchema,
source: Any,
projected: Mapping[str, object],
path: str,
resolve_sid: SidResolver,
) -> dict[str, Any]:
"""Encode projected parent columns, resolving referenced records through ``resolve_sid``.
Connection-free counterpart of :meth:`SqlStore._parent_row`: every branch
but the reference one is pure, and references defer to ``resolve_sid`` so
the same encoder serves both recursive-save and bulk-allocation callers.
:param schema: The parent record's resolved table schema.
:param source: The instance (or projection source) being encoded.
:param projected: The projected field mapping for ``source``.
:param path: The projection path prefix used for diagnostics.
:param resolve_sid: The callback assigning a sid to each referenced record.
:return: The encoded parent-table column values.
"""
values: dict[str, Any] = {}
for spec in schema.fields:
if spec.role == "child":
if spec.optional:
values[f"{spec.field}_present"] = (
SqlStore._projected_value(schema.cls, source, projected, spec) is not None
)
continue
value = SqlStore._projected_value(schema.cls, source, projected, spec)
if value is None:
for column in spec.columns:
values[column.name] = None
elif spec.role == "scalar":
values[spec.columns[0].name] = value
elif spec.role == "encoded":
assert spec.codec_name is not None
encoded = codec_named(spec.codec_name).encode(value)
for column, part in zip(spec.columns, encoded, strict=True):
values[column.name] = part
elif spec.role == "fixed_array":
assert spec.shape is not None
tensor = _as_fixed_tensor(schema, spec, spec.shape, value)
for i, part in enumerate(encode_fracvector_floats(tensor)):
values[f"{spec.field}_{i}"] = part
values[f"{spec.field}_exact"] = encode_fracvector_exact(tensor)
else: # reference
assert spec.target is not None
values[spec.columns[0].name] = resolve_sid(spec.target, value, _field_path(path, spec.field))
return values
def _encode_child_rows(
schema: TableSchema,
spec: FieldSpec,
sid: int,
value: Any,
path: str,
resolve_sid: SidResolver,
) -> list[dict[str, Any]]:
"""Build the child-table rows for one child field, resolving element records through ``resolve_sid``.
Connection-free counterpart of the row-building loop in
:meth:`SqlStore._insert_child_rows`: tensor and codec branches are pure, and
storable-element references defer to ``resolve_sid``; the caller owns the
``executemany`` on the returned rows.
:param schema: The parent record's resolved table schema.
:param spec: The child field specification being encoded.
:param sid: The parent row's sid, stamped into every child row.
:param value: The child field value (a sequence, tensor, or ``None``).
:param path: The projection path prefix used for diagnostics.
:param resolve_sid: The callback assigning a sid to each referenced element record.
:return: The encoded child-table rows in element order.
"""
assert spec.child is not None
parent_column = f"{schema.table_name}_sid"
index_column = f"{spec.field}_index"
rows: list[dict[str, Any]] = []
if spec.shape is not None:
for position, row_tensor in enumerate(_tensor_rows(schema, spec, spec.shape, value)):
row: dict[str, Any] = {parent_column: sid, index_column: position}
for i, part in enumerate(encode_fracvector_floats(row_tensor)):
row[f"{spec.field}_{i}"] = part
row[f"{spec.field}_exact"] = encode_fracvector_exact(row_tensor)
rows.append(row)
else:
codec = codec_named(spec.codec_name) if spec.codec_name is not None else None
for position, element in enumerate(value if value is not None else ()):
row = {parent_column: sid, index_column: position}
if spec.target is not None:
row[spec.child.element_columns[0].name] = resolve_sid(spec.target, element, f"{path}[{position}]")
elif codec is not None:
for column, part in zip(spec.child.element_columns, codec.encode(element), strict=True):
row[column.name] = part
else:
row[spec.child.element_columns[0].name] = element
rows.append(row)
return rows
def _metadata_scalar_equal(left: Any, right: Any) -> bool:
if isinstance(left, list | tuple) or isinstance(right, list | tuple):
if type(left) is not type(right) or len(left) != len(right):
return False
return all(_metadata_scalar_equal(left_item, right_item) for left_item, right_item in zip(left, right))
if isinstance(left, datetime.datetime) and isinstance(right, datetime.datetime):
left_aware = left.utcoffset() is not None
right_aware = right.utcoffset() is not None
if left_aware != right_aware:
return False
if left_aware:
return left.astimezone(datetime.UTC) == right.astimezone(datetime.UTC)
return bool(left == right)