diff --git a/CHANGELOG.md b/CHANGELOG.md index ac3f1c4..36b0171 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.0.20] - 2026-09-18 + +### Added + +- **`Database.insert_if_absent` / `Object.create_if_absent`.** Atomic + insert-if-absent on primary key `id` across SQLite (`INSERT OR IGNORE`), + Postgres (`ON CONFLICT DO NOTHING RETURNING`), MongoDB (`insert_one` + + `DuplicateKeyError`), JsonDB (path lock), and DynamoDB + (`attribute_not_exists(id)`). Returns `InsertIfAbsentResult(record, created)` + / `(entity, created)` without calling through upsert `save()`. Wrappers + (`CachingDatabase`, `ObservableDatabase`) forward correctly. + `DeferredSaveMixin.flush()` runs only when `created=True`. + ## [0.0.19] - 2026-09-14 ### Removed diff --git a/SPEC.md b/SPEC.md index ae9ce65..bcd6306 100644 --- a/SPEC.md +++ b/SPEC.md @@ -102,7 +102,7 @@ AttributeMixin + pydantic.BaseModel `jvspatial/core/entities/object.py`. Provides: - `id`, `entity`, `type_code` fields -- CRUD methods: `create()`, `get()`, `find()`, `save()`, `delete()`, `count()`, `export()` +- CRUD methods: `create()`, `create_if_absent()`, `get()`, `find()`, `save()`, `delete()`, `count()`, `export()` - Context lookup: `set_context()`, `get_context()` (default via `get_default_context()`) - Collection mapping via `get_collection_name()` → `{n: node, e: edge, o: object, w: walker}` @@ -211,6 +211,7 @@ When `DeferredSaveMixin` is mixed into an entity *and* `deferred_saves_globally_ | Method | Required | Description | |---|---|---| | `save(collection, data)` | Yes | Insert-or-replace by ID; returns saved record | +| `insert_if_absent(collection, data, *, conflict_target="id")` | Yes (built-ins) | Atomic insert-if-absent on primary key `id` only (v1); returns `InsertIfAbsentResult(record, created)`. Never updates. Default ABC raises `NotImplementedError` after validating args. | | `get(collection, id)` | Yes | Fetch by ID or `None` | | `delete(collection, id)` | Yes | Idempotent delete by ID | | `find(collection, query, *, limit, sort)` | Yes | Mongo-style query; returns list. Ordering contract below | diff --git a/docs/md/entity-reference.md b/docs/md/entity-reference.md index 6403125..7e52d9e 100644 --- a/docs/md/entity-reference.md +++ b/docs/md/entity-reference.md @@ -21,6 +21,8 @@ class Object(BaseModel): @classmethod async def create(cls, **kwargs) -> "Object" @classmethod + async def create_if_absent(cls, **kwargs) -> tuple["Object", bool] + @classmethod async def find(cls, query: Optional[dict] = None, **filters) -> List["Object"] @classmethod async def find_one(cls, query: Optional[dict] = None, **filters) -> Optional["Object"] @@ -30,6 +32,12 @@ class Object(BaseModel): async def export() -> dict ``` +**`create_if_absent`** — persist only when no row exists for the given `id`. +Returns `(entity, created)`. On conflict, rehydrates the **stored** winner +(never merges proposed fields). Does not call through upsert `save()`. +Requires a deterministic `id` for idempotency; auto-generated ids always +create. See `Database.insert_if_absent` / `InsertIfAbsentResult`. + **Convenience Methods** - `await Object.count()` → count all objects of that type diff --git a/docs/md/stability.md b/docs/md/stability.md index b9218a3..1113cb2 100644 --- a/docs/md/stability.md +++ b/docs/md/stability.md @@ -17,15 +17,18 @@ bump (post-1.0) or a minor bump (pre-1.0) and must be called out in These names are exported from `jvspatial/__init__.py`'s `__all__` and are the canonical import path: -- **Core entities.** `Object`, `Node`, `Edge`, `Walker`, `Root`, - `GraphContext`. +- **Core entities.** `Object` (including `Object.create_if_absent`), + `Node`, `Edge`, `Walker`, `Root`, `GraphContext`. - **Decorators.** `attribute`, `endpoint`. - **Server / config.** `Server`, `ServerConfig`. -- **Database.** `Database`, `create_database`. The +- **Database.** `Database`, `create_database`, `InsertIfAbsentResult`. The `Database.supports_transactions` capability flag is part of this surface, as are the bulk methods `Database.find_many` and - `Database.bulk_save`. Adapters not overriding the bulk methods - fall through to the (slower) default serial implementations. + `Database.bulk_save`, and `Database.insert_if_absent` (v1: + `conflict_target="id"` only). Adapters not overriding the bulk methods + fall through to the (slower) default serial implementations. Custom + adapters must implement `insert_if_absent` or callers hit + `NotImplementedError`. - **Cache.** `create_cache`. - **Mixins.** `DeferredSaveMixin`, `deferred_saves_globally_allowed`, `flush_deferred_entities`. diff --git a/docs/superpowers/plans/2026-09-18-create-if-absent.md b/docs/superpowers/plans/2026-09-18-create-if-absent.md new file mode 100644 index 0000000..863e823 --- /dev/null +++ b/docs/superpowers/plans/2026-09-18-create-if-absent.md @@ -0,0 +1,48 @@ +# Object.create_if_absent Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Ship portable atomic `Database.insert_if_absent` + `Object.create_if_absent` without changing upsert `save()`. + +**Architecture:** New ABC result type + method; each adapter implements insert-or-return-existing on primary `id`; Object layer exports, calls DB, rehydrates on conflict. Never route through `save()`. + +**Tech Stack:** Python 3.9+, aiosqlite, asyncpg, motor, pytest-asyncio. + +**Spec:** `docs/superpowers/specs/2026-09-18-create-if-absent-design.md` + +--- + +### Task 1: ABC + SQLite + Object + tests + +**Files:** +- Modify: `jvspatial/db/database.py` +- Modify: `jvspatial/db/sqlite.py` +- Modify: `jvspatial/core/entities/object.py` +- Modify: `jvspatial/core/context.py` (if needed for cache) +- Create: `tests/core/test_object_create_if_absent.py` +- Create: `tests/db/test_sqlite_insert_if_absent.py` + +- [ ] RED/GREEN Object + SQLite concurrent same-id +- [ ] Commit + +### Task 2: Postgres, Mongo, JsonDB, DynamoDB, wrappers + +**Files:** +- Modify: `jvspatial/db/postgres.py` (+ transaction save path if separate) +- Modify: `jvspatial/db/mongodb.py` +- Modify: `jvspatial/db/jsondb.py` +- Modify: `jvspatial/db/dynamodb.py` (if present) +- Modify: `jvspatial/db/_cache.py`, `jvspatial/db/_observable.py` +- Extend integration tests + +- [ ] RED/GREEN per adapter +- [ ] Commit + +### Task 3: Docs, version, changelog, stability, PR + +**Files:** +- Modify: `CHANGELOG.md`, `jvspatial/version.py`, `docs/md/stability.md`, `docs/md/entity-reference.md`, `SPEC.md` if needed +- Export `__all__` + +- [ ] Docs + 0.0.20 bump +- [ ] Push + `gh pr create` diff --git a/docs/superpowers/specs/2026-09-18-create-if-absent-design.md b/docs/superpowers/specs/2026-09-18-create-if-absent-design.md new file mode 100644 index 0000000..eab0579 --- /dev/null +++ b/docs/superpowers/specs/2026-09-18-create-if-absent-design.md @@ -0,0 +1,134 @@ +# Design: `Object.create_if_absent` / `Database.insert_if_absent` + +**Date:** 2026-09-18 +**Status:** Proposed +**Target version:** 0.0.20 (minor, pre-1.0) +**Motivation:** Durable idempotency records need atomic create-or-return-existing semantics. Today every `Database.save` path is upsert (`INSERT OR REPLACE` / `ON CONFLICT DO UPDATE` / Mongo `replace_one(upsert=True)`), so concurrent writers can overwrite peers and SQLite unique collisions can delete the wrong row via `OR REPLACE`. + +## Goal + +Ship a portable, atomic **insert-if-absent** primitive: + +1. Persist a new record only when no conflict exists on the target key. +2. Never update or replace an existing row. +3. Return the **stored** winner plus a boolean `created` flag. +4. Work identically (semantic contract) across SQLite, Postgres, MongoDB, JsonDB, and DynamoDB. + +Integral Core will use this for `QueryResultSet` and similar receipt/idempotency Objects after jvspatial ships. + +## Non-goals + +- Changing `save()` / `create()` upsert semantics. +- Business-key `ON CONFLICT` on arbitrary unique indexes in v1 (phase 2). +- Automatic migration of webhook `get_or_create_idempotency_key` (follow-up once unique index exists). +- Cross-document multi-key transactions on backends without transactions. + +## Public contract + +### `InsertIfAbsentResult` + +```python +@dataclass(frozen=True) +class InsertIfAbsentResult: + record: Dict[str, Any] # stored document (existing or newly inserted) + created: bool +``` + +### `Database.insert_if_absent` + +```python +async def insert_if_absent( + self, + collection: str, + data: Dict[str, Any], + *, + conflict_target: str = "id", +) -> InsertIfAbsentResult: + ... +``` + +**v1 rules:** + +- `conflict_target` must be `"id"` (primary key). Other values → `ValueError`. +- Missing / empty `id` in `data` → `ValueError` (same as existing Postgres payload split). +- On insert success: `created=True`, return the inserted record (normalized as other writes). +- On conflict: `created=False`, **load and return the existing stored record unchanged** (no merge of proposed fields). +- Must not call through `save()`. + +### `Object.create_if_absent` + +```python +@classmethod +async def create_if_absent(cls, **kwargs) -> tuple["Object", bool]: + """Create and persist only if absent. + + Idempotent when the caller supplies a deterministic ``id``. + Unlike ``create()`` / ``save()``, never updates an existing row. + """ +``` + +**Flow:** + +1. `obj = cls(**kwargs)` (auto-id if omitted — same as `create`). +2. `await context.ensure_indexes(cls)`. +3. Export record; `await database.insert_if_absent(collection, record)`. +4. If `created=False`, rehydrate entity from stored record (not the proposed in-memory instance). +5. Attach `_graph_context`; update entity cache consistently with `save`. +6. If `DeferredSaveMixin` and `created`, `await flush()` (mirror `create()`). +7. Return `(entity, created)`. + +Optional kwargs **not** in v1: `raise_on_conflict`. Callers inspect `created`. + +## Adapter semantics + +| Backend | Insert path | Conflict path | +|---------|-------------|---------------| +| **Postgres** | `INSERT ... ON CONFLICT (id) DO NOTHING RETURNING data`; if no row returned, `SELECT` by id | Return stored row, `created=False` | +| **SQLite** | `INSERT OR IGNORE INTO records ...`; if `changes()==0`, `SELECT` | **Never** `INSERT OR REPLACE` | +| **MongoDB** | `insert_one`; catch `DuplicateKeyError` → `find_one` | Return stored doc | +| **JsonDB** | Under path lock: if file exists → read; else write | Same-id atomic via lock; no secondary unique enforcement | +| **DynamoDB** | `PutItem` with `attribute_not_exists(id)` | Conditional check fail → `GetItem` | + +Wrappers (`CachingDatabase`, `ObservableDatabase`) must override or forward `insert_if_absent` so cache/metrics stay correct on both created and existing paths. + +Default ABC implementation: raise `NotImplementedError` with a clear message, **or** a documented non-atomic find-then-insert only if PRD requires a default — prefer requiring every built-in adapter to implement (PRD §8). + +## Error handling + +- `ValueError` — missing id, unsupported `conflict_target`. +- `DuplicateEntityError` — **not** raised by default in v1 (reserved for optional `raise_on_conflict=True` later). +- Adapter-specific DB errors propagate as today (`DatabaseError` subclasses where applicable). + +## Concurrency guarantees + +- **Same deterministic `id`:** at most one insert wins; losers observe `created=False` and the winner's stored payload. +- **Different ids, same business key:** not atomic in v1. Callers that need business-key uniqueness must declare a unique index **and** wait for phase 2 `conflict_target` extension, or encode the business key into a deterministic `id` (Integral pattern). + +## Testing requirements + +- Unit/integration: create → absent inserts; second call same id returns existing, no field overwrite. +- Concurrent: N parallel `create_if_absent` with same id → exactly one `created=True`, all return identical stored payload. +- Regression: `save()` remains upsert (`test_save_is_upsert` unchanged). +- SQLite: prove unique secondary collision via this API does **not** delete rows (contrast `OR REPLACE` hazard). +- Object-layer: rehydrate on miss; DeferredSaveMixin flush only when created. +- Adapters covered: SQLite (always), Postgres (CI postgres job), MongoDB/JsonDB as existing suite patterns allow. + +## Docs / stability / release + +- Document in `docs/md/entity-reference.md`, `SPEC.md` persistence section, `docs/md/stability.md` (public: `Object.create_if_absent`, `Database.insert_if_absent`, `InsertIfAbsentResult`). +- `CHANGELOG.md` under `[Unreleased]` → shipped as `[0.0.20]`. +- Bump `jvspatial/version.py` to `0.0.20` when merging to main per VERSIONING workflow. +- Export new names from public `__all__` where appropriate. + +## Phase 2 (out of this PR) + +- `conflict_target` naming unique indexes / compound keys from `get_indexes()`. +- Migrate webhook idempotency helper off find-then-save. +- Optional `raise_on_conflict: bool = False`. + +## Success criteria + +1. Concurrent identical-id creates never overwrite stored data on SQLite and Postgres. +2. `save()` behavior unchanged. +3. Public API documented and exported. +4. Integral can replace race-prone QueryResultSet create with `create_if_absent` in a follow-up bump. diff --git a/jvspatial/__init__.py b/jvspatial/__init__.py index 4a9dbbe..9df818b 100644 --- a/jvspatial/__init__.py +++ b/jvspatial/__init__.py @@ -74,7 +74,7 @@ ) # Simplified database and cache -from .db import Database, create_database +from .db import Database, InsertIfAbsentResult, create_database from .db.work_claim import claim_record, delete_claimed_record, release_claim # Observability primitives @@ -140,6 +140,7 @@ "Server", # Database & Cache "Database", + "InsertIfAbsentResult", "create_database", "create_cache", # Observability diff --git a/jvspatial/core/entities/object.py b/jvspatial/core/entities/object.py index acbef62..fce7070 100644 --- a/jvspatial/core/entities/object.py +++ b/jvspatial/core/entities/object.py @@ -191,6 +191,109 @@ async def create(cls: Type["Object"], **kwargs: Any) -> "Object": await flush_fn() return obj + @classmethod + async def create_if_absent( + cls: Type["Object"], **kwargs: Any + ) -> tuple["Object", bool]: + """Create and persist only if absent for the given ``id``. + + Idempotent when the caller supplies a deterministic ``id``. + Unlike :meth:`create` / :meth:`save`, never updates an existing row. + Returns ``(entity, created)`` where ``entity`` is the stored winner + (rehydrated from the database when ``created`` is ``False``). + + Does not call through :meth:`save`. For + :class:`~jvspatial.core.mixins.DeferredSaveMixin` types, ``flush()`` + runs only when ``created`` is ``True`` (mirrors :meth:`create`'s + end-of-batch clear when the insert won). + """ + obj = cls(**kwargs) + context = await obj.get_context() + await context.ensure_indexes(cls) + + # Build the persistence record the same way GraphContext.save does, + # but route through insert_if_absent instead of save. + if hasattr(obj, "type_code"): + type_code = getattr(obj, "type_code", "") + if type_code in ("n", "e", "w"): + record = await obj.export() + if "entity" not in record: + record["entity"] = obj.entity + else: + record = await obj.export() + entity_id = getattr(obj, "id", None) + if entity_id: + id_parts = entity_id.split(".") + entity_name_resolver = getattr(cls, "_entity_name", None) + expected_entity_name = ( + entity_name_resolver() + if callable(entity_name_resolver) + else cls.__name__ + ) + if ( + len(id_parts) != 3 + or id_parts[0] != obj.type_code + or id_parts[1] != expected_entity_name + ): + new_id = generate_id(obj.type_code, expected_entity_name) + object.__setattr__(obj, "id", new_id) + record["id"] = new_id + else: + record = await obj.export() + from jvspatial.utils.serialization import serialize_datetime + + record = serialize_datetime(record) + if "entity" not in record and hasattr(obj, "entity"): + record["entity"] = obj.entity + + from jvspatial.utils.normalization import ( + is_text_normalization_enabled, + normalize_data, + ) + + if is_text_normalization_enabled(): + record = normalize_data(record) + + if hasattr(obj, "get_collection_name"): + collection = obj.get_collection_name() + else: + type_code = obj.type_code + if type_code == "n": + collection = "node" + elif type_code == "e": + collection = "edge" + else: + collection = type_code.lower() + + is_node = hasattr(obj, "type_code") and getattr(obj, "type_code", "") == "n" + if is_node: + record.pop("edges", None) + + result = await context.database.insert_if_absent(collection, record) + + if result.created: + entity: Object = obj + await context._add_to_cache(entity.id, entity) + flush_fn = getattr(entity, "flush", None) + if ( + flush_fn is not None + and callable(flush_fn) + and inspect.iscoroutinefunction(flush_fn) + ): + await flush_fn() + return entity, True + + entity_opt = await context._deserialize_entity(cls, result.record) + if entity_opt is None: + raise RuntimeError( + f"create_if_absent could not rehydrate {cls.__name__} " + f"from stored record id={result.record.get('id')!r}" + ) + entity = entity_opt + entity._graph_context = context + await context._add_to_cache(entity.id, entity) + return entity, False + async def update( self: "Object", properties: Dict[str, Any], diff --git a/jvspatial/db/__init__.py b/jvspatial/db/__init__.py index e77db8e..b0290c0 100644 --- a/jvspatial/db/__init__.py +++ b/jvspatial/db/__init__.py @@ -5,7 +5,12 @@ a prime default database for core persistence operations. """ -from .database import Database, DatabaseError, VersionConflictError +from .database import ( + Database, + DatabaseError, + InsertIfAbsentResult, + VersionConflictError, +) from .factory import ( create_database, create_default_database, @@ -57,6 +62,7 @@ "Database", "DatabaseError", "VersionConflictError", + "InsertIfAbsentResult", "create_database", "create_default_database", "get_prime_database", diff --git a/jvspatial/db/_cache.py b/jvspatial/db/_cache.py index f447d28..a6b7d94 100644 --- a/jvspatial/db/_cache.py +++ b/jvspatial/db/_cache.py @@ -38,7 +38,7 @@ from collections import OrderedDict from typing import Any, Dict, List, Optional, Tuple, Union -from jvspatial.db.database import BulkSaveResult, Database +from jvspatial.db.database import BulkSaveResult, Database, InsertIfAbsentResult from jvspatial.runtime.serverless import is_serverless_mode logger = logging.getLogger(__name__) @@ -158,6 +158,22 @@ async def save(self, collection: str, data: Dict[str, Any]) -> Dict[str, Any]: self._cache_put(collection, str(rec_id), dict(result)) return result + async def insert_if_absent( + self, + collection: str, + data: Dict[str, Any], + *, + conflict_target: str = "id", + ) -> InsertIfAbsentResult: + """Forward insert-if-absent and cache the stored winner.""" + result = await self.inner.insert_if_absent( + collection, data, conflict_target=conflict_target + ) + rec_id = result.record.get("id", result.record.get("_id")) + if rec_id is not None and self._enabled(): + self._cache_put(collection, str(rec_id), dict(result.record)) + return result + async def get(self, collection: str, id: str) -> Optional[Dict[str, Any]]: """Read with cache: hit on the in-memory map, else fetch + cache.""" if not self._enabled(): diff --git a/jvspatial/db/_observable.py b/jvspatial/db/_observable.py index 85ad79e..c88f9cd 100644 --- a/jvspatial/db/_observable.py +++ b/jvspatial/db/_observable.py @@ -47,7 +47,7 @@ Union, ) -from jvspatial.db.database import BulkSaveResult, Database +from jvspatial.db.database import BulkSaveResult, Database, InsertIfAbsentResult from jvspatial.observability import db_op_counter from jvspatial.observability.metrics import ( MetricsRecorder, @@ -212,6 +212,23 @@ async def save(self, collection: str, data: Dict[str, Any]) -> Dict[str, Any]: "save", collection, lambda: self.inner.save(collection, data) ) + async def insert_if_absent( + self, + collection: str, + data: Dict[str, Any], + *, + conflict_target: str = "id", + ) -> InsertIfAbsentResult: + """Instrumented ``insert_if_absent`` (emits structured log + metric).""" + return await self._instrument( + "insert_if_absent", + collection, + lambda: self.inner.insert_if_absent( + collection, data, conflict_target=conflict_target + ), + result_count_extractor=lambda r: (1 if getattr(r, "created", False) else 0), + ) + async def get(self, collection: str, id: str) -> Optional[Dict[str, Any]]: """Instrumented ``get`` (emits structured log + metric).""" return await self._instrument( diff --git a/jvspatial/db/database.py b/jvspatial/db/database.py index 525ee49..5ee93c6 100644 --- a/jvspatial/db/database.py +++ b/jvspatial/db/database.py @@ -67,6 +67,19 @@ def all_saved(self) -> bool: return self.saved == self.attempted and not self.failed_ids +@dataclass(frozen=True) +class InsertIfAbsentResult: + """Outcome of :meth:`Database.insert_if_absent`. + + ``record`` is the **stored** document (newly inserted or the existing + winner). ``created`` is ``True`` only when this call performed the + insert. Never merges proposed fields into an existing row. + """ + + record: Dict[str, Any] + created: bool + + logger = logging.getLogger(__name__) @@ -215,6 +228,60 @@ async def save(self, collection: str, data: Dict[str, Any]) -> Dict[str, Any]: """ pass + @staticmethod + def _validate_insert_if_absent(data: Dict[str, Any], conflict_target: str) -> str: + """Validate v1 ``insert_if_absent`` arguments; return the record id. + + Raises: + ValueError: ``conflict_target`` is not ``'id'``, or ``data`` + lacks a non-empty ``id``. + """ + if conflict_target != "id": + raise ValueError( + "insert_if_absent conflict_target must be 'id' in v1; " + f"got {conflict_target!r}" + ) + if not data or data.get("id") is None or data.get("id") == "": + raise ValueError( + "insert_if_absent requires data with a non-empty 'id' field" + ) + return str(data["id"]) + + async def insert_if_absent( + self, + collection: str, + data: Dict[str, Any], + *, + conflict_target: str = "id", + ) -> InsertIfAbsentResult: + """Insert ``data`` only when no row exists for ``conflict_target``. + + v1 supports ``conflict_target='id'`` (primary key) only. On + conflict, returns the **existing stored** record unchanged + (``created=False``). Must not call through :meth:`save`. + + Built-in adapters override this with an atomic backend primitive. + The default raises :class:`NotImplementedError` after validation + so custom adapters fail loudly until they implement it. + + Args: + collection: Collection name + data: Record data (must include non-empty ``id``) + conflict_target: Conflict key; must be ``'id'`` in v1 + + Returns: + :class:`InsertIfAbsentResult` with the stored record and + whether this call created it + + Raises: + ValueError: Invalid ``conflict_target`` or missing ``id`` + NotImplementedError: Adapter has not implemented this method + """ + self._validate_insert_if_absent(data, conflict_target) + raise NotImplementedError( + f"{type(self).__name__} does not implement insert_if_absent" + ) + @abstractmethod async def get(self, collection: str, id: str) -> Optional[Dict[str, Any]]: """Retrieve a record by ID. @@ -647,6 +714,7 @@ async def drop_deprecated_indexes(self, deprecated: Dict[str, List[str]]) -> Non "DatabaseError", "VersionConflictError", "BulkSaveResult", + "InsertIfAbsentResult", "encode_cursor", "decode_cursor", "finalize_find_results", diff --git a/jvspatial/db/dynamodb.py b/jvspatial/db/dynamodb.py index 2e1b0d0..44f16bb 100644 --- a/jvspatial/db/dynamodb.py +++ b/jvspatial/db/dynamodb.py @@ -36,7 +36,7 @@ ClientError = Exception # type: ignore[assignment, misc] Config = None # type: ignore[assignment, misc] -from jvspatial.db.database import Database, finalize_find_results +from jvspatial.db.database import Database, InsertIfAbsentResult, finalize_find_results from jvspatial.db.query import QueryEngine from jvspatial.exceptions import DatabaseError from jvspatial.utils.retry import retry_async @@ -553,6 +553,65 @@ async def _put_op() -> None: await self._run_with_throttle_retry("save", _put_op) return data + async def insert_if_absent( + self, + collection: str, + data: Dict[str, Any], + *, + conflict_target: str = "id", + ) -> InsertIfAbsentResult: + """Insert via PutItem when id is absent; on conflict, GetItem.""" + record_id = self._validate_insert_if_absent(data, conflict_target) + payload = dict(data) + payload["id"] = record_id + table_name = await self._ensure_table_exists(collection) + + item = { + "collection": {"S": collection}, + "id": {"S": record_id}, + "data": {"S": json.dumps(payload, default=str)}, + } + indexed_attrs = self._extract_indexed_fields(payload, collection) + item.update(indexed_attrs) + + async def _put_op() -> InsertIfAbsentResult: + client = await self._get_client() + try: + await asyncio.wait_for( + client.put_item( + TableName=table_name, + Item=item, + ConditionExpression="attribute_not_exists(id)", + ), + timeout=30.0, + ) + return InsertIfAbsentResult(record=payload, created=True) + except ClientError as e: + code = e.response.get("Error", {}).get("Code", "") + if code != "ConditionalCheckFailedException": + raise + response = await client.get_item( + TableName=table_name, + Key={ + "collection": {"S": collection}, + "id": {"S": record_id}, + }, + ) + if "Item" not in response: + raise DatabaseError( + "insert_if_absent conditional check failed but no " + f"row with id={record_id!r} exists in collection " + f"{collection!r}" + ) from e + existing = json.loads(response["Item"]["data"]["S"]) + return InsertIfAbsentResult(record=existing, created=False) + except asyncio.TimeoutError: + raise DatabaseError( + f"DynamoDB insert_if_absent timed out for table: {table_name}" + ) + + return await self._run_with_throttle_retry("insert_if_absent", _put_op) + async def get(self, collection: str, id: str) -> Optional[Dict[str, Any]]: """Retrieve a record by ID. diff --git a/jvspatial/db/jsondb.py b/jvspatial/db/jsondb.py index d2b828a..c055122 100644 --- a/jvspatial/db/jsondb.py +++ b/jvspatial/db/jsondb.py @@ -32,7 +32,7 @@ from jvspatial.db._atomic import atomic_write_bytes, cleanup_orphan_tmp_files from jvspatial.db._path_locks import PathLockManager -from jvspatial.db.database import Database, finalize_find_results +from jvspatial.db.database import Database, InsertIfAbsentResult, finalize_find_results from jvspatial.db.query import QueryEngine from jvspatial.runtime.serverless import is_serverless_mode @@ -251,6 +251,40 @@ async def save(self, collection: str, data: Dict[str, Any]) -> Dict[str, Any]: await asyncio.to_thread(self._sync_write_record, collection, dict(data)) return data + async def insert_if_absent( + self, + collection: str, + data: Dict[str, Any], + *, + conflict_target: str = "id", + ) -> InsertIfAbsentResult: + """Insert under path lock only when the record file is absent.""" + record_id = self._validate_insert_if_absent(data, conflict_target) + payload = dict(data) + payload["id"] = record_id + + def _sync_insert_if_absent() -> InsertIfAbsentResult: + record_path = self._get_record_path(collection, record_id) + with self._path_locks.lock(str(record_path)): + if record_path.exists(): + try: + with open(record_path, "rb") as f: + existing = _loads(f.read()) + except (ValueError, OSError): + existing = None + if existing is None: + from jvspatial.db.database import DatabaseError + + raise DatabaseError( + "insert_if_absent found an unreadable existing " + f"file for id={record_id!r}" + ) + return InsertIfAbsentResult(record=existing, created=False) + atomic_write_bytes(record_path, _dumps(payload)) + return InsertIfAbsentResult(record=payload, created=True) + + return await asyncio.to_thread(_sync_insert_if_absent) + async def get(self, collection: str, id: str) -> Optional[Dict[str, Any]]: """Retrieve a record by ID.""" record_path = self._get_record_path(collection, id) diff --git a/jvspatial/db/mongodb.py b/jvspatial/db/mongodb.py index 0dde942..53d8d70 100644 --- a/jvspatial/db/mongodb.py +++ b/jvspatial/db/mongodb.py @@ -34,12 +34,13 @@ from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase from pymongo.errors import ( ConnectionFailure, + DuplicateKeyError, OperationFailure, PyMongoError, ServerSelectionTimeoutError, ) -from jvspatial.db.database import Database +from jvspatial.db.database import Database, InsertIfAbsentResult from jvspatial.exceptions import DatabaseError from jvspatial.utils.retry import retry_async @@ -295,6 +296,43 @@ async def _save_op() -> Dict[str, Any]: return await self._run_with_reconnect("save", _save_op) + async def insert_if_absent( + self, + collection: str, + data: Dict[str, Any], + *, + conflict_target: str = "id", + ) -> InsertIfAbsentResult: + """Insert via ``insert_one``; on DuplicateKeyError, ``find_one`` winner.""" + record_id = self._validate_insert_if_absent(data, conflict_target) + payload = dict(data) + payload["id"] = record_id + if "_id" not in payload: + payload["_id"] = record_id + + async def _insert_op() -> InsertIfAbsentResult: + await self._ensure_connected() + if self._db is None: + raise DatabaseError("MongoDB database connection not established") + collection_obj = self._db[collection] + try: + await collection_obj.insert_one(payload) + return InsertIfAbsentResult(record=payload, created=True) + except DuplicateKeyError: + existing = await collection_obj.find_one({"_id": record_id}) + if existing is None: + # Rare: conflict on a secondary unique index, not _id + existing = await collection_obj.find_one({"id": record_id}) + if existing is None: + raise DatabaseError( + "insert_if_absent hit DuplicateKeyError but no row " + f"with id={record_id!r} exists in collection " + f"{collection!r}" + ) + return InsertIfAbsentResult(record=existing, created=False) + + return await self._run_with_reconnect("insert_if_absent", _insert_op) + async def get(self, collection: str, id: str) -> Optional[Dict[str, Any]]: """Retrieve a record by ID.""" diff --git a/jvspatial/db/postgres.py b/jvspatial/db/postgres.py index 2d2543c..f42d7ae 100644 --- a/jvspatial/db/postgres.py +++ b/jvspatial/db/postgres.py @@ -88,6 +88,8 @@ from .database import ( BulkSaveResult, Database, + DatabaseError, + InsertIfAbsentResult, decode_cursor, finalize_find_results, ) @@ -766,6 +768,54 @@ async def save(self, collection: str, data: Dict[str, Any]) -> Dict[str, Any]: ) return data + async def insert_if_absent( + self, + collection: str, + data: Dict[str, Any], + *, + conflict_target: str = "id", + ) -> InsertIfAbsentResult: + """Insert only when ``id`` is absent; never ``ON CONFLICT DO UPDATE``. + + Uses ``INSERT ... ON CONFLICT (id) DO NOTHING RETURNING data``. When + no row is returned, ``SELECT`` loads the existing stored record. + Honors tenant scoping via :meth:`_acquire_conn`. + """ + self._validate_insert_if_absent(data, conflict_target) + await self._bootstrap_collection(collection) + rec_id, entity, tenant, data_json = self._split_payload(data) + col = _safe_collection(collection) + schema = _safe_collection(self.schema_name) + + async with self._acquire_conn() as conn: + row = await conn.fetchrow( + f""" + INSERT INTO {schema}.{col} (id, entity, tenant_id, data, updated_at) + VALUES ($1, $2, $3, $4::jsonb, NOW()) + ON CONFLICT (id) DO NOTHING + RETURNING data + """, + rec_id, + entity, + tenant, + data_json, + ) + if row is not None: + return InsertIfAbsentResult( + record=self._record_from_row(row), created=True + ) + existing = await conn.fetchrow( + f"SELECT data FROM {schema}.{col} WHERE id = $1", rec_id + ) + if existing is None: + raise DatabaseError( + "insert_if_absent conflicted but no row with " + f"id={rec_id!r} exists in collection {collection!r}" + ) + return InsertIfAbsentResult( + record=self._record_from_row(existing), created=False + ) + async def strip_node_edges( self, collection: str = "node", @@ -2170,6 +2220,47 @@ async def save(self, collection: str, data: Dict[str, Any]) -> Dict[str, Any]: ) return data + async def insert_if_absent( + self, + collection: str, + data: Dict[str, Any], + *, + conflict_target: str = "id", + ) -> InsertIfAbsentResult: + """Insert-if-absent within this transaction (same SQL as PostgresDB).""" + self._db._validate_insert_if_absent(data, conflict_target) + col = _safe_collection(collection) + schema = _safe_collection(self._db.schema_name) + await self._db._bootstrap_collection(collection) + rec_id, entity, tenant, data_json = self._db._split_payload(data) + row = await self._connection.fetchrow( + f""" + INSERT INTO {schema}.{col} (id, entity, tenant_id, data, updated_at) + VALUES ($1, $2, $3, $4::jsonb, NOW()) + ON CONFLICT (id) DO NOTHING + RETURNING data + """, + rec_id, + entity, + tenant, + data_json, + ) + if row is not None: + return InsertIfAbsentResult( + record=self._db._record_from_row(row), created=True + ) + existing = await self._connection.fetchrow( + f"SELECT data FROM {schema}.{col} WHERE id = $1", rec_id + ) + if existing is None: + raise DatabaseError( + "insert_if_absent conflicted but no row with " + f"id={rec_id!r} exists in collection {collection!r}" + ) + return InsertIfAbsentResult( + record=self._db._record_from_row(existing), created=False + ) + async def get(self, collection: str, id: str) -> Optional[Dict[str, Any]]: """Fetch a single record by ``id`` from ``collection`` in this transaction.""" col = _safe_collection(collection) diff --git a/jvspatial/db/sqlite.py b/jvspatial/db/sqlite.py index d419004..d298588 100644 --- a/jvspatial/db/sqlite.py +++ b/jvspatial/db/sqlite.py @@ -31,7 +31,7 @@ translate_query, translate_sort, ) -from .database import Database, finalize_find_results +from .database import Database, InsertIfAbsentResult, finalize_find_results from .query import QueryEngine logger = logging.getLogger(__name__) @@ -503,6 +503,52 @@ async def save(self, collection: str, data: Dict[str, Any]) -> Dict[str, Any]: await connection.commit() return record + async def insert_if_absent( + self, + collection: str, + data: Dict[str, Any], + *, + conflict_target: str = "id", + ) -> InsertIfAbsentResult: + """Insert only when ``id`` is absent. Never ``INSERT OR REPLACE``. + + Uses ``INSERT OR IGNORE`` so primary-key (and other unique) conflicts + leave existing rows untouched. On ignore with an existing same-id + row, returns that stored record with ``created=False``. + """ + record_id = self._validate_insert_if_absent(data, conflict_target) + async with self._lock: + connection = await self._get_connection() + record = data.copy() + record["id"] = record_id + payload = json.dumps(record) + cursor = await connection.execute( + """ + INSERT OR IGNORE INTO records (collection, id, data) + VALUES (?, ?, ?) + """, + (collection, record_id, payload), + ) + await connection.commit() + if cursor.rowcount and cursor.rowcount > 0: + return InsertIfAbsentResult(record=record, created=True) + # Conflict (or ignore for another unique constraint): load by id. + sel = await connection.execute( + "SELECT data FROM records WHERE collection = ? AND id = ?", + (collection, record_id), + ) + row = await sel.fetchone() + await sel.close() + if row is None: + from jvspatial.db.database import DatabaseError + + raise DatabaseError( + "insert_if_absent ignored a constraint conflict but no " + f"row with id={record_id!r} exists in collection " + f"{collection!r}" + ) + return InsertIfAbsentResult(record=json.loads(row["data"]), created=False) + async def get(self, collection: str, id: str) -> Optional[Dict[str, Any]]: """Retrieve a record from the database. diff --git a/jvspatial/version.py b/jvspatial/version.py index 294dbbb..b5c8c4a 100644 --- a/jvspatial/version.py +++ b/jvspatial/version.py @@ -9,4 +9,4 @@ # - MAJOR: Breaking changes # - MINOR: New features, backward compatible # - PATCH: Bug fixes, backward compatible -__version__ = "0.0.19" +__version__ = "0.0.20" diff --git a/tests/core/test_create_if_absent.py b/tests/core/test_create_if_absent.py new file mode 100644 index 0000000..38023d6 --- /dev/null +++ b/tests/core/test_create_if_absent.py @@ -0,0 +1,113 @@ +"""Object.create_if_absent — atomic create-or-return-existing.""" + +from __future__ import annotations + +import tempfile +from typing import Any +from unittest.mock import AsyncMock, patch + +import pytest +from pydantic import Field + +from jvspatial.core.context import GraphContext, set_default_context +from jvspatial.core.entities import Object +from jvspatial.core.mixins import DeferredSaveMixin +from jvspatial.db.factory import create_database + + +class Receipt(Object): + __test__ = False + name: str = "" + value: int = 0 + type_code: str = Field(default="o") + + +class DeferredReceipt(DeferredSaveMixin, Object): + __test__ = False + name: str = "" + value: int = 0 + type_code: str = Field(default="o") + + +@pytest.fixture +def temp_context(): + with tempfile.TemporaryDirectory() as tmpdir: + import uuid + + path = f"{tmpdir}/cif_{uuid.uuid4().hex}" + database = create_database("json", base_path=path) + context = GraphContext(database=database) + set_default_context(context) + yield context + + +@pytest.mark.asyncio +async def test_create_if_absent_inserts_then_returns_existing(temp_context): + fixed_id = "o.Receipt.deadbeefcafebabe" + a, created_a = await Receipt.create_if_absent(id=fixed_id, name="first", value=1) + assert created_a is True + assert a.id == fixed_id + assert a.name == "first" + assert a.value == 1 + + b, created_b = await Receipt.create_if_absent(id=fixed_id, name="second", value=999) + assert created_b is False + assert b.id == fixed_id + assert b.name == "first" + assert b.value == 1 # rehydrated from stored, not proposed + + +@pytest.mark.asyncio +async def test_create_if_absent_does_not_call_save(temp_context): + fixed_id = "o.Receipt.nosave00000001" + with patch.object(Receipt, "save", new_callable=AsyncMock) as mock_save: + entity, created = await Receipt.create_if_absent(id=fixed_id, name="x", value=2) + assert created is True + assert entity.name == "x" + mock_save.assert_not_called() + + +@pytest.mark.asyncio +async def test_create_if_absent_auto_id_always_creates(temp_context): + a, ca = await Receipt.create_if_absent(name="a", value=1) + b, cb = await Receipt.create_if_absent(name="b", value=2) + assert ca is True and cb is True + assert a.id != b.id + + +@pytest.mark.asyncio +async def test_deferred_flush_only_when_created(temp_context, monkeypatch): + monkeypatch.setenv("SERVERLESS_MODE", "false") + monkeypatch.setenv("JVSPATIAL_ENABLE_DEFERRED_SAVES", "true") + from jvspatial.runtime.serverless import reset_serverless_mode_cache + + reset_serverless_mode_cache() + + fixed_id = "o.DeferredReceipt.aabbccddeeff0011" + flushes: list[str] = [] + + original_flush = DeferredReceipt.flush + + async def tracking_flush(self: Any) -> None: + flushes.append(self.id) + await original_flush(self) + + with patch.object(DeferredReceipt, "flush", tracking_flush): + e1, c1 = await DeferredReceipt.create_if_absent( + id=fixed_id, name="one", value=1 + ) + assert c1 is True + assert flushes == [fixed_id] + + e2, c2 = await DeferredReceipt.create_if_absent( + id=fixed_id, name="two", value=2 + ) + assert c2 is False + assert e2.value == 1 + assert flushes == [fixed_id] # no second flush + + +@pytest.mark.asyncio +async def test_create_if_absent_attaches_graph_context(temp_context): + entity, _ = await Receipt.create_if_absent(name="ctx", value=0) + assert entity._graph_context is temp_context diff --git a/tests/db/test_insert_if_absent.py b/tests/db/test_insert_if_absent.py new file mode 100644 index 0000000..9acc900 --- /dev/null +++ b/tests/db/test_insert_if_absent.py @@ -0,0 +1,387 @@ +"""Tests for Database.insert_if_absent across adapters and wrappers. + +Contract (v1): conflict on primary key ``id`` only; never update/replace an +existing row; return the stored winner plus ``created``. +""" + +from __future__ import annotations + +import asyncio +import tempfile +from pathlib import Path +from typing import Any, Dict +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from jvspatial.db.database import Database, InsertIfAbsentResult +from jvspatial.db.jsondb import JsonDB + +try: + from jvspatial.db.sqlite import SQLiteDB + + HAS_SQLITE = True +except ImportError: # pragma: no cover + SQLiteDB = None # type: ignore[misc] + HAS_SQLITE = False + + +# --------------------------------------------------------------------------- +# ABC / validation +# --------------------------------------------------------------------------- + + +class _StubDB(Database): + """Minimal concrete Database for ABC default-path tests.""" + + async def save(self, collection: str, data: Dict[str, Any]) -> Dict[str, Any]: + return data + + async def get(self, collection: str, id: str): + return None + + async def delete(self, collection: str, id: str) -> None: + return None + + async def find(self, collection: str, query: Dict[str, Any], **kwargs): + return [] + + +@pytest.mark.asyncio +async def test_abc_default_raises_not_implemented(): + db = _StubDB() + with pytest.raises(NotImplementedError, match="insert_if_absent"): + await db.insert_if_absent("object", {"id": "x", "v": 1}) + + +@pytest.mark.asyncio +async def test_abc_rejects_non_id_conflict_target(): + db = _StubDB() + with pytest.raises(ValueError, match="conflict_target"): + await db.insert_if_absent("object", {"id": "x"}, conflict_target="email") + + +@pytest.mark.asyncio +async def test_abc_rejects_missing_id(): + db = _StubDB() + with pytest.raises(ValueError, match="id"): + await db.insert_if_absent("object", {"v": 1}) + + +@pytest.mark.asyncio +async def test_abc_rejects_empty_id(): + db = _StubDB() + with pytest.raises(ValueError, match="id"): + await db.insert_if_absent("object", {"id": ""}) + + +def test_insert_if_absent_result_frozen(): + r = InsertIfAbsentResult(record={"id": "a"}, created=True) + assert r.created is True + assert r.record["id"] == "a" + with pytest.raises(Exception): + r.created = False # type: ignore[misc] + + +# --------------------------------------------------------------------------- +# SQLite +# --------------------------------------------------------------------------- + + +@pytest.fixture +def temp_db_path(): + with tempfile.TemporaryDirectory() as temp_dir: + yield Path(temp_dir) / "iia.db" + + +@pytest.fixture +async def sqlite_db(temp_db_path): + if not HAS_SQLITE: + pytest.skip("aiosqlite required") + from jvspatial.db import create_database + + db = create_database("sqlite", db_path=str(temp_db_path)) + try: + yield db + finally: + if hasattr(db, "close"): + await db.close() + + +@pytest.mark.asyncio +@pytest.mark.skipif(not HAS_SQLITE, reason="aiosqlite required") +async def test_sqlite_insert_absent_then_conflict_preserves_winner(sqlite_db): + first = {"id": "rec.1", "entity": "X", "context": {"v": 1}} + r1 = await sqlite_db.insert_if_absent("object", first) + assert isinstance(r1, InsertIfAbsentResult) + assert r1.created is True + assert r1.record["context"]["v"] == 1 + + loser = {"id": "rec.1", "entity": "X", "context": {"v": 999}} + r2 = await sqlite_db.insert_if_absent("object", loser) + assert r2.created is False + assert r2.record["context"]["v"] == 1 # winner unchanged + + loaded = await sqlite_db.get("object", "rec.1") + assert loaded is not None + assert loaded["context"]["v"] == 1 + + +@pytest.mark.asyncio +@pytest.mark.skipif(not HAS_SQLITE, reason="aiosqlite required") +async def test_sqlite_save_still_upserts(sqlite_db): + """Regression: save() remains INSERT OR REPLACE upsert.""" + await sqlite_db.save("object", {"id": "u.1", "context": {"v": 1}}) + await sqlite_db.save("object", {"id": "u.1", "context": {"v": 2}}) + loaded = await sqlite_db.get("object", "u.1") + assert loaded is not None + assert loaded["context"]["v"] == 2 + + +@pytest.mark.asyncio +@pytest.mark.skipif(not HAS_SQLITE, reason="aiosqlite required") +async def test_sqlite_concurrent_same_id_one_created(sqlite_db): + rec_id = "conc.same" + payloads = [{"id": rec_id, "entity": "X", "context": {"n": i}} for i in range(16)] + + results = await asyncio.gather( + *[sqlite_db.insert_if_absent("object", p) for p in payloads] + ) + created_flags = [r.created for r in results] + assert sum(1 for c in created_flags if c) == 1 + winner = next(r.record for r in results if r.created) + for r in results: + assert r.record == winner + loaded = await sqlite_db.get("object", rec_id) + assert loaded == winner + + +@pytest.mark.asyncio +@pytest.mark.skipif(not HAS_SQLITE, reason="aiosqlite required") +async def test_sqlite_secondary_unique_conflict_does_not_delete_peer(sqlite_db): + """INSERT OR IGNORE must not wipe a peer row the way OR REPLACE does.""" + conn = await sqlite_db._get_connection() + await conn.execute( + """ + CREATE UNIQUE INDEX IF NOT EXISTS idx_object_ctx_k + ON records (collection, json_extract(data, '$.context.k')) + """ + ) + await conn.commit() + + peer = {"id": "peer.1", "entity": "A", "context": {"k": "shared"}} + await sqlite_db.save("object", peer) + + challenger = {"id": "chal.1", "entity": "B", "context": {"k": "shared"}} + # May raise or return created=False; peer must survive either way. + try: + await sqlite_db.insert_if_absent("object", challenger) + except Exception: + pass + + still = await sqlite_db.get("object", "peer.1") + assert still is not None + assert still["context"]["k"] == "shared" + # Challenger must not have replaced peer via OR REPLACE. + assert still["id"] == "peer.1" + + +@pytest.mark.asyncio +@pytest.mark.skipif(not HAS_SQLITE, reason="aiosqlite required") +async def test_sqlite_rejects_bad_conflict_target(sqlite_db): + with pytest.raises(ValueError, match="conflict_target"): + await sqlite_db.insert_if_absent("object", {"id": "x"}, conflict_target="other") + + +# --------------------------------------------------------------------------- +# JsonDB +# --------------------------------------------------------------------------- + + +@pytest.fixture +async def jsondb(): + with tempfile.TemporaryDirectory() as tmp: + db = JsonDB(base_path=tmp) + yield db + + +@pytest.mark.asyncio +async def test_jsondb_insert_if_absent_roundtrip(jsondb): + r1 = await jsondb.insert_if_absent("object", {"id": "j.1", "context": {"v": 1}}) + assert r1.created is True + r2 = await jsondb.insert_if_absent("object", {"id": "j.1", "context": {"v": 99}}) + assert r2.created is False + assert r2.record["context"]["v"] == 1 + + +@pytest.mark.asyncio +async def test_jsondb_concurrent_same_id(jsondb): + results = await asyncio.gather( + *[ + jsondb.insert_if_absent("object", {"id": "j.conc", "context": {"n": i}}) + for i in range(12) + ] + ) + assert sum(1 for r in results if r.created) == 1 + winner = next(r.record for r in results if r.created) + for r in results: + assert r.record == winner + + +# --------------------------------------------------------------------------- +# Wrappers +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_observable_forwards_insert_if_absent(): + from jvspatial.db._observable import ObservableDatabase + from jvspatial.observability import db_op_counter + + inner = MagicMock() + inner.supports_transactions = False + expected = InsertIfAbsentResult(record={"id": "w.1", "v": 1}, created=True) + inner.insert_if_absent = AsyncMock(return_value=expected) + inner.save = AsyncMock(side_effect=AssertionError("save must not be used")) + + token = db_op_counter.set(0) + try: + wrapped = ObservableDatabase(inner) + result = await wrapped.insert_if_absent("object", {"id": "w.1", "v": 1}) + assert db_op_counter.get() == 1 + finally: + db_op_counter.reset(token) + + assert result is expected + inner.insert_if_absent.assert_awaited_once_with( + "object", {"id": "w.1", "v": 1}, conflict_target="id" + ) + + +@pytest.mark.asyncio +async def test_caching_forwards_insert_if_absent_and_caches(monkeypatch): + from jvspatial.db._cache import CachingDatabase + + monkeypatch.setenv("SERVERLESS_MODE", "false") + inner = MagicMock() + inner.supports_transactions = False + record = {"id": "c.1", "v": 1} + inner.insert_if_absent = AsyncMock( + return_value=InsertIfAbsentResult(record=record, created=True) + ) + inner.get = AsyncMock(side_effect=AssertionError("should be cache hit")) + inner.save = AsyncMock(side_effect=AssertionError("save must not be used")) + + cached = CachingDatabase(inner) + result = await cached.insert_if_absent("object", {"id": "c.1", "v": 1}) + assert result.created is True + assert await cached.get("object", "c.1") == record + inner.get.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_caching_insert_if_absent_existing_still_caches(monkeypatch): + from jvspatial.db._cache import CachingDatabase + + monkeypatch.setenv("SERVERLESS_MODE", "false") + inner = MagicMock() + inner.supports_transactions = False + stored = {"id": "c.2", "v": 7} + inner.insert_if_absent = AsyncMock( + return_value=InsertIfAbsentResult(record=stored, created=False) + ) + inner.get = AsyncMock(side_effect=AssertionError("should be cache hit")) + + cached = CachingDatabase(inner) + result = await cached.insert_if_absent("object", {"id": "c.2", "v": 999}) + assert result.created is False + assert await cached.get("object", "c.2") == stored + + +# --------------------------------------------------------------------------- +# MongoDB (unit, mocked) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_mongodb_insert_one_then_duplicate_finds_existing(): + pytest.importorskip("motor") + from pymongo.errors import DuplicateKeyError + + from jvspatial.db.mongodb import MongoDB + + db = MongoDB.__new__(MongoDB) + db._db = MagicMock() + coll = MagicMock() + db._db.__getitem__ = MagicMock(return_value=coll) + db._ensure_connected = AsyncMock() + + async def _run(_name, factory): + return await factory() + + db._run_with_reconnect = _run + + coll.insert_one = AsyncMock(return_value=None) + data = {"_id": "m.1", "id": "m.1", "v": 1} + r1 = await MongoDB.insert_if_absent(db, "object", dict(data)) + assert r1.created is True + coll.insert_one.assert_awaited() + + coll.insert_one = AsyncMock(side_effect=DuplicateKeyError("dup")) + coll.find_one = AsyncMock(return_value={"_id": "m.1", "id": "m.1", "v": 1}) + r2 = await MongoDB.insert_if_absent( + db, "object", {"_id": "m.1", "id": "m.1", "v": 99} + ) + assert r2.created is False + assert r2.record["v"] == 1 + coll.find_one.assert_awaited() + + +# --------------------------------------------------------------------------- +# DynamoDB (unit, mocked) — only if module importable +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_dynamodb_put_conditional_then_get_on_conflict(): + try: + from botocore.exceptions import ClientError + + from jvspatial.db.dynamodb import DynamoDB + except ImportError: + pytest.skip("dynamodb extras not installed") + + db = DynamoDB.__new__(DynamoDB) + db._ensure_table_exists = AsyncMock(return_value="tbl") + db._extract_indexed_fields = MagicMock(return_value={}) + client = MagicMock() + db._get_client = AsyncMock(return_value=client) + + async def _run(_name, factory): + return await factory() + + db._run_with_throttle_retry = _run + + client.put_item = AsyncMock(return_value={}) + r1 = await DynamoDB.insert_if_absent(db, "object", {"id": "d.1", "v": 1}) + assert r1.created is True + put_kwargs = client.put_item.await_args.kwargs + assert "attribute_not_exists" in put_kwargs.get("ConditionExpression", "") + + err = ClientError( + {"Error": {"Code": "ConditionalCheckFailedException", "Message": "x"}}, + "PutItem", + ) + client.put_item = AsyncMock(side_effect=err) + client.get_item = AsyncMock( + return_value={ + "Item": { + "collection": {"S": "object"}, + "id": {"S": "d.1"}, + "data": {"S": '{"id":"d.1","v":1}'}, + } + } + ) + r2 = await DynamoDB.insert_if_absent(db, "object", {"id": "d.1", "v": 99}) + assert r2.created is False + assert r2.record["v"] == 1 diff --git a/tests/db/test_postgres_integration.py b/tests/db/test_postgres_integration.py index 1ea2dd9..629a876 100644 --- a/tests/db/test_postgres_integration.py +++ b/tests/db/test_postgres_integration.py @@ -127,6 +127,44 @@ async def test_save_is_upsert(self, pg_db: "PostgresDB") -> None: loaded = await pg_db.get("node", "n.x.upsert") assert loaded["context"]["v"] == 2 + async def test_insert_if_absent_creates_then_preserves( + self, pg_db: "PostgresDB" + ) -> None: + from jvspatial.db.database import InsertIfAbsentResult + + first = {"id": "n.x.iia", "entity": "x", "context": {"v": 1}} + r1 = await pg_db.insert_if_absent("node", first) + assert isinstance(r1, InsertIfAbsentResult) + assert r1.created is True + assert r1.record["context"]["v"] == 1 + + r2 = await pg_db.insert_if_absent( + "node", {"id": "n.x.iia", "entity": "x", "context": {"v": 99}} + ) + assert r2.created is False + assert r2.record["context"]["v"] == 1 + loaded = await pg_db.get("node", "n.x.iia") + assert loaded is not None + assert loaded["context"]["v"] == 1 + + async def test_insert_if_absent_concurrent_same_id( + self, pg_db: "PostgresDB" + ) -> None: + rec_id = "n.x.iia.conc" + results = await asyncio.gather( + *[ + pg_db.insert_if_absent( + "node", + {"id": rec_id, "entity": "x", "context": {"n": i}}, + ) + for i in range(12) + ] + ) + assert sum(1 for r in results if r.created) == 1 + winner = next(r.record for r in results if r.created) + for r in results: + assert r.record == winner + async def test_delete_removes(self, pg_db: "PostgresDB") -> None: await pg_db.save("node", {"id": "n.x.del", "entity": "x", "context": {}}) await pg_db.delete("node", "n.x.del")