From 53a836664a12dbbb1180c87c9132f919a798b76a Mon Sep 17 00:00:00 2001 From: Alex Shalaev Date: Sun, 6 Sep 2026 21:14:22 +0300 Subject: [PATCH 1/9] fix(decorator): read a positionally passed key and warn when no coordinator is found The decorator only ever looked in **kwargs for the key, so a key passed positionally was invisible and the operation ran unprotected. It now resolves the key from the positional arguments too, when the parameter can be passed that way. Failing to find a coordinator still degrades to running the function, but it is now logged as a warning instead of happening in silence. --- .../core/decorators/aio/idempotent.py | 106 ++++++++++++------ tests/unit/core/test_idempotent.py | 73 ++++++++++++ 2 files changed, 142 insertions(+), 37 deletions(-) diff --git a/idempotency_kit/core/decorators/aio/idempotent.py b/idempotency_kit/core/decorators/aio/idempotent.py index 661c409..1446712 100644 --- a/idempotency_kit/core/decorators/aio/idempotent.py +++ b/idempotency_kit/core/decorators/aio/idempotent.py @@ -1,4 +1,6 @@ import functools +import inspect +import logging from collections.abc import Awaitable, Callable from typing import Any, TypeVar @@ -7,6 +9,22 @@ T = TypeVar("T") +logger = logging.getLogger(__name__) + +_POSITIONAL_KINDS = (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD) + + +def _positional_index(func: Callable[..., Any], key_param: str) -> int | None: + """Index at which ``key_param`` can arrive positionally, or ``None`` if it cannot.""" + try: + parameters = list(inspect.signature(func).parameters.values()) + except (TypeError, ValueError): + return None + for index, parameter in enumerate(parameters): + if parameter.name == key_param and parameter.kind in _POSITIONAL_KINDS: + return index + return None + def async_idempotent( operation: str, @@ -25,56 +43,38 @@ def async_idempotent( adapter: Result adapter for encoding/decoding. ttl_seconds: Optional TTL for idempotency record in seconds. If not provided, uses value from coordinator settings or global default. - key_param: Name of the argument containing the idempotency key. + key_param: Name of the argument containing the idempotency key. Read from the + keyword arguments, or from the positional arguments when the parameter can + be passed positionally. infra_param: Optional name of the argument or attribute containing AsyncIdempotencyCoordinator. If not provided, searches for AsyncIdempotencyCoordinator by type. """ def decorator(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]: + key_index = _positional_index(func, key_param) + @functools.wraps(func) async def wrapper(*args: Any, **kwargs: Any) -> T: # 1. Resolve idempotency key - idempotency_key = kwargs.get(key_param) + idempotency_key = _resolve_key(args, kwargs) if not idempotency_key: return await func(*args, **kwargs) # 2. Resolve coordinator - coordinator: AsyncIdempotencyCoordinator | None = None + coordinator = _resolve_coordinator(args, kwargs) - # 2.1. Try to find by name if infra_param is provided - if infra_param: - if infra_param in kwargs: - coordinator = kwargs[infra_param] - elif args and hasattr(args[0], infra_param): - coordinator = getattr(args[0], infra_param) - - # 2.2. Try to find by type if not found or infra_param is None - if not coordinator: - # Search in kwargs - for val in kwargs.values(): - if isinstance(val, AsyncIdempotencyCoordinator): - coordinator = val - break - - # Search in args (skipping self if it was already checked) - if not coordinator: - for arg in args: - if isinstance(arg, AsyncIdempotencyCoordinator): - coordinator = arg - break - # Also check self attributes if arg is 'self' - # We do this because DI often injects into attributes - if hasattr(arg, "__dict__"): - for attr_val in vars(arg).values(): - if isinstance(attr_val, AsyncIdempotencyCoordinator): - coordinator = attr_val - break - if coordinator: - break - - if not coordinator: - # If no coordinator found but key is present, we might want to fail or proceed - # Proceeding without idempotency is safer but should probably be logged + if coordinator is None: + # Proceeding without idempotency keeps the operation available, but it is + # never what the decorator was put there for: say so loudly enough to be + # caught by whoever renamed the attribute or forgot the argument. + logger.warning( + "No idempotency coordinator found; running the operation without idempotency", + extra={ + "operation": operation, + "idempotency_key": idempotency_key, + "infra_param": infra_param, + }, + ) return await func(*args, **kwargs) # 3. Delegate to coordinator @@ -88,6 +88,38 @@ async def wrapper(*args: Any, **kwargs: Any) -> T: **kwargs, ) + def _resolve_key(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any: + if key_param in kwargs: + return kwargs[key_param] + if key_index is not None and key_index < len(args): + return args[key_index] + return None + + def _resolve_coordinator(args: tuple[Any, ...], kwargs: dict[str, Any]) -> AsyncIdempotencyCoordinator | None: + # By name, if infra_param is provided + if infra_param: + named: AsyncIdempotencyCoordinator | None = kwargs.get(infra_param) + if named is None and args: + named = getattr(args[0], infra_param, None) + if named is not None: + return named + + # By type, in the keyword arguments + for val in kwargs.values(): + if isinstance(val, AsyncIdempotencyCoordinator): + return val + + # By type, in the positional arguments, then in their attributes: + # DI often injects the coordinator into an attribute of 'self'. + for arg in args: + if isinstance(arg, AsyncIdempotencyCoordinator): + return arg + if hasattr(arg, "__dict__"): + for attr_val in vars(arg).values(): + if isinstance(attr_val, AsyncIdempotencyCoordinator): + return attr_val + return None + return wrapper return decorator diff --git a/tests/unit/core/test_idempotent.py b/tests/unit/core/test_idempotent.py index db65a77..8a5ebc2 100644 --- a/tests/unit/core/test_idempotent.py +++ b/tests/unit/core/test_idempotent.py @@ -1,5 +1,6 @@ """Unit tests for async idempotent decorator.""" +import logging from unittest.mock import ANY, MagicMock import pytest @@ -123,3 +124,75 @@ async def my_func(idempotency_key: str | None) -> str: # Assert assert result == "direct" + + +@pytest.mark.asyncio +async def test__decorator__positional_key__calls_coordinator( + mock_coordinator: MagicMock, mock_adapter: MagicMock +) -> None: + """A key the caller passes positionally must still reach the coordinator.""" + # Arrange + operation = "test.op" + key = "test-key" + mock_coordinator.coordinate.return_value = "ok" + + @async_idempotent(operation=operation, adapter=mock_adapter) + async def my_func(idempotency_key: str | None, coord: AsyncIdempotencyCoordinator) -> str: + return "not used" + + # Act + result = await my_func(key, mock_coordinator) + + # Assert + assert result == "ok" + mock_coordinator.coordinate.assert_called_once_with( + operation, + key, + None, + mock_adapter, + ANY, # the original function + key, + mock_coordinator, + ) + + +@pytest.mark.asyncio +async def test__decorator__keyword_only_key__is_not_read_from_positional_arguments( + mock_coordinator: MagicMock, mock_adapter: MagicMock +) -> None: + """A keyword-only parameter cannot arrive positionally, so nothing else may be read as the key.""" + # Arrange + mock_coordinator.coordinate.return_value = "ok" + + @async_idempotent(operation="test.op", adapter=mock_adapter) + async def my_func(payload: str, *, idempotency_key: str | None = None) -> str: + return "direct" + + # Act + result = await my_func("payload") + + # Assert + assert result == "direct" + mock_coordinator.coordinate.assert_not_called() + + +@pytest.mark.asyncio +async def test__decorator__no_coordinator__warns(mock_adapter: MagicMock, caplog: pytest.LogCaptureFixture) -> None: + """Running unprotected is a fallback, not a silent one.""" + # Arrange + operation = "test.op" + + @async_idempotent(operation=operation, adapter=mock_adapter) + async def my_func(*, idempotency_key: str | None = None) -> str: + return "direct" + + # Act + with caplog.at_level(logging.WARNING, logger="idempotency_kit.core.decorators.aio.idempotent"): + result = await my_func(idempotency_key="test-key") + + # Assert + assert result == "direct" + assert [record.message for record in caplog.records] == [ + "No idempotency coordinator found; running the operation without idempotency" + ] + assert caplog.records[0].operation == operation # type: ignore[attr-defined] From 773dded1837714f9106ae6cde8800ad4b7ca0aa2 Mon Sep 17 00:00:00 2001 From: Alex Shalaev Date: Sun, 6 Sep 2026 21:15:45 +0300 Subject: [PATCH 2/9] fix(adapters): stop PydanticResultAdapter storing a record it can never decode encode() returned null for a falsy value and decode() rejected a falsy payload, so an action returning None stored a record that failed to decode for the life of that record: every replay re-ran the action, collided on save, and failed to decode the winner. A model that dumps to an empty mapping or list hit the same decode guard. encode() now refuses None outright -- reported as record_validation_error and swallowed by the coordinator, so the operation goes uncached rather than poison-cached -- and decode() only rejects a null payload. --- idempotency_kit/core/adapters/basic.py | 26 ++++++++++-- tests/unit/core/test_adapter_round_trip.py | 49 +++++++++++++++++++++- 2 files changed, 70 insertions(+), 5 deletions(-) diff --git a/idempotency_kit/core/adapters/basic.py b/idempotency_kit/core/adapters/basic.py index 75253d5..594b033 100644 --- a/idempotency_kit/core/adapters/basic.py +++ b/idempotency_kit/core/adapters/basic.py @@ -2,23 +2,41 @@ from pydantic import BaseModel +from idempotency_kit.core.exceptions import IdempotencyValidationError from idempotency_kit.core.protocols.adapter import ResultAdapter T = TypeVar("T", bound=BaseModel) class PydanticResultAdapter(ResultAdapter[T]): - """Adapter for Pydantic models.""" + """Adapter for Pydantic models. + + It has no representation for an absent result: an action that may return ``None`` + wants ``VoidResultAdapter`` or ``JsonResultAdapter`` instead. + """ def __init__(self, model_class: type[T]) -> None: self.model_class = model_class def encode(self, value: T) -> Any: - return value.model_dump(mode="json") if value else None + # Encoding None to null would store a record this adapter can never decode again, + # turning every later replay into a miss. Refusing it leaves the operation + # uncached and says why, which the coordinator reports and swallows. + model: BaseModel | None = value + if model is None: + msg = ( + f"{type(self).__name__}({self.model_class.__name__}) cannot encode None. " + "Use VoidResultAdapter for an action that returns None, or JsonResultAdapter " + "for one that may." + ) + raise IdempotencyValidationError(msg) + return model.model_dump(mode="json") def decode(self, data: Any) -> T: - if not data: - msg = "cannot decode empty idempotency payload" + # Only a null payload is undecodable: an empty mapping or list is a model that + # happens to dump to nothing. + if data is None: + msg = "cannot decode a null idempotency payload" raise ValueError(msg) return self.model_class.model_validate(data) diff --git a/tests/unit/core/test_adapter_round_trip.py b/tests/unit/core/test_adapter_round_trip.py index 50c1044..5124d6f 100644 --- a/tests/unit/core/test_adapter_round_trip.py +++ b/tests/unit/core/test_adapter_round_trip.py @@ -6,6 +6,7 @@ """ from typing import Any +from unittest.mock import MagicMock import pytest from fakeredis import FakeAsyncRedis as AsyncRedisClient @@ -14,6 +15,7 @@ from idempotency_kit import ( AsyncIdempotencyCoordinator, IdempotencyDomainService, + IdempotencyMetricsProtocol, JsonResultAdapter, PydanticResultAdapter, ResultAdapter, @@ -27,6 +29,10 @@ class _Order(BaseModel): status: str +class _Ack(BaseModel): + """A model that dumps to an empty mapping -- an acknowledgement carrying no fields.""" + + @pytest.mark.asyncio @pytest.mark.parametrize( ("adapter", "value"), @@ -37,8 +43,17 @@ class _Order(BaseModel): (JsonResultAdapter(), None), (JsonResultAdapter(), "plain string"), (PydanticResultAdapter(_Order), _Order(id=1, status="paid")), + (PydanticResultAdapter(_Ack), _Ack()), + ], + ids=[ + "void", + "json-mapping", + "json-list", + "json-null", + "json-string", + "pydantic-model", + "pydantic-empty-model", ], - ids=["void", "json-mapping", "json-list", "json-null", "json-string", "pydantic-model"], ) async def test__coordinator__shipped_adapter_round_trip__executes_once_and_replays_the_stored_result( fake_redis: AsyncRedisClient, adapter: ResultAdapter[Any], value: Any @@ -62,3 +77,35 @@ async def action() -> Any: assert first == value assert second == value assert await fake_redis.get("probe:op.round-trip:key-1") is not None + + +@pytest.mark.asyncio +async def test__coordinator__pydantic_adapter_on_a_none_result__stores_nothing_and_reports_it( + fake_redis: AsyncRedisClient, +) -> None: + """A None the adapter cannot represent must not become a record that never decodes again.""" + # Arrange + metrics = MagicMock(spec=IdempotencyMetricsProtocol) + repository = RedisAsyncIdempotencyRepository(fake_redis, key_prefix="probe:") + coordinator = AsyncIdempotencyCoordinator( + repository=repository, domain_service=IdempotencyDomainService(), metrics=metrics + ) + calls = 0 + + async def action() -> Any: + nonlocal calls + calls += 1 + return None + + # Act + first = await coordinator.coordinate("op.absent", "key-1", 600, PydanticResultAdapter(_Order), action) + second = await coordinator.coordinate("op.absent", "key-1", 600, PydanticResultAdapter(_Order), action) + + # Assert + assert (first, second) == (None, None) + assert calls == 2 + assert await fake_redis.get("probe:op.absent:key-1") is None + assert metrics.record_error.call_args_list == [ + (("op.absent", "record_validation_error"),), + (("op.absent", "record_validation_error"),), + ] From d24d775eb6823f6902e30a1c0a1f5f8afbf16c51 Mon Sep 17 00:00:00 2001 From: Alex Shalaev Date: Sun, 6 Sep 2026 21:16:47 +0300 Subject: [PATCH 3/9] fix(redis): name the dependency and the extra that actually exist The ImportError told the reader to install redis-client-kit and offered a [redis-aio] extra; the declared extra is [redis] and the requirement is redis plus orjson. The architecture page named the same missing extra, and the API reference described the repository's redis argument as a redis-client-kit client when it is any redis.asyncio.Redis. --- docs/api_reference.md | 2 +- docs/architecture.md | 2 +- .../infra/storage/redis/aio/repository.py | 4 ++-- .../unit/storage/redis/aio/test_repository.py | 20 +++++++++++++++++++ 4 files changed, 24 insertions(+), 4 deletions(-) diff --git a/docs/api_reference.md b/docs/api_reference.md index 323958e..ffdea05 100644 --- a/docs/api_reference.md +++ b/docs/api_reference.md @@ -66,7 +66,7 @@ Interface for metrics collection. Redis implementation of the repository protocol. - **Constructor**: - - `redis` (AsyncRedisClient): Instance of `redis-client-kit` client. + - `redis` (`redis.asyncio.Redis`): Any async Redis client, including a subclass such as an instrumented or fake one. - `key_prefix` (str, default: "idempotency:"): Prefix for all Redis keys. (**keyword-only**) - `metrics` (IdempotencyMetricsProtocol, optional): Metrics collector. (**keyword-only**) diff --git a/docs/architecture.md b/docs/architecture.md index 22050e4..5bc441d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -11,7 +11,7 @@ This is the innermost layer. It contains: - **Domain Services**: Business logic for creating and validating idempotency records (`IdempotencyDomainService`). - **Exceptions**: Domain-specific error classes. -The core domain has **minimal dependencies** (Pydantic for models and validation). Infrastructure layer adds Redis integration via optional `[redis-aio]` extra and uses `orjson` for fast serialization. +The core domain has **minimal dependencies** (Pydantic for models and validation). Infrastructure layer adds Redis integration via optional `[redis]` extra and uses `orjson` for fast serialization. ### 2. Infrastructure Layer (`idempotency_kit.infra`) This layer contains concrete implementations of the protocols defined in the Core layer. diff --git a/idempotency_kit/infra/storage/redis/aio/repository.py b/idempotency_kit/infra/storage/redis/aio/repository.py index 51d4cf9..82c0d7f 100644 --- a/idempotency_kit/infra/storage/redis/aio/repository.py +++ b/idempotency_kit/infra/storage/redis/aio/repository.py @@ -61,8 +61,8 @@ def __init__( """ if not _HAS_REDIS or not _HAS_ORJSON: raise ImportError( - "RedisAsyncIdempotencyRepository requires redis-client-kit and orjson. " - "Install them with: pip install idempotency-kit[redis-aio]" + "RedisAsyncIdempotencyRepository requires redis and orjson. " + "Install them with: pip install idempotency-kit[redis]" ) self._redis = redis self._key_prefix = key_prefix diff --git a/tests/unit/storage/redis/aio/test_repository.py b/tests/unit/storage/redis/aio/test_repository.py index 1ec8d53..7aed75d 100644 --- a/tests/unit/storage/redis/aio/test_repository.py +++ b/tests/unit/storage/redis/aio/test_repository.py @@ -1,6 +1,8 @@ """Unit tests for Redis repository.""" +import re from datetime import UTC, datetime, timedelta +from importlib.metadata import metadata from unittest.mock import AsyncMock, MagicMock, patch import orjson @@ -283,6 +285,24 @@ async def test_repository_import_error() -> None: RedisAsyncIdempotencyRepository(MagicMock()) +@pytest.mark.asyncio +async def test_repository_import_error_names_the_declared_extra() -> None: + """The install hint has to name an extra this distribution actually declares.""" + import idempotency_kit.infra.storage.redis.aio.repository as repo_module # noqa: PLC0415 + + with ( + patch.object(repo_module, "_HAS_ORJSON", False), + pytest.raises(ImportError) as exc_info, + ): + RedisAsyncIdempotencyRepository(MagicMock()) + + message = str(exc_info.value) + named_extra = re.search(r"idempotency-kit\[([^\]]+)\]", message) + assert named_extra is not None + assert named_extra.group(1) in (metadata("idempotency-kit").get_all("Provides-Extra") or []) + assert "redis-client-kit" not in message + + @pytest.mark.asyncio async def test_storage_error_original_error_all_methods(fake_redis: AsyncRedisClient) -> None: """Test that IdempotencyStorageError contains the original exception for all methods.""" From 8e53c360833209cf1ad624e5bab5349dac4444c1 Mon Sep 17 00:00:00 2001 From: Alex Shalaev Date: Sun, 6 Sep 2026 21:18:27 +0300 Subject: [PATCH 4/9] fix(metrics): count a coordinator-driven get and save once The repository and the coordinator both recorded hit, miss, collision and latency for the same call, and the shipped providers give them one APP-scoped collector, so every coordinator-driven get produced two misses and two method="get" observations. The coordinator now owns those four -- its hit and miss are the truthful ones, since a record that fails to decode is a miss for the caller. The repository keeps what the coordinator cannot produce: its error counters, the bulk hit and miss counts of get_many, and the latency of delete and get_many. --- .../infra/storage/redis/aio/repository.py | 160 +++++++++--------- .../unit/storage/redis/aio/test_repository.py | 47 +++-- 2 files changed, 114 insertions(+), 93 deletions(-) diff --git a/idempotency_kit/infra/storage/redis/aio/repository.py b/idempotency_kit/infra/storage/redis/aio/repository.py index 82c0d7f..85ece48 100644 --- a/idempotency_kit/infra/storage/redis/aio/repository.py +++ b/idempotency_kit/infra/storage/redis/aio/repository.py @@ -43,6 +43,12 @@ class RedisAsyncIdempotencyRepository(AsyncIdempotencyRepository): """Redis implementation of idempotency repository. Stores records as JSON with automatic TTL expiration. + + It records only the metrics the coordinator cannot produce for it -- errors, the bulk + hit and miss counts of ``get_many``, and the latency of ``delete`` and ``get_many``. + Hit, miss, collision and the latency of ``get`` and ``save`` belong to + ``AsyncIdempotencyCoordinator``, so that a collector shared by both counts each + operation once. """ def __init__( @@ -134,49 +140,42 @@ async def get(self, operation: str, idempotency_key: str) -> IdempotencyRecord | IdempotencyStorageError: If Redis operation fails IdempotencyError: If data is corrupted """ - start = time.perf_counter() + self._validate_inputs(operation, idempotency_key) + key = self._make_key(operation, idempotency_key) try: - self._validate_inputs(operation, idempotency_key) - key = self._make_key(operation, idempotency_key) - try: - data = await self._redis.get(key) - except Exception as e: - self._metrics.record_error(operation, type(e).__name__) - logger.exception( - "Redis error during get", - extra={"operation": operation, "key": idempotency_key}, - ) - raise IdempotencyStorageError( - f"Redis storage failure during get for {operation}", - operation=operation, - original_error=e, - ) from e + data = await self._redis.get(key) + except Exception as e: + self._metrics.record_error(operation, type(e).__name__) + logger.exception( + "Redis error during get", + extra={"operation": operation, "key": idempotency_key}, + ) + raise IdempotencyStorageError( + f"Redis storage failure during get for {operation}", + operation=operation, + original_error=e, + ) from e - if not data: - self._metrics.record_miss(operation) - return None + if not data: + return None - record = self._deserialize_record(data, operation, idempotency_key) - if record is None: - self._metrics.record_miss(operation) + record = self._deserialize_record(data, operation, idempotency_key) + if record is None: + logger.warning( + "Found expired record in Redis (TTL mismatch). Deleting it.", + extra={"operation": operation, "key": idempotency_key}, + ) + try: + await self._redis.delete(key) + except Exception: logger.warning( - "Found expired record in Redis (TTL mismatch). Deleting it.", + "Failed to delete expired record from Redis", extra={"operation": operation, "key": idempotency_key}, + exc_info=True, ) - try: - await self._redis.delete(key) - except Exception: - logger.warning( - "Failed to delete expired record from Redis", - extra={"operation": operation, "key": idempotency_key}, - exc_info=True, - ) - return None + return None - self._metrics.record_hit(operation) - return record - finally: - self._metrics.record_latency(operation, "get", time.perf_counter() - start) + return record async def save( self, @@ -193,60 +192,55 @@ async def save( IdempotencyStorageError: If Redis operation fails IdempotencyError: If serialization fails """ - start = time.perf_counter() operation = record.operation - try: - self._validate_inputs(operation, record.idempotency_key) - key = self._make_key(operation, record.idempotency_key) + self._validate_inputs(operation, record.idempotency_key) + key = self._make_key(operation, record.idempotency_key) - # Calculate TTL - ttl_seconds = math.ceil(record.ttl_seconds) + # Calculate TTL + ttl_seconds = math.ceil(record.ttl_seconds) - if ttl_seconds <= 0: - self._metrics.record_error(operation, "validation_error") - logger.warning( - "Attempted to save already expired record", - extra={"operation": operation, "key": record.idempotency_key}, - ) - raise IdempotencyValidationError("Cannot save already expired record") + if ttl_seconds <= 0: + self._metrics.record_error(operation, "validation_error") + logger.warning( + "Attempted to save already expired record", + extra={"operation": operation, "key": record.idempotency_key}, + ) + raise IdempotencyValidationError("Cannot save already expired record") - # Serialize record - try: - data = orjson.dumps(record.model_dump(mode="json")) if _HAS_ORJSON else record.model_dump_json() - except Exception as e: - self._metrics.record_error(operation, "serialization_error") - logger.exception( - "Failed to serialize record", - extra={"operation": operation, "key": record.idempotency_key}, - ) - raise IdempotencyError("Serialization failed") from e + # Serialize record + try: + data = orjson.dumps(record.model_dump(mode="json")) if _HAS_ORJSON else record.model_dump_json() + except Exception as e: + self._metrics.record_error(operation, "serialization_error") + logger.exception( + "Failed to serialize record", + extra={"operation": operation, "key": record.idempotency_key}, + ) + raise IdempotencyError("Serialization failed") from e - # Use SET with NX (only if key doesn't exist) and EX (expiration) - try: - was_set = await self._redis.set(key, data, ex=ttl_seconds, nx=True) - except Exception as e: - self._metrics.record_error(operation, type(e).__name__) - logger.exception( - "Redis error during save", - extra={"operation": operation, "key": record.idempotency_key}, - ) - # In case of Redis error, we cannot guarantee idempotency. - raise IdempotencyStorageError( - "Redis storage failure during save", - operation=operation, - original_error=e, - ) from e + # Use SET with NX (only if key doesn't exist) and EX (expiration) + try: + was_set = await self._redis.set(key, data, ex=ttl_seconds, nx=True) + except Exception as e: + self._metrics.record_error(operation, type(e).__name__) + logger.exception( + "Redis error during save", + extra={"operation": operation, "key": record.idempotency_key}, + ) + # In case of Redis error, we cannot guarantee idempotency. + raise IdempotencyStorageError( + "Redis storage failure during save", + operation=operation, + original_error=e, + ) from e - if not was_set: - self._metrics.record_collision(operation) - raise IdempotencyKeyCollisionError(operation, record.idempotency_key) + if not was_set: + raise IdempotencyKeyCollisionError(operation, record.idempotency_key) - logger.debug( - "Saved idempotency record", - extra={"operation": operation, "key": record.idempotency_key, "ttl_seconds": ttl_seconds}, - ) - finally: - self._metrics.record_latency(operation, "save", time.perf_counter() - start) + logger.debug( + "Saved idempotency record", + extra={"operation": operation, "key": record.idempotency_key, "ttl_seconds": ttl_seconds}, + ) async def delete(self, operation: str, idempotency_key: str) -> bool: """Delete record from Redis. diff --git a/tests/unit/storage/redis/aio/test_repository.py b/tests/unit/storage/redis/aio/test_repository.py index 7aed75d..e58ca7d 100644 --- a/tests/unit/storage/redis/aio/test_repository.py +++ b/tests/unit/storage/redis/aio/test_repository.py @@ -10,12 +10,14 @@ from fakeredis import FakeAsyncRedis as AsyncRedisClient from idempotency_kit import ( + AsyncIdempotencyCoordinator, IdempotencyDomainService, IdempotencyError, IdempotencyKeyCollisionError, IdempotencyRecord, IdempotencyStorageError, IdempotencyValidationError, + JsonResultAdapter, ) from idempotency_kit.core.constants import MAX_KEY_LENGTH, MAX_OPERATION_LENGTH from idempotency_kit.core.protocols.metrics import IdempotencyMetricsProtocol @@ -440,22 +442,18 @@ async def test_metrics_comprehensive(fake_redis: AsyncRedisClient) -> None: repo = RedisAsyncIdempotencyRepository(fake_redis, metrics=metrics_mock) service = IdempotencyDomainService() - # record_hit on get + # Hit, miss, collision and the latency of get and save belong to the coordinator, + # so that a collector shared by both layers counts each operation once. record = service.create_record("op", "hit", {"r": 1}) await repo.save(record) - metrics_mock.record_latency.reset_mock() await repo.get("op", "hit") - metrics_mock.record_hit.assert_called_with("op") - metrics_mock.record_latency.assert_called() - - # record_miss on get await repo.get("op", "miss") - metrics_mock.record_miss.assert_called_with("op") - - # record_collision on save with pytest.raises(IdempotencyKeyCollisionError): await repo.save(record) - metrics_mock.record_collision.assert_called_with("op") + metrics_mock.record_hit.assert_not_called() + metrics_mock.record_miss.assert_not_called() + metrics_mock.record_collision.assert_not_called() + metrics_mock.record_latency.assert_not_called() # record_error on serialization failure with ( @@ -483,6 +481,35 @@ async def test_metrics_comprehensive(fake_redis: AsyncRedisClient) -> None: metrics_mock.record_bulk_hit.assert_called_with("bulk", 1) metrics_mock.record_bulk_miss.assert_called_with("bulk", 2) + # delete and get_many have no coordinator equivalent, so the repository times them + await repo.delete("op", "hit") + assert [call.args[1] for call in metrics_mock.record_latency.call_args_list] == ["get_many", "delete"] + + +@pytest.mark.asyncio +async def test_metrics_shared_with_coordinator_count_each_operation_once(fake_redis: AsyncRedisClient) -> None: + """One collector wired to both layers -- what the shipped providers do -- must not count twice.""" + metrics_mock = MagicMock(spec=IdempotencyMetricsProtocol) + repo = RedisAsyncIdempotencyRepository(fake_redis, key_prefix="shared:", metrics=metrics_mock) + coordinator = AsyncIdempotencyCoordinator( + repository=repo, + domain_service=IdempotencyDomainService(), + metrics=metrics_mock, + ) + adapter: JsonResultAdapter = JsonResultAdapter() + + async def action() -> dict[str, int]: + return {"r": 1} + + # Act + await coordinator.coordinate("op.shared", "key", 600, adapter, action) + await coordinator.coordinate("op.shared", "key", 600, adapter, action) + + # Assert + assert metrics_mock.record_miss.call_count == 1 + assert metrics_mock.record_hit.call_count == 1 + assert [call.args[1] for call in metrics_mock.record_latency.call_args_list] == ["get", "save", "get"] + @pytest.mark.asyncio async def test_redis_save_subsecond_ttl(fake_redis: AsyncRedisClient) -> None: From f0ec28436275dbaf2515f3dd83374ec2cbd4015e Mon Sep 17 00:00:00 2001 From: Alex Shalaev Date: Sun, 6 Sep 2026 21:19:13 +0300 Subject: [PATCH 5/9] fix(coordinator): treat an expired record as a miss IdempotencyDomainService.validate_record and IdempotencyRecordExpiredError were public with no caller in the library, and the coordinator trusted the repository to never hand back an expired record. The shipped Redis repository does check, but the rule belongs to the domain and a backend without native expiry cannot enforce it, so a third-party repository replayed stale results forever. The coordinator now validates every record it reads and counts an expired one as a miss. --- .../core/services/aio/coordinator.py | 11 ++++++ tests/unit/core/test_coordinator.py | 36 ++++++++++++++++++- 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/idempotency_kit/core/services/aio/coordinator.py b/idempotency_kit/core/services/aio/coordinator.py index 7ec789d..db41edd 100644 --- a/idempotency_kit/core/services/aio/coordinator.py +++ b/idempotency_kit/core/services/aio/coordinator.py @@ -7,6 +7,7 @@ from idempotency_kit.core.exceptions import ( IdempotencyInvalidTTLError, IdempotencyKeyCollisionError, + IdempotencyRecordExpiredError, IdempotencyValidationError, ) from idempotency_kit.core.protocols.adapter import ResultAdapter @@ -163,6 +164,16 @@ async def _get_and_decode( cached = await self._repo.get(operation, idempotency_key) if cached is None: return None + try: + # The protocol asks a repository not to return an expired record, but expiry is + # the domain's rule to enforce, and a backend without native expiry cannot. + self._svc.validate_record(cached) + except IdempotencyRecordExpiredError: + logger.warning( + "Idempotency record expired; treating it as a miss", + extra={"operation": operation, "idempotency_key": idempotency_key}, + ) + return None return self._decode_safely(adapter, cached.result, operation, idempotency_key) async def _save_to_repo( diff --git a/tests/unit/core/test_coordinator.py b/tests/unit/core/test_coordinator.py index db2d514..12fbfbc 100644 --- a/tests/unit/core/test_coordinator.py +++ b/tests/unit/core/test_coordinator.py @@ -1,6 +1,6 @@ """Unit tests for async idempotency coordinator.""" -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from unittest.mock import AsyncMock, MagicMock import pytest @@ -8,6 +8,7 @@ from idempotency_kit.core.exceptions import IdempotencyKeyCollisionError, IdempotencyValidationError from idempotency_kit.core.models.entities import IdempotencyRecord from idempotency_kit.core.services.aio.coordinator import AsyncIdempotencyCoordinator +from idempotency_kit.core.services.domain import IdempotencyDomainService @pytest.mark.asyncio @@ -297,3 +298,36 @@ async def test__coordinator__record_validation_error__returns_result_and_reports assert result == {"data": "ok"} mock_repo.save.assert_not_called() metrics.record_error.assert_called_once_with("op", "record_validation_error") + + +@pytest.mark.asyncio +async def test__coordinator__expired_record_from_repository__is_a_miss( + mock_repo: AsyncMock, mock_adapter: MagicMock +) -> None: + """Expiry is the domain's rule, so a repository that hands back a stale record must not replay it.""" + # Arrange + now = datetime.now(UTC) + stale = IdempotencyRecord( + operation="op", + idempotency_key="key", + result={"data": "stale"}, + created_at=now - timedelta(hours=2), + expires_at=now - timedelta(hours=1), + ) + metrics = MagicMock() + coordinator = AsyncIdempotencyCoordinator( + repository=mock_repo, + domain_service=IdempotencyDomainService(), + metrics=metrics, + ) + mock_repo.get.return_value = stale + action = AsyncMock(return_value={"data": "fresh"}) + + # Act + result = await coordinator.coordinate("op", "key", 600, mock_adapter, action) + + # Assert + assert result == {"data": "fresh"} + action.assert_called_once() + mock_adapter.decode.assert_not_called() + metrics.record_miss.assert_called_once_with("op") From 401b01bc72b2142681179c5f8ecc67f60dd62e72 Mon Sep 17 00:00:00 2001 From: Alex Shalaev Date: Sun, 6 Sep 2026 21:20:22 +0300 Subject: [PATCH 6/9] feat(dishka): honour settings.enabled BaseIdempotencySettings.enabled was read by nothing: setting it to False changed no behaviour anywhere in the library. AsyncIdempotencyCoordinator now takes enabled (default True), and the shipped coordinator provider passes the settings flag through, so False makes every call a pass-through to the action. The provider reads the flag with getattr so a settings object written against the protocol before enabled was part of it keeps working. --- .../core/services/aio/coordinator.py | 20 +++++++-- idempotency_kit/dishka/aio/coordinator.py | 3 ++ idempotency_kit/dishka/protocols.py | 8 ++++ idempotency_kit/settings.py | 5 ++- tests/unit/core/test_coordinator.py | 26 ++++++++++++ tests/unit/dishka/test_providers.py | 41 ++++++++++++++++++- 6 files changed, 97 insertions(+), 6 deletions(-) diff --git a/idempotency_kit/core/services/aio/coordinator.py b/idempotency_kit/core/services/aio/coordinator.py index db41edd..a394e83 100644 --- a/idempotency_kit/core/services/aio/coordinator.py +++ b/idempotency_kit/core/services/aio/coordinator.py @@ -33,7 +33,15 @@ class _Hit(Generic[T]): class AsyncIdempotencyCoordinator: - """Coordinator for asynchronous idempotent operations.""" + """Coordinator for asynchronous idempotent operations. + + Args: + repository: Storage for idempotency records. + domain_service: Record factory and TTL bounds. + operation_ttls: Per-operation TTL overrides in seconds; wins over the decorator. + metrics: Metrics collector for hits, misses, collisions, errors and latency. + enabled: Set to False to make every call a pass-through to the action. + """ def __init__( self, @@ -41,11 +49,13 @@ def __init__( domain_service: IdempotencyDomainService, operation_ttls: dict[str, int] | None = None, metrics: IdempotencyMetricsProtocol | None = None, + enabled: bool = True, ) -> None: self._repo = repository self._svc = domain_service self._operation_ttls = operation_ttls or {} self._metrics = metrics or NoOpIdempotencyMetrics() + self._enabled = enabled async def coordinate( self, @@ -58,8 +68,12 @@ async def coordinate( *args: Any, **kwargs: Any, ) -> T: - """Coordinate an idempotent operation.""" - if not idempotency_key: + """Coordinate an idempotent operation. + + With ``enabled=False`` the action is simply run: nothing is read, nothing is + written, and no metric is recorded. + """ + if not self._enabled or not idempotency_key: return await action(*args, **kwargs) # 1. Try to get from storage diff --git a/idempotency_kit/dishka/aio/coordinator.py b/idempotency_kit/dishka/aio/coordinator.py index 1b0ff67..25db995 100644 --- a/idempotency_kit/dishka/aio/coordinator.py +++ b/idempotency_kit/dishka/aio/coordinator.py @@ -31,4 +31,7 @@ def get_coordinator( domain_service=domain_service, operation_ttls=settings.operation_ttls, metrics=metrics, + # Read defensively: settings objects written against the protocol before + # ``enabled`` was part of it stay valid, and they mean enabled. + enabled=getattr(settings, "enabled", True), ) diff --git a/idempotency_kit/dishka/protocols.py b/idempotency_kit/dishka/protocols.py index 08c2c04..7b5c850 100644 --- a/idempotency_kit/dishka/protocols.py +++ b/idempotency_kit/dishka/protocols.py @@ -7,6 +7,14 @@ class IdempotencySettingsProtocol(Protocol): """Protocol for idempotency settings.""" + @property + def enabled(self) -> bool: + """Whether the shipped coordinator applies idempotency at all. + + A settings object without the attribute is read as enabled. + """ + ... + @property def key_prefix(self) -> str: """Key prefix for Redis.""" diff --git a/idempotency_kit/settings.py b/idempotency_kit/settings.py index 320c65f..33f4e6a 100644 --- a/idempotency_kit/settings.py +++ b/idempotency_kit/settings.py @@ -6,7 +6,10 @@ class BaseIdempotencySettings(BaseModel): """Common configuration for idempotency kit.""" - enabled: bool = Field(default=True, description="Whether idempotency is enabled") + enabled: bool = Field( + default=True, + description="Whether the coordinator applies idempotency; False makes every call a pass-through", + ) key_prefix: str = Field(description="Redis key prefix for idempotency records") metrics_enabled: bool = Field(default=False, description="Whether idempotency metrics are enabled") default_ttl_minutes: int = Field(default=60, description="Default TTL for records in minutes") diff --git a/tests/unit/core/test_coordinator.py b/tests/unit/core/test_coordinator.py index 12fbfbc..b516dd3 100644 --- a/tests/unit/core/test_coordinator.py +++ b/tests/unit/core/test_coordinator.py @@ -331,3 +331,29 @@ async def test__coordinator__expired_record_from_repository__is_a_miss( action.assert_called_once() mock_adapter.decode.assert_not_called() metrics.record_miss.assert_called_once_with("op") + + +@pytest.mark.asyncio +async def test__coordinator__disabled__runs_the_action_without_touching_storage( + mock_repo: AsyncMock, mock_domain_service: MagicMock, mock_adapter: MagicMock +) -> None: + """enabled=False is a kill switch: no read, no write, no metric.""" + # Arrange + metrics = MagicMock() + coordinator = AsyncIdempotencyCoordinator( + repository=mock_repo, + domain_service=mock_domain_service, + metrics=metrics, + enabled=False, + ) + action = AsyncMock(return_value="fresh") + + # Act + result = await coordinator.coordinate("op", "key", 60, mock_adapter, action) + + # Assert + assert result == "fresh" + action.assert_called_once() + mock_repo.get.assert_not_called() + mock_repo.save.assert_not_called() + metrics.record_miss.assert_not_called() diff --git a/tests/unit/dishka/test_providers.py b/tests/unit/dishka/test_providers.py index 7c86b36..00bf069 100644 --- a/tests/unit/dishka/test_providers.py +++ b/tests/unit/dishka/test_providers.py @@ -10,6 +10,7 @@ AsyncIdempotencyCoordinator, AsyncIdempotencyRepository, IdempotencyMetricsProtocol, + JsonResultAdapter, NoOpIdempotencyMetrics, ) from idempotency_kit.dishka import ( @@ -27,14 +28,19 @@ class _AppProvider(Provider): scope = Scope.APP - def __init__(self, *, metrics_enabled: bool) -> None: + def __init__(self, *, metrics_enabled: bool, enabled: bool = True) -> None: super().__init__() self._metrics_enabled = metrics_enabled + self._enabled = enabled @provide def settings(self) -> IdempotencySettingsProtocol: """Provide idempotency settings.""" - return BaseIdempotencySettings(key_prefix="probe:", metrics_enabled=self._metrics_enabled) + return BaseIdempotencySettings( + key_prefix="probe:", + metrics_enabled=self._metrics_enabled, + enabled=self._enabled, + ) @provide def redis(self) -> AsyncRedisClient: @@ -150,3 +156,34 @@ async def test__idempotency_provider__without_redis_provider__resolves_metrics() assert isinstance(metrics, NoOpIdempotencyMetrics) finally: await container.close() + + +@pytest.mark.asyncio +async def test__shipped_providers__settings_disabled__coordinator_is_a_pass_through() -> None: + """settings.enabled=False has to reach the coordinator the providers build.""" + # Arrange + container = make_async_container( + _AppProvider(metrics_enabled=False, enabled=False), + IdempotencyProvider(), + AsyncRedisIdempotencyProvider(), + AsyncIdempotencyCoordinatorProvider(), + ) + calls = 0 + + async def action() -> str: + nonlocal calls + calls += 1 + return "ran" + + # Act + try: + coordinator = await container.get(AsyncIdempotencyCoordinator) + await coordinator.coordinate("op.disabled", "key", 600, JsonResultAdapter(), action) + await coordinator.coordinate("op.disabled", "key", 600, JsonResultAdapter(), action) + + # Assert + assert calls == 2 + repository = await container.get(AsyncIdempotencyRepository) + assert await repository.get("op.disabled", "key") is None + finally: + await container.close() From 452729a5576f7c44c3ba3ac07603f2f7bc7f989c Mon Sep 17 00:00:00 2001 From: Alex Shalaev Date: Sun, 6 Sep 2026 21:21:39 +0300 Subject: [PATCH 7/9] feat!: one canonical set of TTL defaults IdempotencyDomainService defaulted to 30 minutes, a floor of 60 seconds and a ceiling of 24 hours, while BaseIdempotencySettings shipped 60 minutes, a floor of 1 second and a ceiling of 30 days -- so the effective bounds depended on whether the service was built by hand or from the settings object. core/constants.py is now the single source and the settings model takes its field defaults from it. The wider pair won on both bounds, because narrowing them would have started rejecting TTLs that work today, and the coordinator swallows an out-of-range TTL: the operation would have gone quietly uncached. BREAKING CHANGE: DEFAULT_TTL_MINUTES is 60 (was 30) and MAX_TTL_SECONDS is 2592000 (was 86400), so IdempotencyDomainService() built without arguments now keeps records for an hour and accepts a TTL of up to 30 days. BaseIdempotencySettings.min_ttl_seconds defaults to 60 (was 1), which no shipped path can observe: the coordinator already floors every TTL at one minute. Pass the arguments explicitly to keep the old values. --- idempotency_kit/core/constants.py | 14 +++++++++----- idempotency_kit/settings.py | 10 +++++++--- tests/unit/core/test_services.py | 19 ++++++++++++++++++- 3 files changed, 34 insertions(+), 9 deletions(-) diff --git a/idempotency_kit/core/constants.py b/idempotency_kit/core/constants.py index d62bdf7..7395fc3 100644 --- a/idempotency_kit/core/constants.py +++ b/idempotency_kit/core/constants.py @@ -8,11 +8,15 @@ MAX_OPERATION_LENGTH: int = 100 # TTL defaults -# Default TTL in minutes when not explicitly specified. -DEFAULT_TTL_MINUTES: int = 30 +# The single source for both IdempotencyDomainService's own defaults and the field +# defaults of BaseIdempotencySettings, so the bounds do not depend on how the service +# was built. -# Minimum allowed TTL in seconds (1 minute). +# Default TTL in minutes when not explicitly specified (1 hour). +DEFAULT_TTL_MINUTES: int = 60 + +# Minimum allowed TTL in seconds (1 minute); the coordinator floors every TTL at a minute. MIN_TTL_SECONDS: int = 60 -# Maximum allowed TTL in seconds (24 hours). -MAX_TTL_SECONDS: int = 86400 # 24 hours +# Maximum allowed TTL in seconds (30 days). +MAX_TTL_SECONDS: int = 30 * 24 * 3600 diff --git a/idempotency_kit/settings.py b/idempotency_kit/settings.py index 33f4e6a..bc01d44 100644 --- a/idempotency_kit/settings.py +++ b/idempotency_kit/settings.py @@ -2,6 +2,8 @@ from pydantic import BaseModel, Field +from .core.constants import DEFAULT_TTL_MINUTES, MAX_TTL_SECONDS, MIN_TTL_SECONDS + class BaseIdempotencySettings(BaseModel): """Common configuration for idempotency kit.""" @@ -12,9 +14,11 @@ class BaseIdempotencySettings(BaseModel): ) key_prefix: str = Field(description="Redis key prefix for idempotency records") metrics_enabled: bool = Field(default=False, description="Whether idempotency metrics are enabled") - default_ttl_minutes: int = Field(default=60, description="Default TTL for records in minutes") - min_ttl_seconds: int = Field(default=1, description="Minimum allowed TTL in seconds") - max_ttl_seconds: int = Field(default=30 * 24 * 3600, description="Maximum allowed TTL in seconds (30 days)") + default_ttl_minutes: int = Field( + default=DEFAULT_TTL_MINUTES, description="Default TTL for records in minutes (1 hour)" + ) + min_ttl_seconds: int = Field(default=MIN_TTL_SECONDS, description="Minimum allowed TTL in seconds (1 minute)") + max_ttl_seconds: int = Field(default=MAX_TTL_SECONDS, description="Maximum allowed TTL in seconds (30 days)") operation_ttls: dict[str, int] = Field( default_factory=dict, description="Operation-specific TTLs in seconds (overrides decorator and default)", diff --git a/tests/unit/core/test_services.py b/tests/unit/core/test_services.py index 0c84bfe..7f9e5c6 100644 --- a/tests/unit/core/test_services.py +++ b/tests/unit/core/test_services.py @@ -12,7 +12,9 @@ IdempotencyRecordExpiredError, IdempotencyValidationError, ) +from idempotency_kit.core.constants import DEFAULT_TTL_MINUTES, MAX_TTL_SECONDS, MIN_TTL_SECONDS from idempotency_kit.core.protocols.metrics import NoOpIdempotencyMetrics +from idempotency_kit.settings import BaseIdempotencySettings def test__domain_service__valid_input__creates_record() -> None: @@ -75,7 +77,7 @@ def test__domain_service__empty_field__raises_validation_error(field: str, value ("ttl_minutes", "reason"), [ (0, "below_minimum"), - (2000, "above_maximum"), + (MAX_TTL_SECONDS // 60 + 1, "above_maximum"), ], ids=["ttl_zero", "ttl_too_large"], ) @@ -251,3 +253,18 @@ def test__noop_metrics__all_methods__do_not_raise() -> None: metrics.record_latency("op", "method", 0.1) metrics.record_bulk_hit("op", 5) metrics.record_bulk_miss("op", 5) + + +def test__domain_service_and_settings__agree_on_the_ttl_defaults() -> None: + """One canonical set of TTL defaults, so the bounds do not depend on how the service was built.""" + # Arrange + service = IdempotencyDomainService() + settings = BaseIdempotencySettings(key_prefix="probe:") + + # Act + from_service = (service.default_ttl_minutes, service.min_ttl_seconds, service.max_ttl_seconds) + from_settings = (settings.default_ttl_minutes, settings.min_ttl_seconds, settings.max_ttl_seconds) + + # Assert + assert from_service == from_settings + assert from_settings == (DEFAULT_TTL_MINUTES, MIN_TTL_SECONDS, MAX_TTL_SECONDS) From 2f0bb8f99a1b1d12af7e55b543c39f5c8ce8a0dd Mon Sep 17 00:00:00 2001 From: Alex Shalaev Date: Sun, 6 Sep 2026 21:24:34 +0300 Subject: [PATCH 8/9] docs: bring the pages in line with the fixes agents.md stated four of the defects as rules -- the silent coordinator lookup, the positional key the decorator could not see, the poison null from PydanticResultAdapter, and both layers counting the same get -- so those rules had to go with the code. The TTL defaults, the coordinator signature and the settings field list follow the new canonical numbers, and the reference, guide, quickstart and architecture pages carry the same changes. --- docs/agents.md | 98 +++++++++++++++++++++++++------------------ docs/api_reference.md | 10 +++-- docs/architecture.md | 9 +++- docs/quickstart.md | 12 +++--- docs/user_guide.md | 15 ++++--- 5 files changed, 87 insertions(+), 57 deletions(-) diff --git a/docs/agents.md b/docs/agents.md index 652d3e3..d5ecfcf 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -58,13 +58,14 @@ Four nouns and one flow. `PydanticResultAdapter(model_class)`, `JsonResultAdapter()`, `VoidResultAdapter()`. * **`AsyncIdempotencyCoordinator`** — the flow: read the record, decode it, return it if it is there; otherwise run the action, encode the result, write it under `SET NX`, and on a - collision re-read and return the winner's result instead. It swallows every storage and + collision re-read and return the winner's result instead. A record whose `expires_at` + has passed is a miss even if the repository handed it back. It swallows every storage and decode failure and degrades to running the action, which is the deliberate trade of exactly-once for availability. `IdempotencyDomainService` sits between the coordinator and the record: it applies the TTL bounds and turns Pydantic validation errors into `IdempotencyValidationError`. -`@async_idempotent` is the same flow as a decorator — it finds the key in the call's keyword +`@async_idempotent` is the same flow as a decorator — it finds the key in the call's arguments and the coordinator in the call's arguments or on `self`, then delegates. The storage key is `{key_prefix}{operation}:{idempotency_key}`, which is why neither part @@ -106,9 +107,10 @@ class CreateOrder: return OrderDTO.from_entity(order) ``` -`idempotency_key` is keyword-only on purpose: the decorator reads it out of `**kwargs` and -nowhere else. `infra_param="coordinator"` names the attribute rather than leaving the -decorator to find a coordinator by type. +`idempotency_key` is keyword-only on purpose: nothing else can land in it by position, and +the call site has to name it. The decorator reads it from the keyword arguments, or from the +positional ones when the parameter can be passed that way. `infra_param="coordinator"` names +the attribute rather than leaving the decorator to find a coordinator by type. The same call without the decorator — the five leading arguments are positional-only, and everything after them is forwarded to the action: @@ -131,8 +133,8 @@ result = await coordinator.coordinate( | Name | Signature | What it is | |---|---|---| | `async_idempotent` | `(operation, adapter, ttl_seconds=None, key_param="idempotency_key", infra_param=None)` | decorator for an async function or method | -| `AsyncIdempotencyCoordinator` | `(repository, domain_service, operation_ttls=None, metrics=None)` | the flow; `operation_ttls` is `dict[str, int]` in seconds | -| `IdempotencyDomainService` | `(*, default_ttl_minutes=30, min_ttl_seconds=60, max_ttl_seconds=86400)` | record factory and TTL bounds; keyword-only | +| `AsyncIdempotencyCoordinator` | `(repository, domain_service, operation_ttls=None, metrics=None, enabled=True)` | the flow; `operation_ttls` is `dict[str, int]` in seconds; `enabled=False` runs the action and nothing else | +| `IdempotencyDomainService` | `(*, default_ttl_minutes=60, min_ttl_seconds=60, max_ttl_seconds=2592000)` | record factory and TTL bounds; keyword-only, defaults from `core.constants` | | `IdempotencyRecord` | frozen Pydantic model | the cached result | | `IdempotencyIdentifiers` | Pydantic model | `operation` + `idempotency_key`, and the rules they obey | | `AsyncIdempotencyRepository` | runtime-checkable `Protocol` | storage contract | @@ -160,7 +162,7 @@ result = await coordinator.coordinate( |---|---|---| | `AsyncIdempotencyCoordinator.coordinate(operation, idempotency_key, ttl_seconds, adapter, action, /, *args, **kwargs)` | `T` | never raises for storage or decode trouble | | `IdempotencyDomainService.create_record(operation, idempotency_key, result, *, ttl_minutes=None)` | `IdempotencyRecord` | raises `IdempotencyInvalidTTLError`, `IdempotencyValidationError` | -| `IdempotencyDomainService.validate_record(record)` | `None` | raises `IdempotencyRecordExpiredError`; nothing in the library calls it | +| `IdempotencyDomainService.validate_record(record)` | `None` | raises `IdempotencyRecordExpiredError`; the coordinator calls it on every record it reads | ### Record @@ -199,12 +201,18 @@ and `record_bulk_miss(operation, count)`. `PrometheusIdempotencyMetrics(prefix=N `idempotency_operations_total{operation,status}` and `idempotency_operation_duration_seconds{operation,method}`. +Each metric has one owner, so the same collector can go to both layers: the coordinator +records hit, miss, collision and the latency of `get` and `save`; the repository records +errors, the bulk hit and miss counts of `get_many`, and the latency of `delete` and +`get_many`. + ### Settings and Dishka `BaseIdempotencySettings` is a plain Pydantic model with `enabled=True`, `key_prefix` (required, no default), `metrics_enabled=False`, `default_ttl_minutes=60`, -`min_ttl_seconds=1`, `max_ttl_seconds=2592000` and `operation_ttls={}`. It satisfies -`IdempotencySettingsProtocol`, which is what the providers ask for. +`min_ttl_seconds=60`, `max_ttl_seconds=2592000` and `operation_ttls={}` — the three TTL +fields default to the `core.constants` values, which is also what `IdempotencyDomainService` +uses. It satisfies `IdempotencySettingsProtocol`, which is what the providers ask for. ```python from dishka import make_async_container @@ -227,6 +235,8 @@ container = make_async_container( All three are `Scope.APP`. `IdempotencyProvider` gives one metrics collector to the whole process — `PrometheusIdempotencyMetrics` when `settings.metrics_enabled`, a no-op otherwise. Another backend goes in with `@provide(override=True)` in a provider listed after it. +`settings.enabled` reaches the coordinator: `False` makes it a pass-through. A settings +object written before that field existed is read as enabled. ## Rules that hold or break the code @@ -239,17 +249,19 @@ Another backend goes in with `@provide(override=True)` in a provider listed afte execute, the loser's `save` collides, and it re-reads and returns the winner's record. The callers see one result; the side effects happened twice. Make the action safe to run twice — a database upsert, an outbox row keyed by the same key — or take your own lock. -3. **The decorator reads the key from keyword arguments only.** `kwargs.get(key_param)`. - Passed positionally, the key is invisible, the function runs unprotected, and nothing is - logged. Declare the parameter keyword-only. +3. **Declare the key keyword-only anyway.** The decorator reads `key_param` from the + keyword arguments, and from the positional arguments when the parameter can be passed + that way. Keyword-only is still the shape to write: nothing can land in it by position, + and the call site has to name it. 4. **A falsy key means no idempotency.** `None` and `""` both short-circuit straight to the action, in the decorator and in `coordinate()`. -5. **A coordinator the decorator cannot find means no idempotency, silently.** It looks for +5. **A coordinator the decorator cannot find means no idempotency.** It looks for `infra_param` by name in `kwargs` then as an attribute of the first positional argument, then for an `AsyncIdempotencyCoordinator` by type in `kwargs`, in `args`, and in the - instance dictionary of every positional argument. Finding none, it just calls the - function — no exception, no log line. Pass `infra_param=` so a renamed attribute fails - loudly in review instead of quietly at runtime. + instance dictionary of every positional argument. Finding none, it logs a `WARNING` from + `idempotency_kit.core.decorators.aio.idempotent` and calls the function anyway — the + operation stays available, unprotected. Pass `infra_param=` so a renamed attribute is one + grep away, and alert on that warning. 6. **`coordinate()` never raises for storage trouble.** A Redis failure on read is counted as `storage_get_error` and treated as a miss; a failure on write is `storage_save_error` and the fresh result is returned uncached. Availability over exactly-once, deliberately. @@ -267,19 +279,19 @@ Another backend goes in with `@provide(override=True)` in a provider listed afte 10. **`operation_ttls` wins over the decorator, and a zero there is not a value.** The coordinator resolves `self._operation_ttls.get(operation) or ttl_seconds`, so an entry of `0` falls through to the decorator's number rather than meaning "no TTL". -11. **The domain service's bounds decide what is storable, and the shipped settings widen - them.** `IdempotencyDomainService()` alone is 30 minutes by default, floor 60 seconds, - ceiling 86400. `BaseIdempotencySettings` defaults to 60 minutes, floor 1 second, ceiling - 30 days, and the Dishka provider builds the service from those. Out of range raises +11. **The domain service's bounds decide what is storable.** `IdempotencyDomainService()` + and `BaseIdempotencySettings` agree on them: 60 minutes by default, floor 60 seconds, + ceiling 30 days, all three from `idempotency_kit.core.constants`. Out of range raises `IdempotencyInvalidTTLError`, which the coordinator catches: the operation is simply not cached, and the caller gets its result anyway. 12. **Neither identifier may contain a colon**, both are stripped of surrounding whitespace, and the lengths are 100 for `operation` and 255 for `idempotency_key`. The Redis repository re-validates on every call, so an over-long key raises there too. -13. **`PydanticResultAdapter` cannot represent an absent result.** `encode` returns `None` - for a falsy value and `decode` raises on a falsy payload, so a function that may return - `None` stores `null` and then fails to decode it forever — rule 7, permanently. Use - `VoidResultAdapter` when the action returns `None` and `JsonResultAdapter` when it may. +13. **`PydanticResultAdapter` cannot represent an absent result.** `encode` raises + `IdempotencyValidationError` when handed `None` rather than storing a `null` it could + never decode again; the coordinator counts that as `record_validation_error`, logs it, + and leaves the operation uncached. Use `VoidResultAdapter` when the action returns + `None` and `JsonResultAdapter` when it may. 14. **What is stored has to be a JSON value.** `IdempotencyRecord.result` is Pydantic's `JsonValue` and `JsonResultAdapter` passes the value through untouched, so a `datetime`, a `Decimal` or a `set` fails record validation, is logged as @@ -293,31 +305,35 @@ Another backend goes in with `@provide(override=True)` in a provider listed afte 17. **`PrometheusIdempotencyMetrics` is one instance per process.** It registers its collectors in the constructor; a second instance with the same prefix raises from `prometheus_client`. -18. **Both layers count.** The repository and the coordinator each record hit, miss and - latency, so one collector shared between them — which is what the Dishka providers wire - — counts every coordinator-driven get twice. +18. **Each metric has one owner.** The coordinator records hit, miss, collision and the + latency of `get` and `save`; the repository records errors, the bulk hit and miss counts + of `get_many`, and the latency of `delete` and `get_many`. One collector shared between + them — which is what the Dishka providers wire — counts every operation once. 19. **Async only.** There is no sync mirror, and no `__init__.py` name that gives you one. +20. **`enabled=False` switches the whole thing off.** The coordinator runs the action and + nothing else: no read, no write, no metric. The Dishka providers pass `settings.enabled` + through to it. ## Common mistakes ```python -# WRONG — the key arrives positionally, so the decorator never sees it and the -# function runs unprotected on every call -@async_idempotent(operation="order.create", adapter=PydanticResultAdapter(OrderDTO)) -async def execute(self, dto: CreateOrderDTO, idempotency_key: str | None = None) -> OrderDTO: ... - -await use_case.execute(dto, key) +# WRONG — infra_param names an attribute that no longer exists, so no coordinator is +# found: the call runs unprotected and only a WARNING says so +class CreateOrder: + def __init__(self, idempotency: AsyncIdempotencyCoordinator) -> None: + self._idempotency = idempotency -# RIGHT — keyword-only, so it cannot be passed any other way -@async_idempotent(operation="order.create", adapter=PydanticResultAdapter(OrderDTO)) -async def execute(self, dto: CreateOrderDTO, *, idempotency_key: str | None = None) -> OrderDTO: ... + @async_idempotent(operation="order.create", adapter=..., infra_param="coordinator") + async def execute(self, dto: CreateOrderDTO, *, idempotency_key: str | None = None) -> OrderDTO: ... -await use_case.execute(dto, idempotency_key=key) +# RIGHT — name the attribute that is actually there + @async_idempotent(operation="order.create", adapter=..., infra_param="_idempotency") + async def execute(self, dto: CreateOrderDTO, *, idempotency_key: str | None = None) -> OrderDTO: ... ``` ```python -# WRONG — a Pydantic adapter on an action that may return nothing: the record -# stores null, decode raises, and the action re-runs on every replay +# WRONG — a Pydantic adapter on an action that may return nothing: the adapter refuses +# to encode the None, so the operation is never cached and the action re-runs every time @async_idempotent(operation="user.deactivate", adapter=PydanticResultAdapter(UserDTO)) async def deactivate(self, *, idempotency_key: str | None = None) -> UserDTO | None: ... @@ -377,7 +393,7 @@ All derive from `IdempotencyError`, which is exported alongside them. |---|---|---| | `IdempotencyError` | `(message)` | the base, and what a corrupted stored payload raises | | `IdempotencyKeyCollisionError` | `(operation, key)` | `save` found the key already there; `key` is a `str`, or a `list[str]` from `save_many`. Carries `.operation` and `.key` | -| `IdempotencyRecordExpiredError` | `(operation, key)` | `validate_record` was given an expired record. Nothing in the library calls it — it is for your own code | +| `IdempotencyRecordExpiredError` | `(operation, key)` | `validate_record` was given an expired record. The coordinator raises it internally and turns it into a miss; through the repository or your own call to `validate_record` you meet it directly | | `IdempotencyStorageError` | `(message, operation=None, original_error=None)` | the backend failed. Carries `.operation` and `.original_error` | | `IdempotencyValidationError` | `(message, errors=None)` | an identifier or a result failed validation. `.errors` holds the Pydantic error list when there is one | | `IdempotencyInvalidTTLError` | `(ttl_seconds, min_ttl, max_ttl)` | the TTL is outside the domain service's range. Carries all three | diff --git a/docs/api_reference.md b/docs/api_reference.md index ffdea05..12129a8 100644 --- a/docs/api_reference.md +++ b/docs/api_reference.md @@ -29,13 +29,14 @@ A frozen Pydantic model representing an idempotency result. Inherits from `Idemp Service for creating and validating records. - **Constructor** (Parameters are **keyword-only**): - - `default_ttl_minutes` (int, default: 30): Default TTL in minutes. Must be >= 1. + - `default_ttl_minutes` (int, default: 60): Default TTL in minutes. Must be >= 1. - `min_ttl_seconds` (int, default: 60): Minimum allowed TTL. Must be >= 1. - - `max_ttl_seconds` (int, default: 86400): Maximum allowed TTL (24 hours). Must be >= `min_ttl_seconds`. + - `max_ttl_seconds` (int, default: 2592000): Maximum allowed TTL (30 days). Must be >= `min_ttl_seconds`. + - *Note*: These three defaults live in `idempotency_kit.core.constants` and are also the field defaults of `BaseIdempotencySettings`. - *Note*: Constructor validates that `default_ttl_minutes` (converted to seconds) is within the `[min_ttl_seconds, max_ttl_seconds]` range. - **Methods**: - `create_record(operation, idempotency_key, result, *, ttl_minutes=None)`: Creates a new `IdempotencyRecord` with validation and TTL management. - - `validate_record(record)`: Validates that a record is still usable (not expired). + - `validate_record(record)`: Validates that a record is still usable, raising `IdempotencyRecordExpiredError` if not. The coordinator calls it on every record it reads. ### AsyncIdempotencyRepository (Protocol) Interface for idempotency storage. @@ -59,6 +60,9 @@ Interface for metrics collection. - `record_latency(operation, method, duration_seconds)` - `record_bulk_hit(operation, count)`: For bulk operations. - `record_bulk_miss(operation, count)`: For bulk operations. +- **Who records what**: the coordinator records hit, miss, collision and the latency of `get` and `save`; the repository + records errors, the bulk hit and miss counts of `get_many`, and the latency of `delete` and `get_many`. One collector + handed to both — which is what the Dishka providers do — therefore counts each operation once. ## Infrastructure Layer diff --git a/docs/architecture.md b/docs/architecture.md index 5bc441d..80b7876 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -118,10 +118,15 @@ The library provides `IdempotencyMetricsProtocol` for observability: - `record_bulk_hit` - multiple hits in bulk operation - `record_bulk_miss` - multiple misses in bulk operation -Wire metrics via repository constructor: +Each metric has one owner, so a single collector can be handed to both layers without +double counting: the coordinator records hit, miss, collision and the latency of `get` and +`save`; the repository records errors, the bulk hit and miss counts of `get_many`, and the +latency of `delete` and `get_many`. ```python -repo = RedisAsyncIdempotencyRepository(redis, metrics=PrometheusMetrics()) +metrics = PrometheusMetrics() +repo = RedisAsyncIdempotencyRepository(redis, metrics=metrics) +coordinator = AsyncIdempotencyCoordinator(repo, IdempotencyDomainService(), metrics=metrics) ``` The library follows the **Idempotency Key Pattern**: diff --git a/docs/quickstart.md b/docs/quickstart.md index dcc737f..bb8baec 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -157,17 +157,17 @@ await repo.get("order.create", "abc123") # Different record How long the cached result should be kept: ```python -# Default: 30 minutes +# Default: 60 minutes record = service.create_record("op", "key", result) -# Custom TTL: 1 hour -record = service.create_record("op", "key", result, ttl_minutes=60) +# Custom TTL: 5 minutes +record = service.create_record("op", "key", result, ttl_minutes=5) # Service configuration service = IdempotencyDomainService( - default_ttl_minutes=30, # Default TTL - min_ttl_seconds=60, # Minimum: 1 minute - max_ttl_seconds=86400 # Maximum: 24 hours + default_ttl_minutes=60, # Default TTL + min_ttl_seconds=60, # Minimum: 1 minute + max_ttl_seconds=2592000 # Maximum: 30 days ) ``` diff --git a/docs/user_guide.md b/docs/user_guide.md index 5eee826..89141b2 100644 --- a/docs/user_guide.md +++ b/docs/user_guide.md @@ -184,7 +184,7 @@ The `RedisAsyncIdempotencyRepository` is fully compatible with Redis Cluster. It ## Metrics and Observability -You can inject a metrics collector into the repository to track hits, misses, collisions, and latency. +You can inject a metrics collector into the repository and the coordinator to track hits, misses, collisions, errors and latency. Each metric has one owner, so the same collector goes to both without double counting: the coordinator records hit, miss, collision and the latency of `get` and `save`; the repository records errors, the bulk hit and miss counts of `get_many`, and the latency of `delete` and `get_many`. ```python from idempotency_kit.core.protocols.metrics import IdempotencyMetricsProtocol @@ -218,7 +218,9 @@ class PrometheusMetrics(IdempotencyMetricsProtocol): # Increment bulk miss counter pass -repo = RedisAsyncIdempotencyRepository(redis, metrics=PrometheusMetrics()) +metrics = PrometheusMetrics() +repo = RedisAsyncIdempotencyRepository(redis, metrics=metrics) +coordinator = AsyncIdempotencyCoordinator(repo, IdempotencyDomainService(), metrics=metrics) ``` ## Configuration Reference @@ -227,9 +229,9 @@ repo = RedisAsyncIdempotencyRepository(redis, metrics=PrometheusMetrics()) ```python service = IdempotencyDomainService( - default_ttl_minutes=30, # Default: 30 minutes - min_ttl_seconds=60, # Default: 60 seconds (1 min) - max_ttl_seconds=86400 # Default: 86400 seconds (24 hours) + default_ttl_minutes=60, # Default: 60 minutes + min_ttl_seconds=60, # Default: 60 seconds (1 min) + max_ttl_seconds=2592000 # Default: 2592000 seconds (30 days) ) ``` @@ -245,6 +247,7 @@ repo = RedisAsyncIdempotencyRepository( ### Constants +These three defaults are the same numbers `BaseIdempotencySettings` uses for its own fields. See `idempotency_kit.core.constants` for: - `MAX_KEY_LENGTH`, `MAX_OPERATION_LENGTH` - `DEFAULT_TTL_MINUTES`, `MIN_TTL_SECONDS`, `MAX_TTL_SECONDS` @@ -552,6 +555,8 @@ def create_api_container(settings: Settings) -> AsyncContainer: Your own providers supply the `Redis` client and the settings object; the settings object must satisfy `IdempotencySettingsProtocol`, which `BaseIdempotencySettings` does. `IdempotencyProvider` also provides the metrics collector: `PrometheusIdempotencyMetrics` when `metrics_enabled` is true (install the `prometheus` extra), a no-op collector otherwise. To use another metrics backend, provide `IdempotencyMetricsProtocol` yourself with `@provide(override=True)` in a provider listed after `IdempotencyProvider()`. +`enabled` on the settings object is the kill switch: with `enabled=False` the coordinator these providers build runs the action and nothing else — no read, no write, no metric. A settings object that predates the field is read as enabled. + ## Migration Guide ### From older versions or other libraries From 1bfbc6e2fd33c3a2c870e35f82f8e95fc4510812 Mon Sep 17 00:00:00 2001 From: Alex Shalaev Date: Sun, 6 Sep 2026 21:29:43 +0300 Subject: [PATCH 9/9] test: cover the branches the fixes added The decode guard for a null payload, the infra_param-on-self lookup and the fallback for a callable inspect cannot describe. --- tests/unit/core/test_adapter_round_trip.py | 10 +++++ tests/unit/core/test_idempotent.py | 51 +++++++++++++++++++++- 2 files changed, 60 insertions(+), 1 deletion(-) diff --git a/tests/unit/core/test_adapter_round_trip.py b/tests/unit/core/test_adapter_round_trip.py index 5124d6f..847c11f 100644 --- a/tests/unit/core/test_adapter_round_trip.py +++ b/tests/unit/core/test_adapter_round_trip.py @@ -109,3 +109,13 @@ async def action() -> Any: (("op.absent", "record_validation_error"),), (("op.absent", "record_validation_error"),), ] + + +def test__pydantic_adapter__null_payload__raises() -> None: + """A stored null is the one payload this adapter cannot turn back into a model.""" + # Arrange + adapter: PydanticResultAdapter[_Order] = PydanticResultAdapter(_Order) + + # Act & Assert + with pytest.raises(ValueError, match="null idempotency payload"): + adapter.decode(None) diff --git a/tests/unit/core/test_idempotent.py b/tests/unit/core/test_idempotent.py index 8a5ebc2..a30e192 100644 --- a/tests/unit/core/test_idempotent.py +++ b/tests/unit/core/test_idempotent.py @@ -1,7 +1,8 @@ """Unit tests for async idempotent decorator.""" import logging -from unittest.mock import ANY, MagicMock +from typing import Any +from unittest.mock import ANY, MagicMock, patch import pytest @@ -196,3 +197,51 @@ async def my_func(*, idempotency_key: str | None = None) -> str: "No idempotency coordinator found; running the operation without idempotency" ] assert caplog.records[0].operation == operation # type: ignore[attr-defined] + + +@pytest.mark.asyncio +async def test__decorator__infra_param_on_self__calls_coordinator( + mock_coordinator: MagicMock, mock_adapter: MagicMock +) -> None: + """infra_param names an attribute of the instance, not only a keyword argument.""" + # Arrange + mock_coordinator.coordinate.return_value = "ok" + + class MyService: + def __init__(self, coordinator: AsyncIdempotencyCoordinator) -> None: + self._idempotency = coordinator + + @async_idempotent(operation="test.op", adapter=mock_adapter, infra_param="_idempotency") + async def my_method(self, *, idempotency_key: str | None = None) -> str: + return "not used" + + service = MyService(mock_coordinator) + + # Act + result = await service.my_method(idempotency_key="test-key") + + # Assert + assert result == "ok" + mock_coordinator.coordinate.assert_called_once() + + +@pytest.mark.asyncio +async def test__decorator__uninspectable_signature__still_reads_the_key_from_kwargs( + mock_coordinator: MagicMock, mock_adapter: MagicMock +) -> None: + """A callable inspect cannot describe keeps the keyword-argument lookup.""" + # Arrange + mock_coordinator.coordinate.return_value = "ok" + + with patch("inspect.signature", side_effect=ValueError("no signature found")): + + @async_idempotent(operation="test.op", adapter=mock_adapter) + async def my_func(**kwargs: Any) -> str: + return "not used" + + # Act + result = await my_func(idempotency_key="test-key", coord=mock_coordinator) + + # Assert + assert result == "ok" + mock_coordinator.coordinate.assert_called_once()