feat(backends): add set_if_absent and delete_if_equals for locks and slots

CHANGELOG.md

@@ -15,6 +15,16 @@ Note that 0.3.3 was never released; 0.3.4 follows 0.3.2.
 
 ## [Unreleased]
 
+### Added
+
+- `BaseCacheBackend.set_if_absent(key, value, ttl=None) -> bool` and
+  `delete_if_equals(key, expected) -> bool`, the atomic pair for locks and
+  per-user slots: claim a key only when it is free, and release it only while
+  it still holds your entry, so a holder whose entry expired cannot free a slot
+  someone else has claimed since. Redis uses `SET NX EX` and a Lua
+  compare-and-delete, Memcached `ADD` and `GETS` + `CAS`, memory its lock.
+  Third-party backends inherit non-atomic fallbacks. ([#62](https://github.com/allen0099/FastAPI-CacheX/issues/62))
+
 ## [0.3.5] - 2026-09-15
 
 ### Security

CLAUDE.md

@@ -84,11 +84,13 @@ All backends implement `BaseCacheBackend` (abstract base in `backends/base.py`):
 
 Backend keys are namespaced automatically (default prefix: `fastapi_cachex:`).
 
-Two non-abstract atomic primitives live on the base class with non-atomic fallbacks, and every built-in backend overrides them (see README "Atomic backend primitives"):
+Four non-abstract atomic primitives live on the base class with non-atomic fallbacks, and every built-in backend overrides them (see README "Atomic backend primitives"):
 - `increment(key, delta=1, ttl=None) -> int`: fixed-window counter; `ttl` applies only when the counter is created. Redis runs a registered Lua script, Memcached uses `ADD` + `INCR`/`DECR`, memory works under its lock. A counter reads back through `get()` as a `CacheEntry` with `COUNTER_FINGERPRINT` (`types.py`).
 - `get_and_delete(key) -> CacheEntry | None`: one-shot retrieval (Redis `GETDEL`, Memcached get + `delete(noreply=False)` winner check). `StateManager.consume_state`, `delete_state`, `CacheManager.delete` and `invalidate()` use it. `delete()` keeps returning `None` for 0.3.x compatibility.
+- `set_if_absent(key, value, ttl=None) -> bool`: claim-if-free for locks/slots. Redis `SET NX EX`, Memcached `ADD`, memory under its lock.
+- `delete_if_equals(key, expected) -> bool`: release only while the key still holds `expected` (compared as decoded `CacheEntry`). Redis compares in Python then deletes via a Lua script that re-checks the raw bytes; Memcached uses `GETS` + `CAS` with exptime `-1` (immediate expiry), since classic `DELETE` has no CAS.
 
-`delete_many(keys) -> int` is the third non-abstract base method: a per-key loop by default, one batched operation on Redis (`DEL`) and Memory (single lock).
+`delete_many(keys) -> int` is the fifth non-abstract base method: a per-key loop by default, one batched operation on Redis (`DEL`) and Memory (single lock).
 
 `backends/codec.py` holds the JSON `CacheEntry` codec shared by Redis and Memcached; `decode_entry` maps a bare integer to a counter entry and every malformed value to `None`.
 

README.md

@@ -352,11 +352,14 @@ stored or replayed.
 
 ### Atomic backend primitives
 
-Every backend exposes two atomic operations on top of `get`/`set`/`delete`, for
+Every backend exposes atomic operations on top of `get`/`set`/`delete`, for
 values that are read and written by many concurrent requests:
 
 ```python
+import secrets
+
 from fastapi_cachex import BackendProxy
+from fastapi_cachex.types import CacheEntry
 
 backend = BackendProxy.get()
 
@@ -367,6 +370,14 @@ if hits > 3:
 
 # One-shot value: of several concurrent callers exactly one gets the entry.
 grant = await backend.get_and_delete(f"grant:{token}")
+
+# Lock / slot: claim only if free, release only while it is still yours.
+owner = CacheEntry(fingerprint="lock", content=secrets.token_bytes(16))
+if await backend.set_if_absent(f"stream:{user_id}", owner, ttl=300):
+    try:
+        ...
+    finally:
+        await backend.delete_if_equals(f"stream:{user_id}", owner)
 ```
 
 - `increment(key, delta=1, ttl=None) -> int` — Memory does the read-modify-write
@@ -380,8 +391,18 @@ grant = await backend.get_and_delete(f"grant:{token}")
   uses `GETDEL` (server 6.2+) and Memcached returns the value only when its own
   `DELETE` won. `StateManager.consume_state`, `CacheManager.delete` and
   `invalidate()` are built on it.
-
-Both have a non-atomic fallback on `BaseCacheBackend`, so a third-party backend
+- `set_if_absent(key, value, ttl=None) -> bool` — stores `value` only when
+  `key` does not exist (an expired key counts as absent) and reports whether it
+  did. Memory checks under its lock, Redis uses `SET NX EX` and Memcached `ADD`.
+- `delete_if_equals(key, expected) -> bool` — removes `key` only while it still
+  holds `expected`, so a holder whose entry expired cannot release a lock that
+  someone else has claimed since. Put a unique token in the entry you store and
+  release with that same entry. Memory compares under its lock, Redis deletes
+  through a Lua script that re-checks the value it compared, and Memcached uses
+  `GETS` + a `CAS` write that expires the entry immediately (the classic
+  protocol's `DELETE` takes no CAS token).
+
+All four have a non-atomic fallback on `BaseCacheBackend`, so a third-party backend
 that only implements the abstract methods keeps working; override them to get
 real atomicity.
 

fastapi_cachex/backends/base.py

@@ -83,6 +83,58 @@ async def get_and_delete(self, key: str) -> CacheEntry | None:
             await self.delete(key)
         return value
 
+    async def set_if_absent(
+        self, key: str, value: CacheEntry, ttl: int | None = None
+    ) -> bool:
+        """Store ``value`` only when ``key`` does not exist yet.
+
+        The building block for locks and slots: of several concurrent callers
+        exactly one stores its value and gets ``True``, every other caller
+        gets ``False`` and the stored value is left untouched. An expired key
+        counts as absent. Pair it with ``delete_if_equals`` to release only
+        what you still hold.
+
+        The base implementation is a best-effort, NON-atomic get-then-set
+        fallback for third-party backends; the built-in backends override it
+        with an atomic implementation.
+
+        Args:
+            key: Cache key to claim
+            value: Entry to store, typically carrying a unique owner token
+            ttl: Time to live in seconds (``None`` = never expires)
+
+        Returns:
+            Whether ``value`` was stored
+        """
+        if await self.get(key) is not None:
+            return False
+        await self.set(key, value, ttl=ttl)
+        return True
+
+    async def delete_if_equals(self, key: str, expected: CacheEntry) -> bool:
+        """Remove ``key`` only while it still holds ``expected``.
+
+        Releasing a lock with a plain ``delete`` is unsafe: if the holder's
+        entry expired and someone else claimed the key in the meantime, the
+        delete removes the new holder's entry. Comparing against the value the
+        caller stored makes the release a no-op in that case.
+
+        The base implementation is a best-effort, NON-atomic get-compare-delete
+        fallback for third-party backends; the built-in backends override it
+        with an atomic implementation.
+
+        Args:
+            key: Cache key to release
+            expected: The entry the caller stored (compared with ``==``)
+
+        Returns:
+            Whether the entry was removed
+        """
+        if await self.get(key) != expected:
+            return False
+        await self.delete(key)
+        return True
+
     async def increment(self, key: str, delta: int = 1, ttl: int | None = None) -> int:
         """Atomically add ``delta`` to the integer counter stored at ``key``.
 

fastapi_cachex/backends/memcached.py

@@ -167,6 +167,51 @@ async def get_and_delete(self, key: str) -> CacheEntry | None:
         logger.debug("Memcached GET_AND_DELETE HIT; key=%s", key)
         return decode_entry(raw)
 
+    async def set_if_absent(
+        self, key: str, value: CacheEntry, ttl: int | None = None
+    ) -> bool:
+        """Atomically store ``value`` unless ``key`` exists (see base class).
+
+        Memcached's ``ADD`` is exactly this operation.
+        """
+        stored = await asyncio.to_thread(
+            self.client.add,
+            self._make_key(key),
+            encode_entry(value),
+            _expiry(ttl),
+            noreply=False,
+        )
+        logger.debug(
+            "Memcached SET_IF_ABSENT %s; key=%s ttl=%s",
+            "STORED" if stored else "EXISTS",
+            key,
+            ttl,
+        )
+        return bool(stored)
+
+    async def delete_if_equals(self, key: str, expected: CacheEntry) -> bool:
+        """Atomically remove ``key`` while it holds ``expected`` (see base class).
+
+        The classic protocol's DELETE takes no CAS token, so the release is a
+        CAS write with a negative exptime, which Memcached treats as "expired
+        immediately": it succeeds only if nothing wrote the key since ``GETS``
+        read the value that was compared.
+        """
+        prefixed_key = self._make_key(key)
+        raw, cas_token = await asyncio.to_thread(self.client.gets, prefixed_key)
+        if raw is None or decode_entry(raw) != expected:
+            logger.debug("Memcached DELETE_IF_EQUALS MISMATCH; key=%s", key)
+            return False
+        deleted = await asyncio.to_thread(
+            self.client.cas, prefixed_key, b"", cas_token, -1, noreply=False
+        )
+        logger.debug(
+            "Memcached DELETE_IF_EQUALS %s; key=%s",
+            "HIT" if deleted else "LOST RACE",
+            key,
+        )
+        return bool(deleted)
+
     def _add_delta(self, prefixed_key: str, delta: int) -> int | None:
         """Apply ``delta`` with INCR/DECR; ``None`` when the key does not exist."""
         if delta < 0:

fastapi_cachex/backends/memory.py

@@ -157,6 +157,37 @@ async def get_and_delete(self, key: str) -> CacheEntry | None:
             logger.debug("Memory cache GET_AND_DELETE HIT; key=%s", key)
             return item.value
 
+    async def set_if_absent(
+        self, key: str, value: CacheEntry, ttl: int | None = None
+    ) -> bool:
+        """Atomically store ``value`` unless ``key`` exists (see base class)."""
+        self._ensure_cleanup_started()
+
+        async with self.lock:
+            now = time.time()
+            item = self.cache.get(key)
+            if item is not None and _is_live(item, now):
+                logger.debug("Memory cache SET_IF_ABSENT EXISTS; key=%s", key)
+                return False
+            expiry = now + ttl if ttl is not None else None
+            self.cache[key] = CacheItem(value=value, expiry=expiry)
+            logger.debug("Memory cache SET_IF_ABSENT STORED; key=%s ttl=%s", key, ttl)
+            return True
+
+    async def delete_if_equals(self, key: str, expected: CacheEntry) -> bool:
+        """Atomically remove ``key`` while it holds ``expected`` (see base class)."""
+        async with self.lock:
+            item = self.cache.get(key)
+            if item is None or not _is_live(item, time.time()):
+                logger.debug("Memory cache DELETE_IF_EQUALS MISS; key=%s", key)
+                return False
+            if item.value != expected:
+                logger.debug("Memory cache DELETE_IF_EQUALS MISMATCH; key=%s", key)
+                return False
+            del self.cache[key]
+            logger.debug("Memory cache DELETE_IF_EQUALS HIT; key=%s", key)
+            return True
+
     async def increment(self, key: str, delta: int = 1, ttl: int | None = None) -> int:
         """Atomically add ``delta`` to the counter at ``key`` (see base class).
 

fastapi_cachex/backends/redis.py

@@ -36,6 +36,15 @@
 return value
 """
 
+# DEL that only fires while the key still holds the exact bytes the caller read,
+# so a value replaced in the meantime survives. KEYS[1] = key, ARGV[1] = bytes.
+_DELETE_IF_EQUALS_SCRIPT = """
+if redis.call('GET', KEYS[1]) == ARGV[1] then
+    return redis.call('DEL', KEYS[1])
+end
+return 0
+"""
+
 
 class AsyncRedisCacheBackend(BaseCacheBackend):
     """Async Redis cache backend implementation.
@@ -108,6 +117,9 @@ def __init__(
         # Registered once so every call is an EVALSHA (redis-py reloads the
         # script transparently if the server has flushed it).
         self._increment_script = self.client.register_script(_INCREMENT_SCRIPT)
+        self._delete_if_equals_script = self.client.register_script(
+            _DELETE_IF_EQUALS_SCRIPT
+        )
 
     @staticmethod
     def load_from_config(config: RedisConfig) -> "AsyncRedisCacheBackend":
@@ -189,6 +201,42 @@ async def get_and_delete(self, key: str) -> CacheEntry | None:
         logger.debug("Redis GETDEL %s; key=%s", "HIT" if value else "MISS", key)
         return value
 
+    async def set_if_absent(
+        self, key: str, value: CacheEntry, ttl: int | None = None
+    ) -> bool:
+        """Atomically store ``value`` unless ``key`` exists (see base class).
+
+        A single ``SET ... NX EX``.
+        """
+        stored = await self.client.set(
+            self._make_key(key), encode_entry(value), ex=ttl, nx=True
+        )
+        logger.debug(
+            "Redis SET_IF_ABSENT %s; key=%s ttl=%s",
+            "STORED" if stored else "EXISTS",
+            key,
+            ttl,
+        )
+        return bool(stored)
+
+    async def delete_if_equals(self, key: str, expected: CacheEntry) -> bool:
+        """Atomically remove ``key`` while it holds ``expected`` (see base class).
+
+        The stored value is decoded and compared here, then a Lua script
+        deletes the key only if it still holds the bytes that were compared,
+        so a value written in between is never removed.
+        """
+        prefixed_key = self._make_key(key)
+        raw = await self.client.get(prefixed_key)
+        if raw is None or decode_entry(raw) != expected:
+            logger.debug("Redis DELETE_IF_EQUALS MISMATCH; key=%s", key)
+            return False
+        deleted = await self._delete_if_equals_script(keys=[prefixed_key], args=[raw])
+        logger.debug(
+            "Redis DELETE_IF_EQUALS %s; key=%s", "HIT" if deleted else "LOST RACE", key
+        )
+        return bool(deleted)
+
     async def increment(self, key: str, delta: int = 1, ttl: int | None = None) -> int:
         """Atomically add ``delta`` to the counter at ``key`` (see base class).
 

tests/backends/test_base.py

@@ -86,3 +86,30 @@ async def test_delete_many_fallback_deletes_one_by_one(backend: DictBackend) ->
 
     assert await backend.delete_many(["a", "b", "missing"]) == 3
     assert backend.store == {}
+
+
+@pytest.mark.asyncio
+async def test_set_if_absent_fallback_stores_only_the_first_value(
+    backend: DictBackend,
+) -> None:
+    first = CacheEntry(fingerprint="lock", content=b"owner-a")
+    second = CacheEntry(fingerprint="lock", content=b"owner-b")
+
+    assert await backend.set_if_absent("slot", first, ttl=30) is True
+    assert await backend.set_if_absent("slot", second, ttl=30) is False
+    assert backend.store["slot"] == (first, 30)
+
+
+@pytest.mark.asyncio
+async def test_delete_if_equals_fallback_removes_only_a_matching_entry(
+    backend: DictBackend,
+) -> None:
+    mine = CacheEntry(fingerprint="lock", content=b"owner-a")
+    theirs = CacheEntry(fingerprint="lock", content=b"owner-b")
+    await backend.set("slot", theirs)
+
+    assert await backend.delete_if_equals("slot", mine) is False
+    assert "slot" in backend.store
+    assert await backend.delete_if_equals("slot", theirs) is True
+    assert "slot" not in backend.store
+    assert await backend.delete_if_equals("slot", theirs) is False