From 419b09c104f0854637a3ace70b8a5e644ec4a885 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Tue, 13 Oct 2026 04:34:49 +0000 Subject: [PATCH 1/2] fix: key-taking operations become free functions that honour dol wrappers (#1) Every non-dunder method of the stores that took a key (replace, batch, partition, add_container, add_database, CosmosDatabase.delete, CosmosAccount.delete) ran bound to the leaf under a dol wrapper and acted on the caller's unmapped key: under a key codec, replace overwrote another document, delete(force=True) emptied and dropped the wrong container or database, and partition() stripped every wrapper. They move to cosmodol.ops (re-exported): replace_item, batch_items, partition, add_container, delete_container, add_database, delete_database. Each takes the store first and maps the key through every layer with dol.inner_most_key (values through _data_of_obj). partition refuses a wrapped store unless given wrap=. batch_items takes (op, key[, value]) so its ids are mappable. The old method names raise AttributeError naming the replacement; a conformance test inventories every public method. Also: delete_container(force=True) now deletes items with the container's real partition key (the old cascade read a non-existent _partitionKey field), and FakeContainerProxy gains execute_item_batch. Same design as s3dol ADR-0011. Closes #1 Co-Authored-By: Claude Opus 5.5 --- cosmodol/__init__.py | 19 + cosmodol/errors.py | 25 ++ cosmodol/ops.py | 340 +++++++++++++++++ cosmodol/stores.py | 63 ++-- cosmodol/testing.py | 34 ++ cosmodol/tests/test_integration_emulator.py | 27 +- cosmodol/tests/test_ops.py | 399 ++++++++++++++++++++ cosmodol/tests/test_stores.py | 8 +- cosmodol/trees.py | 120 ++---- misc/docs/architecture.md | 14 +- misc/docs/design_decisions.md | 6 +- 11 files changed, 916 insertions(+), 139 deletions(-) create mode 100644 cosmodol/ops.py create mode 100644 cosmodol/tests/test_ops.py diff --git a/cosmodol/__init__.py b/cosmodol/__init__.py index b800532..4d48a81 100644 --- a/cosmodol/__init__.py +++ b/cosmodol/__init__.py @@ -57,6 +57,16 @@ CosmosAccount, CosmosDatabase, ) +from cosmodol.ops import ( + BATCH_OPS, + add_container, + add_database, + batch_items, + delete_container, + delete_database, + partition, + replace_item, +) from cosmodol.recipes import ( cosmos_store, strip_system_fields, @@ -94,6 +104,15 @@ # trees "CosmosAccount", "CosmosDatabase", + # key-taking operations (free functions; honour dol wrappers) + "BATCH_OPS", + "add_container", + "add_database", + "batch_items", + "delete_container", + "delete_database", + "partition", + "replace_item", # recipes "cosmos_store", "strip_system_fields", diff --git a/cosmodol/errors.py b/cosmodol/errors.py index 011d1bb..b6839a8 100644 --- a/cosmodol/errors.py +++ b/cosmodol/errors.py @@ -141,3 +141,28 @@ def wrapper(*args, **kwargs): return wrapper return decorator + + +class MovedToFunction: + """Class attribute marking a method that moved to a free function. + + Instance access raises an ``AttributeError`` naming the replacement, so + ``hasattr(store, name)`` is ``False`` and the error says where to go. Class access + returns the marker itself, which keeps ``dol``'s class-wrapping (``delegate_to`` + walks ``dir(cls)``) working. + """ + + def __init__(self, replacement: str, *, reason: str = ""): + self.replacement = replacement + self.reason = reason + + def __set_name__(self, owner, name): + self.name = name + + def __get__(self, instance, owner=None): + if instance is None: + return self + raise AttributeError( + f"{type(instance).__name__}.{self.name} was removed; use " + f"`{self.replacement}`. {self.reason}".rstrip() + ) diff --git a/cosmodol/ops.py b/cosmodol/ops.py new file mode 100644 index 0000000..e656931 --- /dev/null +++ b/cosmodol/ops.py @@ -0,0 +1,340 @@ +"""Key-taking operations, as free functions over the caller's (possibly wrapped) store. + +Why free functions and not methods: ``dol`` wraps stores by delegation, so a *method* of a +wrapped store runs bound to the inner leaf and receives the caller's **outer**, unmapped key +(i2mint/cosmodol#1, i2mint/dol#83). Under a key codec that silently replaced, batch-wrote, +cascade-deleted or created the wrong resource. A free function is handed the outer store +itself, so it can resolve the key through every wrapping layer before touching Cosmos: + +>>> from dol import KeyCodecs +>>> from cosmodol import CosmosItems, replace_item +>>> from cosmodol.testing import FakeContainerProxy +>>> c = FakeContainerProxy(partition_key_path="/_pk") +>>> view = KeyCodecs.prefixed("app_")( +... CosmosItems(c, partition_key_value="t", partition_key_path="/_pk") +... ) +>>> view["doc1"] = {"v": 1} +>>> _ = c.upsert_item({"id": "doc1", "_pk": "t", "v": "someone else's"}) +>>> _ = replace_item(view, "doc1", {"v": 2}) +>>> sorted((i, d["v"]) for (_pk, i), d in c._data.items()) +[('app_doc1', 2), ('doc1', "someone else's")] + +The same design is s3dol's ADR-0011 (keyed capabilities leave the store's method surface; +destructive ones become free functions). The removed methods now raise an +``AttributeError`` that names their replacement here. + +Caveats, stated rather than hidden (ADR-0011 D3a): + +- Keys are resolved with ``dol.inner_most_key``, i.e. each wrapping layer's + ``_id_of_key``. Every ``dol``-shipped wrapper (``wrap_kvs``, ``KeyCodecs``, ``Pipe``, + ``filt_iter``, ``cached_keys``) does this right; a hand-rolled ``__getattr__`` + passthrough layer can make it apply a transform twice. +- Values (``replace_item``, ``batch_items``) go through each layer's ``_data_of_obj``, + which is what ``wrap_kvs(data_of_obj=...)`` and ``ValueCodecs`` install. A key-aware + ``preset=`` write hook is **not** followed — such a hook only runs inside + ``__setitem__``. +""" + +from __future__ import annotations + +from typing import Any, Callable, Iterable, Optional + +from azure.cosmos import PartitionKey +from dol import inner_most_key +from dol.dig import inner_most + +from cosmodol.base import batch as _batch, point_replace +from cosmodol.errors import ( + ContainerNotFoundError, + DatabaseNotFoundError, + validate_cosmos_id, +) +from cosmodol.stores import CosmosItems, CosmosPartitionedItems +from cosmodol.trees import CosmosAccount, CosmosDatabase + + +# --------------------------------------------------------------------------- +# Resolution: outer store -> leaf, outer key -> leaf key, outer value -> leaf value +# --------------------------------------------------------------------------- + + +def _leaf_of(store, leaf_type, *, fn: str): + """The cosmodol leaf at the bottom of ``store``'s ``.store`` chain. + + Validates the first argument (a free function accepts any object, where a method + could not be called on the wrong one). + """ + leaf = store + while not isinstance(leaf, leaf_type) and hasattr(leaf, "store"): + leaf = leaf.store + if not isinstance(leaf, leaf_type): + raise TypeError( + f"{fn}() needs a {leaf_type.__name__} (or a dol wrapper around one); " + f"got {type(store).__name__}." + ) + return leaf + + +def _leaf_key(store, leaf, k): + """``k`` as the leaf sees it: mapped through every wrapping layer of ``store``. + + Replaces, never composes with, any leaf-level key mapping (``inner_most_key`` walks + the whole chain). + """ + if store is leaf: + return k + return inner_most_key(store, k) + + +def _leaf_value(store, leaf, v): + if store is leaf: + return v + return inner_most(store, v, "_data_of_obj", default=v) + + +# --------------------------------------------------------------------------- +# Item-level operations (CosmosItems) +# --------------------------------------------------------------------------- + + +def replace_item(store, k, v, *, etag: Optional[str] = None) -> dict: + """Full replace of the existing item at key ``k``, optionally ETag-conditional. + + Unlike ``store[k] = v`` (an upsert), this fails with ``ItemNotFoundError`` if the item + does not exist, and with a precondition error if ``etag`` no longer matches. + + Args: + store: A ``CosmosItems``, or any ``dol`` wrapper around one. ``k`` and ``v`` are in + the *caller's* key and value space; they are mapped down to the leaf. + k: The key, as ``store[k]`` would take it. + v: The value, as ``store[k] = v`` would take it. + etag: If given, only replace when the stored item's ``_etag`` still equals it. + + Returns: + The stored document as Cosmos returned it (leaf form; system fields stripped if the + leaf strips them). + """ + leaf = _leaf_of(store, CosmosItems, fn="replace_item") + k = _leaf_key(store, leaf, k) + body = leaf._prepare_body(k, _leaf_value(store, leaf, v)) + item, hdrs = point_replace( + leaf.container, k, body, leaf.partition_key_value, etag=etag + ) + leaf._observe("replace", hdrs) + return leaf._maybe_strip(item) + + +#: Operations ``batch_items`` accepts, and whether each carries a value. +BATCH_OPS = { + "create": True, + "upsert": True, + "replace": True, + "patch": True, # the "value" is the list of patch operations, passed through as-is + "read": False, + "delete": False, +} + + +def batch_items(store, operations: Iterable[tuple]) -> list[dict]: + """Transactional batch within the store's partition, in the caller's key space. + + Each operation is ``(op, key)`` or ``(op, key, value)``, with ``op`` one of + ``BATCH_OPS``:: + + [("upsert", "a", {"v": 1}), ("replace", "b", {"v": 2}), ("delete", "c")] + + Keys (and values, except ``patch`` operation lists) are mapped down to the leaf exactly + as ``replace_item`` maps them, and bodies get the leaf's usual ``id`` / + partition-key injection. The SDK's raw ``(op, args, kwargs)`` triples are not accepted + here, because the ids inside them are opaque to any key mapping; use the metal-layer + ``cosmodol.batch(container, triples, partition_key)`` for those. + """ + leaf = _leaf_of(store, CosmosItems, fn="batch_items") + triples = [] + for operation in operations: + op, k, *rest = operation + if op not in BATCH_OPS: + raise ValueError( + f"Unknown batch op {op!r}; expected one of {list(BATCH_OPS)}." + ) + if len(rest) != int(BATCH_OPS[op]): + shape = "(op, key, value)" if BATCH_OPS[op] else "(op, key)" + raise ValueError(f"Batch op {op!r} takes {shape}; got {operation!r}.") + k = _leaf_key(store, leaf, k) + if op in ("create", "upsert", "replace"): + body = leaf._prepare_body(k, _leaf_value(store, leaf, rest[0])) + args = (body,) if op != "replace" else (k, body) + elif op == "patch": + if leaf.strict_keys: + validate_cosmos_id(k) + args = (k, rest[0]) + else: + if leaf.strict_keys: + validate_cosmos_id(k) + args = (k,) + triples.append((op, args, {})) + return _batch(leaf.container, triples, leaf.partition_key_value) + + +def partition(store, pk_value, *, wrap: Optional[Callable] = None) -> Any: + """Narrow a ``CosmosPartitionedItems`` to one partition; zero round-trips. + + On an unwrapped store this returns a ``CosmosItems`` sharing the parent's settings. + + A wrapped parent's transforms cannot be carried onto the child automatically (the + parent's keys are ``(pk, id)`` tuples, the child's are ids), so a wrapped store is + refused unless you say how to wrap the child:: + + sub = partition(redacted_store, "tenant-1", wrap=my_value_codec) + + Args: + store: A ``CosmosPartitionedItems``, or a ``dol`` wrapper around one. + pk_value: The partition-key value, as stored (not passed through any key codec). + wrap: Applied to the child store before it is returned. Required when ``store`` + is wrapped; ``wrap=lambda s: s`` explicitly asks for the bare child. + """ + leaf = _leaf_of(store, CosmosPartitionedItems, fn="partition") + if store is not leaf and wrap is None: + raise TypeError( + f"partition() refused: {type(store).__name__} wraps a CosmosPartitionedItems, " + "and the child partition would silently lose those wrappers (key and value " + "codecs alike). Pass wrap= — or " + "wrap=lambda s: s to explicitly take the bare child." + ) + child = CosmosItems( + leaf.container, + partition_key_value=pk_value, + partition_key_path=leaf.partition_key_path, + inject_id=leaf.inject_id, + inject_partition_key=leaf.inject_partition_key, + strict_keys=leaf.strict_keys, + strip_system_fields=leaf.strip_system_fields, + record_ru=leaf.record_ru, + ) + return child if wrap is None else wrap(child) + + +# --------------------------------------------------------------------------- +# Tree-level operations (CosmosDatabase, CosmosAccount) +# --------------------------------------------------------------------------- + + +def add_container( + db_store, + name: str, + *, + partition_key_path: str = "/id", + throughput: Optional[int] = None, + indexing_policy: Optional[dict] = None, + default_ttl: Optional[int] = None, + unique_key_policy: Optional[dict] = None, + conflict_resolution_policy: Optional[dict] = None, + **extra, +): + """Create a container named ``name`` (in the caller's key space) in ``db_store``. + + Args: + db_store: A ``CosmosDatabase``, or a ``dol`` wrapper around one. + name: Container name, as ``db_store[name]`` would take it. + partition_key_path: Partition-key path (e.g. ``"/id"``, ``"/tenantId"``). + Defaults to ``"/id"`` per ``misc/docs/design_decisions.md`` §3. + throughput: Provisioned RU/s (e.g. 400). ``None`` uses the database's + shared throughput if any, otherwise the account default. + indexing_policy, default_ttl, unique_key_policy, conflict_resolution_policy: + Standard Cosmos container kwargs. + **extra: Forwarded to ``DatabaseProxy.create_container``. + + Returns: + The SDK's ``ContainerProxy``. + """ + leaf = _leaf_of(db_store, CosmosDatabase, fn="add_container") + return leaf.database.create_container( + id=_leaf_key(db_store, leaf, name), + partition_key=PartitionKey(path=partition_key_path), + offer_throughput=throughput, + indexing_policy=indexing_policy, + default_ttl=default_ttl, + unique_key_policy=unique_key_policy, + conflict_resolution_policy=conflict_resolution_policy, + **extra, + ) + + +def _partition_key_expr(container) -> Optional[str]: + """Cosmos SQL expression for the container's partition key, e.g. ``c["tenant"]``.""" + paths = (container.read().get("partitionKey") or {}).get("paths") or [] + if not paths: + return None + return "c" + "".join(f'["{p}"]' for p in paths[0].strip("/").split("/")) + + +def delete_container(db_store, name: str, *, force: bool = False) -> None: + """Drop container ``name`` (in the caller's key space); ``force`` empties it first. + + Without ``force`` this is ``del db_store[name]``, which refuses a non-empty container. + With ``force=True`` every item is deleted, one point delete each, and then the + container is dropped. Not recoverable without a Cosmos backup. + """ + leaf = _leaf_of(db_store, CosmosDatabase, fn="delete_container") + if not force: + del db_store[name] + return + k = _leaf_key(db_store, leaf, name) + try: + c = leaf.database.get_container_client(k) + pk_expr = _partition_key_expr(c) + except Exception as e: + raise ContainerNotFoundError(k) from e + select = ( + "SELECT c.id FROM c" + if pk_expr is None + else (f"SELECT c.id, {pk_expr} AS pk FROM c") + ) + for item in c.query_items(query=select, enable_cross_partition_query=True): + c.delete_item(item=item["id"], partition_key=item.get("pk")) + leaf.database.delete_container(k) + + +def add_database( + account_store, name: str, *, throughput: Optional[int] = None, **extra +): + """Create database ``name`` (in the caller's key space); return it as the account's + values are returned (``account_store[name]``).""" + leaf = _leaf_of(account_store, CosmosAccount, fn="add_database") + leaf.client.create_database( + id=_leaf_key(account_store, leaf, name), offer_throughput=throughput, **extra + ) + return account_store[name] + + +def delete_database(account_store, name: str, *, force: bool = False) -> None: + """Drop database ``name`` (in the caller's key space); ``force`` drops its containers. + + Without ``force`` this is ``del account_store[name]``, which refuses a non-empty + database. Not recoverable without a Cosmos backup. + """ + leaf = _leaf_of(account_store, CosmosAccount, fn="delete_database") + if not force: + del account_store[name] + return + k = _leaf_key(account_store, leaf, name) + try: + db = leaf.client.get_database_client(k) + db.read() + except Exception as e: + raise DatabaseNotFoundError(k) from e + for c in db.list_containers(): + db.delete_container(c["id"]) + leaf.client.delete_database(k) + + +__all__ = [ + "replace_item", + "batch_items", + "BATCH_OPS", + "partition", + "add_container", + "delete_container", + "add_database", + "delete_database", +] diff --git a/cosmodol/stores.py b/cosmodol/stores.py index 177a0e3..b10c975 100644 --- a/cosmodol/stores.py +++ b/cosmodol/stores.py @@ -30,9 +30,15 @@ from cosmodol.connection import CosmosConnection from cosmodol.errors import ( KeyMismatchError, + MovedToFunction, validate_cosmos_id, ) +_MOVED_REASON = ( + "Key-taking methods ignored any dol key codec wrapped around the store " + "(i2mint/cosmodol#1); the function maps the key through every layer." +) + # Cosmos system fields stripped from returned items when ``strip_system_fields=True``. SYSTEM_FIELDS = ("_etag", "_ts", "_rid", "_self", "_attachments", "_lsn") @@ -219,28 +225,18 @@ def __repr__(self) -> str: # ---- explicit methods (not Mapping-shaped) ---- - def replace( - self, - k: str, - v: dict, - *, - etag: Optional[str] = None, - ) -> dict: - """Full replace with optional ETag-conditional write.""" - body = self._prepare_body(k, v) - item, hdrs = point_replace( - self.container, k, body, self.partition_key_value, etag=etag - ) - self._observe("replace", hdrs) - return self._maybe_strip(item) - def query( self, sql: str, *, parameters: Optional[list[dict]] = None, ) -> Iterator[dict]: - """Run a SQL query scoped to this partition. Yields raw item dicts (no stripping).""" + """Run a SQL query scoped to this partition. Yields raw item dicts (no stripping). + + SQL addresses the stored documents, so this speaks the leaf's language on both + ends: ids in the results are Cosmos ids, and documents are as stored. Neither is + passed through any ``dol`` wrapper around this store. + """ return _query( self.container, sql, @@ -248,9 +244,14 @@ def query( partition_key=self.partition_key_value, ) - def batch(self, operations: list[tuple]) -> list[dict]: - """Transactional batch within this partition. See ``base.batch``.""" - return _batch(self.container, operations, self.partition_key_value) + # Key-taking methods moved to free functions (i2mint/cosmodol#1): as methods, a dol + # wrapper handed them its outer key unmapped. + replace = MovedToFunction( + "cosmodol.replace_item(store, k, v, etag=...)", reason=_MOVED_REASON + ) + batch = MovedToFunction( + "cosmodol.batch_items(store, [(op, k[, v]), ...])", reason=_MOVED_REASON + ) def _infer_pk_path(container: ContainerProxy) -> str: @@ -406,8 +407,8 @@ def __iter__(self): if not self._warned and not self.silent_full_scan: warnings.warn( "Iterating a CosmosPartitionedItems is a cross-partition scan; RU cost " - "scales with container size. Use .partition(pk_value) or .with_filter(...) " - "to narrow, or pass `silent_full_scan=True` to suppress this warning.", + "scales with container size. Use cosmodol.partition(store, pk_value) to narrow, " + "or pass `silent_full_scan=True` to suppress this warning.", UserWarning, stacklevel=2, ) @@ -439,20 +440,9 @@ def __repr__(self) -> str: f"partition_key_path={self.partition_key_path!r})" ) - # ---- narrowing helpers (sub-stores) ---- - - def partition(self, pk_value) -> CosmosItems: - """Narrow to a single partition; zero round-trips. Returns ``CosmosItems``.""" - return CosmosItems( - self.container, - partition_key_value=pk_value, - partition_key_path=self.partition_key_path, - inject_id=self.inject_id, - inject_partition_key=self.inject_partition_key, - strict_keys=self.strict_keys, - strip_system_fields=self.strip_system_fields, - record_ru=self.record_ru, - ) + partition = MovedToFunction( + "cosmodol.partition(store, pk_value)", reason=_MOVED_REASON + ) def query( self, @@ -462,7 +452,8 @@ def query( partition_key: Any = None, cross_partition: bool = False, ) -> Iterator[dict]: - """Pass-through to ``base.query``.""" + """Pass-through to ``base.query``: raw Cosmos ids and documents, as for + ``CosmosItems.query``; ``partition_key`` is a stored partition-key value.""" return _query( self.container, sql, diff --git a/cosmodol/testing.py b/cosmodol/testing.py index fa16144..52010c3 100644 --- a/cosmodol/testing.py +++ b/cosmodol/testing.py @@ -261,6 +261,40 @@ def delete_item( del self._data[key] self._set_hdrs(charge=5.0) + def execute_item_batch( + self, batch_operations: list[tuple], partition_key: Any + ) -> list[dict]: + """All-or-nothing batch: on any failure the container is restored and the error + re-raised. Supports ``create``/``upsert``/``replace``/``read``/``delete``.""" + from azure.cosmos.exceptions import CosmosResourceExistsError + + snapshot = dict(self._data) + results = [] + try: + for op, args, kwargs in batch_operations: + if op == "create": + (body,) = args + if (partition_key, body["id"]) in self._data: + raise CosmosResourceExistsError( + message=f"Exists: {body['id']}", response=None + ) + results.append(self.upsert_item(body)) + elif op == "upsert": + results.append(self.upsert_item(*args, **kwargs)) + elif op == "replace": + results.append(self.replace_item(*args, **kwargs)) + elif op == "read": + results.append(self.read_item(args[0], partition_key)) + elif op == "delete": + self.delete_item(args[0], partition_key, **kwargs) + results.append({}) + else: + raise NotImplementedError(f"FakeContainerProxy batch op {op!r}") + except Exception: + self._data = snapshot + raise + return results + def query_items( self, query: str, diff --git a/cosmodol/tests/test_integration_emulator.py b/cosmodol/tests/test_integration_emulator.py index 45d1c34..0c4e99a 100644 --- a/cosmodol/tests/test_integration_emulator.py +++ b/cosmodol/tests/test_integration_emulator.py @@ -23,7 +23,12 @@ CosmosPartitionedItems, ItemNotFoundError, KeyMismatchError, + add_container, + add_database, cosmos_store, + delete_container, + delete_database, + partition, ) from cosmodol.testing import ( EMULATOR_CONNECTION_STRING, @@ -67,12 +72,12 @@ def database(cosmos_connection): """A fresh database for the session; dropped at the end.""" name = f"cosmodol-test-{uuid.uuid4().hex[:8]}" account = CosmosAccount(cosmos_connection) - db = account.add_database(name) + db = add_database(account, name) try: yield db finally: try: - account.delete(name, force=True) + delete_database(account, name, force=True) except Exception: pass @@ -81,12 +86,12 @@ def database(cosmos_connection): def container_proxy(database): """A fresh container per test, dropped after.""" name = f"c-{uuid.uuid4().hex[:8]}" - c = database.add_container(name, partition_key_path="/tenant") + c = add_container(database, name, partition_key_path="/tenant") try: yield c finally: try: - database.delete(name, force=True) + delete_container(database, name, force=True) except Exception: pass @@ -95,12 +100,12 @@ def container_proxy(database): def container_id_partition(database): """A fresh container partitioned by /id, dropped after.""" name = f"c-{uuid.uuid4().hex[:8]}" - c = database.add_container(name, partition_key_path="/id") + c = add_container(database, name, partition_key_path="/id") try: yield c finally: try: - database.delete(name, force=True) + delete_container(database, name, force=True) except Exception: pass @@ -257,7 +262,7 @@ def test_partitioned_partition_narrowing(container_proxy): parts[("A", "k1")] = {"v": 1} parts[("A", "k2")] = {"v": 2} parts[("B", "k3")] = {"v": 3} - a_only = parts.partition("A") + a_only = partition(parts, "A") assert set(a_only) == {"k1", "k2"} @@ -298,15 +303,15 @@ def test_partitioned_len_via_query_opt_in(container_proxy): def test_database_lists_containers(database): n1 = f"c-{uuid.uuid4().hex[:6]}" n2 = f"c-{uuid.uuid4().hex[:6]}" - database.add_container(n1, partition_key_path="/id") - database.add_container(n2, partition_key_path="/id") + add_container(database, n1, partition_key_path="/id") + add_container(database, n2, partition_key_path="/id") try: names = set(database) assert n1 in names assert n2 in names finally: - database.delete(n1, force=True) - database.delete(n2, force=True) + delete_container(database, n1, force=True) + delete_container(database, n2, force=True) def test_database_setitem_disabled(database): diff --git a/cosmodol/tests/test_ops.py b/cosmodol/tests/test_ops.py new file mode 100644 index 0000000..0504b03 --- /dev/null +++ b/cosmodol/tests/test_ops.py @@ -0,0 +1,399 @@ +"""Key-taking operations honour dol wrappers (i2mint/cosmodol#1). + +Every repro from the issue, on both of dol's delegation routes: class-wraps +(``KeyCodecs.prefixed(...)(CosmosItems)(...)``, via ``DelegatedAttribute``) and +instance-wraps (``KeyCodecs.prefixed(...)(CosmosItems(...))``, via ``Store.__getattr__``). +""" + +import inspect + +import pytest +from azure.cosmos import DatabaseProxy +from dol import KeyCodecs, wrap_kvs + +from cosmodol import ( + CosmosAccount, + CosmosDatabase, + CosmosItems, + CosmosPartitionedItems, + ItemNotFoundError, + add_container, + add_database, + batch_items, + delete_container, + delete_database, + partition, + replace_item, +) +from cosmodol.testing import FakeContainerProxy + +PK = "tenant-1" +PREFIX = "app_" + + +def _container(): + c = FakeContainerProxy(partition_key_path="/_pk") + c.upsert_item({"id": "app_doc1", "_pk": PK, "payload": "MINE"}) + c.upsert_item({"id": "doc1", "_pk": PK, "payload": "SOMEONE ELSE'S"}) + return c + + +def _state(c): + return {i: d["payload"] for (_pk, i), d in sorted(c._data.items())} + + +def _class_wrapped(c): + return KeyCodecs.prefixed(PREFIX)(CosmosItems)( + c, partition_key_value=PK, partition_key_path="/_pk" + ) + + +def _instance_wrapped(c): + return KeyCodecs.prefixed(PREFIX)( + CosmosItems(c, partition_key_value=PK, partition_key_path="/_pk") + ) + + +ROUTES = pytest.mark.parametrize( + "wrap", [_class_wrapped, _instance_wrapped], ids=["class-wrap", "instance-wrap"] +) + + +# --------------------------------------------------------------------------- +# Items +# --------------------------------------------------------------------------- + + +@ROUTES +def test_replace_item_maps_the_key(wrap): + c = _container() + w = wrap(c) + assert w["doc1"]["id"] == "app_doc1" + replace_item(w, "doc1", {"payload": "UPDATED"}) + assert _state(c) == {"app_doc1": "UPDATED", "doc1": "SOMEONE ELSE'S"} + + +def test_replace_item_through_a_temporary_wrapper(): + """The lost-reference case that defeats ``wrapped_self``: nothing else holds the + wrapper, and the function is still right because it is handed the wrapper.""" + c = _container() + replace_item(_instance_wrapped(c), "doc1", {"payload": "UPDATED"}) + assert _state(c)["app_doc1"] == "UPDATED" + + +def test_replace_item_unwrapped_is_unchanged(): + c = _container() + leaf = CosmosItems(c, partition_key_value=PK, partition_key_path="/_pk") + replace_item(leaf, "doc1", {"payload": "NEW"}) + assert _state(c) == {"app_doc1": "MINE", "doc1": "NEW"} + + +@ROUTES +def test_replace_item_missing_mapped_key_is_loud(wrap): + c = _container() + with pytest.raises(ItemNotFoundError): + replace_item(wrap(c), "nope", {"payload": "x"}) + assert _state(c) == {"app_doc1": "MINE", "doc1": "SOMEONE ELSE'S"} + + +def test_replace_item_applies_value_codec(): + c = _container() + Encoded = wrap_kvs( + KeyCodecs.prefixed(PREFIX)(CosmosItems), + data_of_obj=lambda s: {"payload": s.upper()}, + obj_of_data=lambda d: d["payload"].lower(), + ) + w = Encoded(c, partition_key_value=PK, partition_key_path="/_pk") + replace_item(w, "doc1", "shout") + assert _state(c)["app_doc1"] == "SHOUT" + assert w["doc1"] == "shout" + + +@ROUTES +def test_batch_items_maps_every_key(wrap): + c = _container() + c.upsert_item({"id": "app_gone", "_pk": PK, "payload": "x"}) + c.upsert_item({"id": "gone", "_pk": PK, "payload": "keep"}) + batch_items( + wrap(c), + [ + ("upsert", "new", {"payload": "N"}), + ("replace", "doc1", {"payload": "R"}), + ("delete", "gone"), + ], + ) + assert _state(c) == { + "app_doc1": "R", + "app_new": "N", + "doc1": "SOMEONE ELSE'S", + "gone": "keep", + } + + +def test_batch_items_rejects_sdk_triples_and_bad_shapes(): + w = _class_wrapped(_container()) + with pytest.raises(ValueError, match="Unknown batch op"): + batch_items(w, [("frobnicate", "k")]) + with pytest.raises(ValueError, match=r"takes \(op, key\)"): + batch_items(w, [("delete", "doc1", {"payload": 1})]) + with pytest.raises(ValueError, match=r"takes \(op, key, value\)"): + batch_items(w, [("upsert", "doc1")]) + + +def test_functions_validate_their_first_argument(): + with pytest.raises(TypeError, match="needs a CosmosItems"): + replace_item({"doc1": {}}, "doc1", {}) + with pytest.raises(TypeError, match="needs a CosmosPartitionedItems"): + partition(_class_wrapped(_container()), PK) + + +# --------------------------------------------------------------------------- +# partition +# --------------------------------------------------------------------------- + + +def _redacted_partitioned(c): + Redacted = wrap_kvs( + CosmosPartitionedItems, obj_of_data=lambda d: {**d, "payload": ""} + ) + return Redacted(c, partition_key_path="/_pk", silent_full_scan=True) + + +def test_partition_refuses_to_strip_wrappers(): + p = _redacted_partitioned(_container()) + assert p[(PK, "doc1")]["payload"] == "" + with pytest.raises(TypeError, match="partition\\(\\) refused"): + partition(p, PK) + + +def test_partition_wrap_rewraps_the_child(): + p = _redacted_partitioned(_container()) + sub = partition(p, PK, wrap=wrap_kvs(obj_of_data=lambda d: {**d, "payload": "-"})) + assert sub["doc1"]["payload"] == "-" + + +def test_partition_unwrapped_returns_leaf(): + leaf = CosmosPartitionedItems(_container(), partition_key_path="/_pk") + sub = partition(leaf, PK) + assert type(sub) is CosmosItems + assert sub["doc1"]["payload"] == "SOMEONE ELSE'S" + + +# --------------------------------------------------------------------------- +# Trees (hand-rolled backend fakes; CosmosDatabase / CosmosAccount are the real ones) +# --------------------------------------------------------------------------- + + +class _FakeContainer: + def __init__(self, cid, items, *, pk_path="/tenant"): + self.id, self.items, self.pk_path = cid, list(items), pk_path + self.deleted = [] + + def read(self): + return {"id": self.id, "partitionKey": {"paths": [self.pk_path]}} + + def query_items(self, query, enable_cross_partition_query=False, **kw): + prop = self.pk_path.strip("/") + for i in self.items: + yield {"id": i["id"], "pk": i.get(prop)} if "AS pk" in query else dict(i) + + def delete_item(self, item, partition_key=None): + self.deleted.append((item, partition_key)) + self.items = [i for i in self.items if i["id"] != item] + + +class _FakeDatabase(DatabaseProxy): + def __init__(self, containers, *, dbid="db"): + self.id, self.containers, self.created = dbid, containers, [] + + def read(self): + return {"id": self.id} + + def list_containers(self): + return [{"id": n} for n in self.containers] + + def get_container_client(self, k): + if k not in self.containers: + raise LookupError(k) + return self.containers[k] + + def create_container(self, id, **kw): + self.created.append(id) + self.containers[id] = _FakeContainer(id, []) + return self.containers[id] + + def delete_container(self, k): + del self.containers[k] + + +def _db(): + return _FakeDatabase( + { + "app_events": _FakeContainer("app_events", [{"id": "e1", "tenant": "t"}]), + "events": _FakeContainer( + "events", [{"id": f"prod-{i}", "tenant": "t"} for i in range(3)] + ), + } + ) + + +DB_ROUTES = pytest.mark.parametrize( + "wrap", + [ + lambda db: KeyCodecs.prefixed(PREFIX)(CosmosDatabase)(db), + lambda db: KeyCodecs.prefixed(PREFIX)(CosmosDatabase(db)), + ], + ids=["class-wrap", "instance-wrap"], +) + + +@DB_ROUTES +def test_delete_container_force_drops_the_mapped_container(wrap): + db = _db() + events = db.containers["events"] + view = wrap(db) + assert view["events"].id == "app_events" + delete_container(view, "events", force=True) + assert sorted(db.containers) == ["events"] + assert len(events.items) == 3 + + +def test_delete_container_force_passes_the_real_partition_key(): + db = _db() + target = db.containers["events"] + delete_container(CosmosDatabase(db), "events", force=True) + assert target.deleted == [(f"prod-{i}", "t") for i in range(3)] + + +@DB_ROUTES +def test_delete_container_without_force_is_delitem(wrap): + db = _db() + with pytest.raises(Exception, match="not empty"): + delete_container(wrap(db), "events") + assert sorted(db.containers) == ["app_events", "events"] + + +@DB_ROUTES +def test_add_container_maps_the_name(wrap): + db = _db() + add_container(wrap(db), "logs") + assert db.created == ["app_logs"] + + +class _FakeClient: + def __init__(self, dbs): + self.dbs = dbs + + def list_databases(self): + return [{"id": n} for n in self.dbs] + + def get_database_client(self, k): + if k not in self.dbs: + raise LookupError(k) + return self.dbs[k] + + def create_database(self, id, offer_throughput=None, **kw): + self.dbs[id] = _FakeDatabase({}, dbid=id) + return self.dbs[id] + + def delete_database(self, k): + del self.dbs[k] + + +class _FakeConnection: + def __init__(self, client): + self.client = client + + +def _account(wrap_cls): + client = _FakeClient( + { + "app_main": _FakeDatabase({"c": _FakeContainer("c", [])}, dbid="app_main"), + "main": _FakeDatabase({"prod": _FakeContainer("prod", [])}, dbid="main"), + } + ) + from unittest.mock import patch + + with patch( + "cosmodol.trees.CosmosConnection.from_anything", + side_effect=lambda c: c, + ): + acct = ( + KeyCodecs.prefixed(PREFIX)(CosmosAccount)(_FakeConnection(client)) + if wrap_cls + else KeyCodecs.prefixed(PREFIX)(CosmosAccount(_FakeConnection(client))) + ) + return acct, client + + +@pytest.mark.parametrize("wrap_cls", [True, False], ids=["class-wrap", "instance-wrap"]) +def test_delete_database_force_drops_the_mapped_database(wrap_cls): + acct, client = _account(wrap_cls) + delete_database(acct, "main", force=True) + assert sorted(client.dbs) == ["main"] + assert "prod" in client.dbs["main"].containers + + +@pytest.mark.parametrize("wrap_cls", [True, False], ids=["class-wrap", "instance-wrap"]) +def test_add_database_maps_the_name(wrap_cls): + acct, client = _account(wrap_cls) + db = add_database(acct, "extra") + assert "app_extra" in client.dbs and "extra" not in client.dbs + assert db.name == "app_extra" + + +# --------------------------------------------------------------------------- +# The moved methods, and the conformance inventory (s3dol ADR-0011 D5) +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "obj, name, replacement", + [ + (lambda: _class_wrapped(_container()), "replace", "cosmodol.replace_item"), + (lambda: _instance_wrapped(_container()), "batch", "cosmodol.batch_items"), + ( + lambda: CosmosPartitionedItems(_container(), partition_key_path="/_pk"), + "partition", + "cosmodol.partition", + ), + (lambda: CosmosDatabase(_db()), "delete", "cosmodol.delete_container"), + ( + lambda: KeyCodecs.prefixed(PREFIX)(CosmosDatabase)(_db()), + "add_container", + "cosmodol.add_container", + ), + ], +) +def test_moved_methods_say_where_they_went(obj, name, replacement): + o = obj() + assert not hasattr(o, name) + with pytest.raises(AttributeError, match=replacement): + getattr(o, name) + + +#: Every public method of every store class, reviewed for taking a key at any position +#: (or nested in a structure). A NEW method fails this test on purpose: decide whether it +#: takes a key, and if it does, make it a free function in ``cosmodol.ops`` instead. +PUBLIC_METHOD_INVENTORY = { + # ``query``: takes SQL, not a store key; documented as speaking leaf ids/documents. + CosmosItems: {"query"}, + CosmosPartitionedItems: {"query"}, + CosmosDatabase: set(), + CosmosAccount: set(), +} + + +@pytest.mark.parametrize("cls", list(PUBLIC_METHOD_INVENTORY), ids=lambda c: c.__name__) +def test_no_unreviewed_public_methods(cls): + from dol.base import KvPersister + + inherited = set(dir(KvPersister)) + own = { + name + for name, attr in inspect.getmembers(cls) + if not name.startswith("_") + and name not in inherited + and inspect.isfunction(attr) + } + assert own == PUBLIC_METHOD_INVENTORY[cls] diff --git a/cosmodol/tests/test_stores.py b/cosmodol/tests/test_stores.py index d5ed2a7..f0c02e6 100644 --- a/cosmodol/tests/test_stores.py +++ b/cosmodol/tests/test_stores.py @@ -9,6 +9,8 @@ CosmosPartitionedItems, ItemNotFoundError, KeyMismatchError, + partition, + replace_item, ) from cosmodol.testing import FakeContainerProxy @@ -83,7 +85,7 @@ def test_ru_observability(items): def test_replace_with_etag_conditional(items): items["k1"] = {"v": 1} etag = items.last_response_headers.etag - items.replace("k1", {"v": 2}, etag=etag) + replace_item(items, "k1", {"v": 2}, etag=etag) assert items["k1"]["v"] == 2 @@ -92,7 +94,7 @@ def test_replace_with_stale_etag_raises(items): from azure.cosmos.exceptions import CosmosAccessConditionFailedError with pytest.raises(CosmosAccessConditionFailedError): - items.replace("k1", {"v": 2}, etag="stale-etag") + replace_item(items, "k1", {"v": 2}, etag="stale-etag") def test_record_ru_callback(): @@ -178,7 +180,7 @@ def test_partitioned_partition_narrowing(parts): parts[("A", "k1")] = {"v": 1} parts[("A", "k2")] = {"v": 2} parts[("B", "k3")] = {"v": 3} - a_only = parts.partition("A") + a_only = partition(parts, "A") assert set(a_only) == {"k1", "k2"} diff --git a/cosmodol/trees.py b/cosmodol/trees.py index 056c86e..d2cc9c7 100644 --- a/cosmodol/trees.py +++ b/cosmodol/trees.py @@ -5,7 +5,8 @@ - ``CosmosAccount`` — keys are database names; values are ``CosmosDatabase`` instances. Container/database creation is NOT via ``__setitem__`` — too parameter-rich; use -``add_container(...)`` / ``add_database(...)``. See +``cosmodol.add_container(db_store, ...)`` / ``cosmodol.add_database(account_store, ...)`` +(free functions, so a ``dol`` key codec around the store is honoured). See ``misc/docs/design_decisions.md`` §7. """ @@ -13,7 +14,7 @@ from typing import Any, Callable, Iterator, Optional, Union -from azure.cosmos import CosmosClient, DatabaseProxy, PartitionKey +from azure.cosmos import CosmosClient, DatabaseProxy from dol import KvReader from cosmodol.connection import CosmosConnection @@ -22,6 +23,12 @@ ContainerNotFoundError, DatabaseNotEmptyError, DatabaseNotFoundError, + MovedToFunction, +) + +_MOVED_REASON = ( + "As a method it ignored any dol key codec wrapped around the store and acted on the " + "unmapped name (i2mint/cosmodol#1); the function maps it through every layer." ) @@ -33,8 +40,9 @@ class CosmosDatabase(KvReader): """Mapping of container name → ``ContainerProxy`` (or store-wrapped form). - ``__setitem__`` is disabled — use ``add_container(name, partition_key_path=..., ...)``. - ``__delitem__`` refuses non-empty containers; use ``self.delete(name, force=True)``. + ``__setitem__`` is disabled — use ``cosmodol.add_container(db_store, name, ...)``. + ``__delitem__`` refuses non-empty containers; use + ``cosmodol.delete_container(db_store, name, force=True)``. Args: database: Either an existing ``DatabaseProxy``, a database name string (resolved @@ -92,7 +100,7 @@ def __getitem__(self, k: str): def __setitem__(self, k, v): raise TypeError( "Container creation does not go through __setitem__ (too parameter-rich). " - "Use `database_store.add_container(name, partition_key_path=..., ...)` instead." + "Use `cosmodol.add_container(db_store, name, partition_key_path=..., ...)` instead." ) def __delitem__(self, k: str) -> None: @@ -101,7 +109,7 @@ def __delitem__(self, k: str) -> None: c.read() except Exception as e: raise ContainerNotFoundError(k) from e - # Refuse non-empty container; force-delete via .delete(k, force=True). + # Refuse non-empty container; force-delete via cosmodol.delete_container. try: next( iter( @@ -113,7 +121,8 @@ def __delitem__(self, k: str) -> None: ) raise ContainerNotEmptyError( f"Container {k!r} is not empty. Call " - f"`db_store.delete({k!r}, force=True)` to cascade-delete its items." + f"`cosmodol.delete_container(db_store, {k!r}, force=True)` to " + "cascade-delete its items." ) except StopIteration: pass @@ -122,57 +131,15 @@ def __delitem__(self, k: str) -> None: def __repr__(self) -> str: return f"CosmosDatabase({self.name!r})" - # ---- explicit creators / destroyers ---- + # ---- creators / destroyers: free functions in cosmodol.ops (i2mint/cosmodol#1) ---- - def add_container( - self, - name: str, - *, - partition_key_path: str = "/id", - throughput: Optional[int] = None, - indexing_policy: Optional[dict] = None, - default_ttl: Optional[int] = None, - unique_key_policy: Optional[dict] = None, - conflict_resolution_policy: Optional[dict] = None, - **extra, - ): - """Create a container. - - Args: - name: Container id. - partition_key_path: Partition-key path (e.g. ``"/id"``, ``"/tenantId"``). - Defaults to ``"/id"`` per ``misc/docs/design_decisions.md`` §3. - throughput: Provisioned RU/s (e.g. 400). ``None`` uses the database's - shared throughput if any, otherwise the account default. - indexing_policy, default_ttl, unique_key_policy, conflict_resolution_policy: - Standard Cosmos container kwargs. - **extra: Forwarded to ``DatabaseProxy.create_container``. - """ - return self.database.create_container( - id=name, - partition_key=PartitionKey(path=partition_key_path), - offer_throughput=throughput, - indexing_policy=indexing_policy, - default_ttl=default_ttl, - unique_key_policy=unique_key_policy, - conflict_resolution_policy=conflict_resolution_policy, - **extra, - ) - - def delete(self, k: str, *, force: bool = False) -> None: - """Delete a container; if ``force=True``, cascade-delete its items first.""" - try: - c = self.database.get_container_client(k) - c.read() - except Exception as e: - raise ContainerNotFoundError(k) from e - if force: - for item in c.query_items( - query="SELECT c.id, c._partitionKey FROM c", - enable_cross_partition_query=True, - ): - c.delete_item(item=item["id"], partition_key=item.get("_partitionKey")) - self.database.delete_container(k) + add_container = MovedToFunction( + "cosmodol.add_container(db_store, name, partition_key_path=..., ...)", + reason=_MOVED_REASON, + ) + delete = MovedToFunction( + "cosmodol.delete_container(db_store, name, force=True)", reason=_MOVED_REASON + ) # --------------------------------------------------------------------------- @@ -183,8 +150,9 @@ def delete(self, k: str, *, force: bool = False) -> None: class CosmosAccount(KvReader): """Mapping of database name → ``CosmosDatabase``. - ``__setitem__`` is disabled — use ``add_database(name, throughput=...)``. - ``__delitem__`` refuses non-empty databases; use ``self.delete(name, force=True)``. + ``__setitem__`` is disabled — use ``cosmodol.add_database(account_store, name, ...)``. + ``__delitem__`` refuses non-empty databases; use + ``cosmodol.delete_database(account_store, name, force=True)``. """ def __init__( @@ -228,7 +196,7 @@ def __getitem__(self, k: str) -> CosmosDatabase: def __setitem__(self, k, v): raise TypeError( "Database creation does not go through __setitem__ (too parameter-rich). " - "Use `account_store.add_database(name, throughput=...)` instead." + "Use `cosmodol.add_database(account_store, name, throughput=...)` instead." ) def __delitem__(self, k: str) -> None: @@ -241,7 +209,8 @@ def __delitem__(self, k: str) -> None: next(iter(db.list_containers())) raise DatabaseNotEmptyError( f"Database {k!r} is not empty. Call " - f"`account_store.delete({k!r}, force=True)` to cascade-delete its containers." + f"`cosmodol.delete_database(account_store, {k!r}, force=True)` to " + "cascade-delete its containers." ) except StopIteration: pass @@ -254,24 +223,13 @@ def __repr__(self) -> str: url = "" return f"CosmosAccount(endpoint={url!r})" - def add_database( - self, - name: str, - *, - throughput: Optional[int] = None, - **extra, - ) -> CosmosDatabase: - db = self._client.create_database(id=name, offer_throughput=throughput, **extra) - return self.database_factory(db) + # ---- creators / destroyers: free functions in cosmodol.ops (i2mint/cosmodol#1) ---- - def delete(self, k: str, *, force: bool = False) -> None: - """Delete a database; if ``force=True``, cascade-delete its containers first.""" - try: - db = self._client.get_database_client(k) - db.read() - except Exception as e: - raise DatabaseNotFoundError(k) from e - if force: - for c in db.list_containers(): - db.delete_container(c["id"]) - self._client.delete_database(k) + add_database = MovedToFunction( + "cosmodol.add_database(account_store, name, throughput=...)", + reason=_MOVED_REASON, + ) + delete = MovedToFunction( + "cosmodol.delete_database(account_store, name, force=True)", + reason=_MOVED_REASON, + ) diff --git a/misc/docs/architecture.md b/misc/docs/architecture.md index 7b28002..db066aa 100644 --- a/misc/docs/architecture.md +++ b/misc/docs/architecture.md @@ -144,7 +144,7 @@ items = CosmosPartitionedItems( items[("tenant-42", "k1")] = {"v": 1} items[("tenant-42", "k1")] # point read list(items) # cross-partition SELECT c.id, c. FROM c — warned by default -items.partition("tenant-42") # → CosmosItems scoped to that partition (no round-trip) +cosmodol.partition(items, "tenant-42") # → CosmosItems scoped to that partition (no round-trip) ``` When `partition_key_path == "/id"` (the recommended default for new containers), @@ -153,10 +153,10 @@ to `CosmosItems(partition_key_value=k)` for every `k` — see [design_decisions. #### `cosmodol.trees` -**`CosmosDatabase`** — `Mapping[str, ContainerProxy]` (or a wrapped store; see below) — keys are container names, values are containers. `__setitem__` is **disabled** by default (use `add_container(name, partition_key_path=..., throughput=...)`); container creation is too parameter-rich to fit one-arg dict assignment. +**`CosmosDatabase`** — `Mapping[str, ContainerProxy]` (or a wrapped store; see below) — keys are container names, values are containers. `__setitem__` is **disabled** by default (use `cosmodol.add_container(db_store, name, partition_key_path=..., throughput=...)`); container creation is too parameter-rich to fit one-arg dict assignment. **`CosmosAccount`** — `Mapping[str, CosmosDatabase]` — keys are database names. Same opt-out -on `__setitem__`; explicit `add_database(name, throughput=...)`. +on `__setitem__`; explicit `cosmodol.add_database(account_store, name, throughput=...)`. Both tree-level stores accept a `store_factory` callable that decides how to wrap each child container — e.g. `store_factory=lambda c: CosmosItems(c, partition_key_value="default")` — @@ -228,8 +228,12 @@ items = CosmosItems( ## Sub-stores -- `CosmosPartitionedItems.partition(pk_value) → CosmosItems` — zero round-trip narrowing. -- `CosmosItems.with_filter(sql_predicate) → CosmosItems` — narrows iteration with an extra `WHERE` clause; `__getitem__` is unaffected (point reads always hit the underlying item). +- `cosmodol.partition(store, pk_value) → CosmosItems` — zero round-trip narrowing. On a wrapped store it refuses unless given `wrap=` (the child cannot inherit the parent's wrappers automatically). +- `CosmosItems.with_filter(sql_predicate)` is **not implemented** (an earlier draft of this doc listed it). + +## Key-taking operations are free functions + +`dol` wraps by delegation, so a *method* of a wrapped store runs bound to the leaf and receives the caller's unmapped key ([#1](https://github.com/i2mint/cosmodol/issues/1), [dol#83](https://github.com/i2mint/dol/issues/83)). So the stores carry no public method that takes a key: `replace_item`, `batch_items`, `partition`, `add_container`, `delete_container`, `add_database` and `delete_database` live in `cosmodol.ops` (re-exported from `cosmodol`) and take the store first, resolving the key through every wrapping layer. The old method names raise an `AttributeError` naming their replacement. `query` stays a method: it takes SQL, and speaks raw Cosmos ids and documents on both ends. A conformance test (`tests/test_ops.py`) inventories every public method and fails on a new one. Same design as [s3dol ADR-0011](https://github.com/i2mint/s3dol/blob/master/misc/docs/decisions/0011-keyed-capability-surface.md). We do **not** overload `s[prefix]` for narrowing — Cosmos keys are flat strings and the prefix convention has no native server-side support. diff --git a/misc/docs/design_decisions.md b/misc/docs/design_decisions.md index 839fa83..5f2766b 100644 --- a/misc/docs/design_decisions.md +++ b/misc/docs/design_decisions.md @@ -85,7 +85,7 @@ call: ``` UserWarning: Iterating a CosmosPartitionedItems is a cross-partition scan; RU cost -scales with container size. Use .partition(pk_value) or .with_filter(...) to narrow, +scales with container size. Use cosmodol.partition(store, pk_value) to narrow, or pass `silent_full_scan=True` to suppress this warning. ``` @@ -100,7 +100,7 @@ throughput offer mode, default TTL, unique key policy, conflict resolution polic So: - `CosmosAccount.__setitem__` / `CosmosDatabase.__setitem__` are **disabled** (raise `TypeError` with a directive message). -- `add_database(name, throughput=None)` and `add_container(name, partition_key_path, throughput=None, indexing_policy=None, default_ttl=None, ...)` are the explicit creators. +- `cosmodol.add_database(account_store, name, throughput=None)` and `cosmodol.add_container(db_store, name, partition_key_path, throughput=None, indexing_policy=None, default_ttl=None, ...)` are the explicit creators. They are free functions, not methods, so a `dol` key codec around the store is honoured (see architecture.md, "Key-taking operations are free functions"). This mirrors how `chromadol` disables `clear()` for safety — the surface owes the user clarity over conciseness here. @@ -108,7 +108,7 @@ clarity over conciseness here. ## 8. `__delitem__` on tree-level stores never auto-empties `del account_store['mydb']` deletes only an *empty* database. If it has containers, -raises with a directive message. `account_store.delete('mydb', force=True)` is the explicit +raises with a directive message. `cosmodol.delete_database(account_store, 'mydb', force=True)` is the explicit "yes, drop everything below" escape hatch. Same on `CosmosDatabase.__delitem__` for non-empty containers. From 33d5785edd2013515eded9d74a5cf6e110507ed1 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Tue, 13 Oct 2026 04:41:38 +0000 Subject: [PATCH 2/2] fix: honour scope filters, hierarchical pks; pin dol (adversarial review of #2) - Operations on existing resources check k in store through the outer store first, so a filt_iter scope stops replace_item, batch read/replace/patch/ delete, and the forced delete_container/delete_database. - add_database returns the leaf's value instead of re-reading through the wrapper (which could raise after the database was created). - Forced container delete projects hierarchical (MultiHash) partition keys as a list and passes NonePartitionKeyValue for items without one. - batch_items accepts per-op SDK kwargs and names the SDK-triple mistake. - Conformance test also sees classmethods and overridden Mapping methods. - dol>=0.3.59 (inner_most_key exported). Co-Authored-By: Claude Opus 5.5 --- cosmodol/ops.py | 131 ++++++++++++++++++++++++++----------- cosmodol/tests/test_ops.py | 98 ++++++++++++++++++++++++++- pyproject.toml | 2 +- 3 files changed, 187 insertions(+), 44 deletions(-) diff --git a/cosmodol/ops.py b/cosmodol/ops.py index e656931..44bcbe3 100644 --- a/cosmodol/ops.py +++ b/cosmodol/ops.py @@ -23,23 +23,36 @@ destructive ones become free functions). The removed methods now raise an ``AttributeError`` that names their replacement here. -Caveats, stated rather than hidden (ADR-0011 D3a): - -- Keys are resolved with ``dol.inner_most_key``, i.e. each wrapping layer's - ``_id_of_key``. Every ``dol``-shipped wrapper (``wrap_kvs``, ``KeyCodecs``, ``Pipe``, - ``filt_iter``, ``cached_keys``) does this right; a hand-rolled ``__getattr__`` - passthrough layer can make it apply a transform twice. -- Values (``replace_item``, ``batch_items``) go through each layer's ``_data_of_obj``, - which is what ``wrap_kvs(data_of_obj=...)`` and ``ValueCodecs`` install. A key-aware - ``preset=`` write hook is **not** followed — such a hook only runs inside - ``__setitem__``. +What a function can and cannot honour, stated rather than hidden (ADR-0011 D3a): + +- **Key mapping** — honoured. Keys are resolved with ``dol.inner_most_key``, i.e. each + wrapping layer's ``_id_of_key`` (``wrap_kvs``, ``KeyCodecs``, ``Pipe``, class- and + instance-wraps alike). A hand-rolled ``__getattr__`` passthrough layer can make it + apply a transform twice. +- **Scope** — honoured for operations on *existing* resources. On a wrapped store, + ``replace_item``, ``batch_items``'s read/replace/patch/delete, and the forced + ``delete_container`` / ``delete_database`` first check ``k in store`` through the + outer store, so a ``filt_iter`` scope (or any wrapper whose ``__contains__`` says no) + stops them. That is one extra point read per key. Operations that *create* (batch + ``create``/``upsert``, ``add_container``, ``add_database``) cannot ask a filter, since + the key does not exist yet: they honour the key mapping only. +- **Values** — ``replace_item`` and ``batch_items`` pass values through each layer's + ``_data_of_obj`` (what ``wrap_kvs(data_of_obj=...)`` and ``ValueCodecs`` install). A + key-aware ``preset=`` write hook is **not** followed (it only runs inside + ``__setitem__``), and returned documents are in stored (leaf) form. +- **Guards and caches** — not seen. A layer that refuses writes in its own + ``__setitem__``/``__delitem__``, or caches values, is bypassed by any function, + exactly as by the old methods. Pass such a store only if you mean to bypass it. """ from __future__ import annotations +import json + from typing import Any, Callable, Iterable, Optional from azure.cosmos import PartitionKey +from azure.cosmos.partition_key import NonePartitionKeyValue from dol import inner_most_key from dol.dig import inner_most @@ -47,6 +60,7 @@ from cosmodol.errors import ( ContainerNotFoundError, DatabaseNotFoundError, + ItemNotFoundError, validate_cosmos_id, ) from cosmodol.stores import CosmosItems, CosmosPartitionedItems @@ -86,6 +100,15 @@ def _leaf_key(store, leaf, k): return inner_most_key(store, k) +def _existing_leaf_key(store, leaf, k, not_found: type[KeyError]): + """``_leaf_key``, after checking ``k in store`` through every wrapping layer, so a + scope filter around the store is honoured. Unwrapped stores skip the round-trip + (the operation itself fails on a missing resource).""" + if store is not leaf and k not in store: + raise not_found(k) + return _leaf_key(store, leaf, k) + + def _leaf_value(store, leaf, v): if store is leaf: return v @@ -115,7 +138,7 @@ def replace_item(store, k, v, *, etag: Optional[str] = None) -> dict: leaf strips them). """ leaf = _leaf_of(store, CosmosItems, fn="replace_item") - k = _leaf_key(store, leaf, k) + k = _existing_leaf_key(store, leaf, k, ItemNotFoundError) body = leaf._prepare_body(k, _leaf_value(store, leaf, v)) item, hdrs = point_replace( leaf.container, k, body, leaf.partition_key_value, etag=etag @@ -138,41 +161,58 @@ def replace_item(store, k, v, *, etag: Optional[str] = None) -> dict: def batch_items(store, operations: Iterable[tuple]) -> list[dict]: """Transactional batch within the store's partition, in the caller's key space. - Each operation is ``(op, key)`` or ``(op, key, value)``, with ``op`` one of - ``BATCH_OPS``:: + Each operation is ``(op, key)`` or ``(op, key, value)`` (``op`` one of + ``BATCH_OPS``), optionally followed by a dict of per-operation SDK keyword arguments + such as ``{"if_match_etag": etag}``:: [("upsert", "a", {"v": 1}), ("replace", "b", {"v": 2}), ("delete", "c")] Keys (and values, except ``patch`` operation lists) are mapped down to the leaf exactly as ``replace_item`` maps them, and bodies get the leaf's usual ``id`` / - partition-key injection. The SDK's raw ``(op, args, kwargs)`` triples are not accepted - here, because the ids inside them are opaque to any key mapping; use the metal-layer - ``cosmodol.batch(container, triples, partition_key)`` for those. + partition-key injection. On a wrapped store, every key a ``read``, ``replace``, + ``patch`` or ``delete`` targets must be in the store (one point read each), unless an + earlier ``create``/``upsert`` of the same batch makes it. The SDK's raw + ``(op, args, kwargs)`` triples are not accepted here, because the ids inside them are + opaque to any key mapping; use ``cosmodol.batch(container, triples, partition_key)``. """ leaf = _leaf_of(store, CosmosItems, fn="batch_items") - triples = [] + triples, created = [], set() for operation in operations: op, k, *rest = operation if op not in BATCH_OPS: raise ValueError( f"Unknown batch op {op!r}; expected one of {list(BATCH_OPS)}." ) - if len(rest) != int(BATCH_OPS[op]): - shape = "(op, key, value)" if BATCH_OPS[op] else "(op, key)" - raise ValueError(f"Batch op {op!r} takes {shape}; got {operation!r}.") - k = _leaf_key(store, leaf, k) + if isinstance(k, tuple): + raise ValueError( + f"Batch op {operation!r} looks like an SDK (op, args, kwargs) triple. " + "batch_items takes (op, key[, value][, kwargs]); for raw triples use " + "cosmodol.batch(container, triples, partition_key)." + ) + has_value = BATCH_OPS[op] + kwargs = {} + if len(rest) == has_value + 1 and isinstance(rest[-1], dict): + *rest, kwargs = rest + if len(rest) != has_value: + shape = "(op, key, value)" if has_value else "(op, key)" + raise ValueError( + f"Batch op {op!r} takes {shape}[, kwargs]; got {operation!r}." + ) + if op in ("create", "upsert"): + created.add(k) + k = _leaf_key(store, leaf, k) + elif k in created: + k = _leaf_key(store, leaf, k) + else: + k = _existing_leaf_key(store, leaf, k, ItemNotFoundError) if op in ("create", "upsert", "replace"): body = leaf._prepare_body(k, _leaf_value(store, leaf, rest[0])) args = (body,) if op != "replace" else (k, body) - elif op == "patch": - if leaf.strict_keys: - validate_cosmos_id(k) - args = (k, rest[0]) else: if leaf.strict_keys: validate_cosmos_id(k) - args = (k,) - triples.append((op, args, {})) + args = (k, rest[0]) if op == "patch" else (k,) + triples.append((op, args, kwargs)) return _batch(leaf.container, triples, leaf.partition_key_value) @@ -196,8 +236,8 @@ def partition(store, pk_value, *, wrap: Optional[Callable] = None) -> Any: leaf = _leaf_of(store, CosmosPartitionedItems, fn="partition") if store is not leaf and wrap is None: raise TypeError( - f"partition() refused: {type(store).__name__} wraps a CosmosPartitionedItems, " - "and the child partition would silently lose those wrappers (key and value " + "partition() refused: this CosmosPartitionedItems is wrapped by a dol " + "wrapper, and the child partition would silently lose those wrappers (key and value " "codecs alike). Pass wrap= — or " "wrap=lambda s: s to explicitly take the bare child." ) @@ -261,11 +301,16 @@ def add_container( def _partition_key_expr(container) -> Optional[str]: - """Cosmos SQL expression for the container's partition key, e.g. ``c["tenant"]``.""" + """Cosmos SQL expression for the container's partition key: ``c["tenant"]``, or an + array of them for a hierarchical (MultiHash) key, which the SDK takes as a list.""" paths = (container.read().get("partitionKey") or {}).get("paths") or [] - if not paths: - return None - return "c" + "".join(f'["{p}"]' for p in paths[0].strip("/").split("/")) + + def expr(path): + return "c" + "".join(f"[{json.dumps(p)}]" for p in path.strip("/").split("/")) + + if len(paths) <= 1: + return expr(paths[0]) if paths else None + return "[" + ", ".join(map(expr, paths)) + "]" def delete_container(db_store, name: str, *, force: bool = False) -> None: @@ -279,7 +324,7 @@ def delete_container(db_store, name: str, *, force: bool = False) -> None: if not force: del db_store[name] return - k = _leaf_key(db_store, leaf, name) + k = _existing_leaf_key(db_store, leaf, name, ContainerNotFoundError) try: c = leaf.database.get_container_client(k) pk_expr = _partition_key_expr(c) @@ -291,20 +336,26 @@ def delete_container(db_store, name: str, *, force: bool = False) -> None: else (f"SELECT c.id, {pk_expr} AS pk FROM c") ) for item in c.query_items(query=select, enable_cross_partition_query=True): - c.delete_item(item=item["id"], partition_key=item.get("pk")) + # A projection omits an undefined value: the item has no partition-key property. + pk = item["pk"] if "pk" in item else NonePartitionKeyValue + c.delete_item(item=item["id"], partition_key=pk) leaf.database.delete_container(k) def add_database( account_store, name: str, *, throughput: Optional[int] = None, **extra ): - """Create database ``name`` (in the caller's key space); return it as the account's - values are returned (``account_store[name]``).""" + """Create database ``name`` (in the caller's key space). + + Returns the leaf's value for it (its ``database_factory`` output, a + ``CosmosDatabase`` by default), not ``account_store[name]``: a wrapper that does not + expose the new name would otherwise raise after the database was created. + """ leaf = _leaf_of(account_store, CosmosAccount, fn="add_database") - leaf.client.create_database( + db = leaf.client.create_database( id=_leaf_key(account_store, leaf, name), offer_throughput=throughput, **extra ) - return account_store[name] + return leaf.database_factory(db) def delete_database(account_store, name: str, *, force: bool = False) -> None: @@ -317,7 +368,7 @@ def delete_database(account_store, name: str, *, force: bool = False) -> None: if not force: del account_store[name] return - k = _leaf_key(account_store, leaf, name) + k = _existing_leaf_key(account_store, leaf, name, DatabaseNotFoundError) try: db = leaf.client.get_database_client(k) db.read() diff --git a/cosmodol/tests/test_ops.py b/cosmodol/tests/test_ops.py index 0504b03..14b3d96 100644 --- a/cosmodol/tests/test_ops.py +++ b/cosmodol/tests/test_ops.py @@ -135,7 +135,7 @@ def test_batch_items_rejects_sdk_triples_and_bad_shapes(): with pytest.raises(ValueError, match="Unknown batch op"): batch_items(w, [("frobnicate", "k")]) with pytest.raises(ValueError, match=r"takes \(op, key\)"): - batch_items(w, [("delete", "doc1", {"payload": 1})]) + batch_items(w, [("delete", "doc1", "extra", {})]) with pytest.raises(ValueError, match=r"takes \(op, key, value\)"): batch_items(w, [("upsert", "doc1")]) @@ -393,7 +393,99 @@ def test_no_unreviewed_public_methods(cls): name for name, attr in inspect.getmembers(cls) if not name.startswith("_") - and name not in inherited - and inspect.isfunction(attr) + and (name not in inherited or name in vars(cls)) # incl. overridden Mapping API + and callable(attr) # functions, classmethods, staticmethods + and not isinstance(attr, type) } assert own == PUBLIC_METHOD_INVENTORY[cls] + + +def test_inventory_sees_classmethods_and_overrides(): + class Grown(CosmosItems): + @classmethod + def wipe(cls, k): ... + + def get(self, k, default=None): ... + + with pytest.raises(AssertionError): + PUBLIC_METHOD_INVENTORY[Grown] = PUBLIC_METHOD_INVENTORY[CosmosItems] + try: + test_no_unreviewed_public_methods(Grown) + finally: + del PUBLIC_METHOD_INVENTORY[Grown] + + +# --------------------------------------------------------------------------- +# Adversarial-review findings +# --------------------------------------------------------------------------- + + +def _scoped_items(c): + from dol import filt_iter + + return filt_iter( + CosmosItems(c, partition_key_value=PK, partition_key_path="/_pk"), + filt=lambda k: k.startswith(PREFIX), + ) + + +def test_scope_filter_stops_replace_and_batch_on_out_of_scope_keys(): + c = _container() + view = _scoped_items(c) + with pytest.raises(ItemNotFoundError): + replace_item(view, "doc1", {"payload": "OVERWRITTEN"}) + with pytest.raises(ItemNotFoundError): + batch_items(view, [("delete", "doc1")]) + assert _state(c) == {"app_doc1": "MINE", "doc1": "SOMEONE ELSE'S"} + replace_item(view, "app_doc1", {"payload": "OK"}) # in scope: allowed + assert _state(c)["app_doc1"] == "OK" + + +def test_batch_key_created_earlier_in_the_batch_needs_no_membership(): + c = _container() + batch_items( + _class_wrapped(c), [("upsert", "fresh", {"payload": "1"}), ("replace", "fresh", {"payload": "2"})] + ) + assert _state(c)["app_fresh"] == "2" + + +def test_batch_per_op_kwargs_and_sdk_triples(): + c = _container() + etag = c._data[(PK, "app_doc1")]["_etag"] + batch_items(_class_wrapped(c), [("replace", "doc1", {"payload": "E"}, {"if_match_etag": etag})]) + assert _state(c)["app_doc1"] == "E" + with pytest.raises(ValueError, match="SDK"): + batch_items(_class_wrapped(c), [("upsert", ({"id": "x"},), {})]) + + +def test_scope_filter_stops_forced_container_delete(): + from dol import filt_iter + + db = _db() + view = filt_iter(CosmosDatabase(db), filt=lambda k: k.startswith(PREFIX)) + with pytest.raises(Exception, match="events"): + delete_container(view, "events", force=True) + assert sorted(db.containers) == ["app_events", "events"] + + +def test_add_database_does_not_raise_after_creating(): + from dol import filt_iter + + acct, client = _account(True) + hidden = filt_iter(acct, filt=lambda k: k.startswith("x")) + db = add_database(hidden, "extra") + assert db.name == "app_extra" and "app_extra" in client.dbs + + +def test_forced_delete_handles_hierarchical_and_missing_partition_keys(): + from cosmodol.ops import NonePartitionKeyValue, _partition_key_expr + + multi = _FakeContainer("m", []) + multi.read = lambda: {"id": "m", "partitionKey": {"paths": ["/t", "/u/v"], "kind": "MultiHash"}} + assert _partition_key_expr(multi) == '[c["t"], c["u"]["v"]]' + + db = _db() + target = db.containers["events"] + target.query_items = lambda query, **kw: iter([{"id": "a", "pk": ["T", "U"]}, {"id": "b"}]) + delete_container(CosmosDatabase(db), "events", force=True) + assert target.deleted == [("a", ["T", "U"]), ("b", NonePartitionKeyValue)] diff --git a/pyproject.toml b/pyproject.toml index b9eadf3..2a6b95a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,7 +31,7 @@ classifiers = [ "Topic :: Database", ] dependencies = [ - "dol", + "dol>=0.3.59", # inner_most_key exported from dol (cosmodol.ops) "azure-cosmos", ]