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,22 @@ 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,
+    )
+```
+
 ### Stores
 
 The library provides multiple store implementations organized into three

docs/api/protocols.md

@@ -11,3 +11,27 @@ 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,
+    )
+```
+
+::: 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,2 @@
 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]]]
@@ -407,6 +408,47 @@ 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."""
+        collection = collection or self.default_collection
+        await self.setup_collection(collection=collection)
+
+        created_at, _, expires_at = prepare_entry_timestamps(ttl=ttl)
+        managed_entry = ManagedEntry(
+            value=value,
+            created_at=created_at,
+            expires_at=expires_at,
+        )
+        return await self._put_managed_entry_if_absent(
+            collection=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,6 +1,7 @@
 import sys
 from dataclasses import dataclass
 from datetime import datetime
+from threading import RLock
 from typing import Any
 
 from typing_extensions import override
@@ -13,6 +14,7 @@
     BaseDestroyStore,
     BaseEnumerateCollectionsStore,
     BaseEnumerateKeysStore,
+    BasePutIfAbsentStore,
 )
 
 try:
@@ -64,30 +66,45 @@ def __init__(self, max_entries: int | None = None):
         )
 
         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 = self.get(key)
+            if existing is not None and not existing.is_expired:
+                return False
+            self.put(key, value)
+            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 +190,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)

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

@@ -1,3 +1,4 @@
+import math
 from collections.abc import Sequence
 from datetime import datetime
 from typing import Any, Literal, overload
@@ -10,7 +11,13 @@
 from key_value.aio._utils.managed_entry import ManagedEntry
 from key_value.aio._utils.serialization import BasicSerializationAdapter, SerializationAdapter
 from key_value.aio.errors import DeserializationError
-from key_value.aio.stores.base import BaseContextManagerStore, BaseDestroyStore, BaseEnumerateKeysStore, BaseStore
+from key_value.aio.stores.base import (
+    BaseContextManagerStore,
+    BaseDestroyStore,
+    BaseEnumerateKeysStore,
+    BasePutIfAbsentStore,
+    BaseStore,
+)
 
 try:
     from redis.asyncio import Redis
@@ -153,14 +160,25 @@ async def _redis_mget(client: Redis, keys: list[str]) -> list[Any]:
     return await client.mget(keys=keys)
 
 
-async def _redis_set(client: Redis, name: str, value: str) -> None:
-    """Set a value in Redis without TTL."""
-    _ = await client.set(name=name, value=value)
+def _ttl_to_milliseconds(ttl: float | None) -> int | None:
+    """Preserve TTL precision while keeping Redis expiry positive."""
+    return max(math.ceil(ttl * 1000), 1) if ttl is not None else None
 
 
-async def _redis_setex(client: Redis, name: str, time: int, value: str) -> None:
-    """Set a value in Redis with TTL."""
-    _ = await client.setex(name=name, time=time, value=value)
+async def _redis_set(client: Redis, name: str, value: str, ttl: float | None = None) -> None:
+    """Set a value in Redis with an optional TTL."""
+    _ = await client.set(name=name, value=value, px=_ttl_to_milliseconds(ttl))
+
+
+async def _redis_set_if_absent(
+    client: Redis,
+    name: str,
+    value: str,
+    ttl: float | None,
+) -> bool:
+    """Set a value atomically when its key does not exist."""
+    result = await client.set(name=name, value=value, nx=True, px=_ttl_to_milliseconds(ttl))
+    return bool(result)
 
 
 async def _redis_pipeline_execute(pipeline: Any) -> None:
@@ -183,7 +201,7 @@ async def _redis_flushdb(client: Redis) -> bool:
     return await client.flushdb()  # pyright: ignore[reportUnknownMemberType]
 
 
-class RedisStore(BaseDestroyStore, BaseEnumerateKeysStore, BaseContextManagerStore, BaseStore):
+class RedisStore(BasePutIfAbsentStore, BaseDestroyStore, BaseEnumerateKeysStore, BaseContextManagerStore, BaseStore):
     """Redis-based key-value store."""
 
     _client: Redis
@@ -351,13 +369,28 @@ async def _put_managed_entry(
 
         json_value: str = self._adapter.dump_json(entry=managed_entry, key=key, collection=collection)
 
-        if managed_entry.ttl is not None:
-            # Redis does not support <= 0 TTLs
-            ttl = max(int(managed_entry.ttl), 1)
+        await _redis_set(self._client, combo_key, json_value, managed_entry.ttl)
 
-            await _redis_setex(self._client, combo_key, ttl, json_value)
-        else:
-            await _redis_set(self._client, combo_key, json_value)
+    @override
+    async def _put_managed_entry_if_absent(
+        self,
+        *,
+        key: str,
+        collection: str,
+        managed_entry: ManagedEntry,
+    ) -> bool:
+        combo_key = compound_key(collection=collection, key=key)
+        json_value = self._adapter.dump_json(
+            entry=managed_entry,
+            key=key,
+            collection=collection,
+        )
+        return await _redis_set_if_absent(
+            self._client,
+            combo_key,
+            json_value,
+            managed_entry.ttl,
+        )
 
     @override
     async def _put_managed_entries(
@@ -384,8 +417,7 @@ async def _put_managed_entries(
 
             return
 
-        # Convert TTL to integer seconds for Redis
-        ttl_seconds: int = max(int(ttl), 1)
+        ttl_ms = _ttl_to_milliseconds(ttl)
 
         # Use pipeline for bulk operations
         pipeline = self._client.pipeline()
@@ -394,7 +426,7 @@ async def _put_managed_entries(
             combo_key: str = compound_key(collection=collection, key=key)
             json_value = self._adapter.dump_json(entry=managed_entry, key=key, collection=collection)
 
-            pipeline.setex(name=combo_key, time=ttl_seconds, value=json_value)
+            pipeline.set(name=combo_key, value=json_value, px=ttl_ms)
 
         await _redis_pipeline_execute(pipeline)