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 323958e..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 @@ -66,7 +70,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..80b7876 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. @@ -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 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/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/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/idempotency_kit/core/services/aio/coordinator.py b/idempotency_kit/core/services/aio/coordinator.py index 7ec789d..a394e83 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 @@ -32,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, @@ -40,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, @@ -57,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 @@ -163,6 +178,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/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/infra/storage/redis/aio/repository.py b/idempotency_kit/infra/storage/redis/aio/repository.py index 51d4cf9..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__( @@ -61,8 +67,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 @@ -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/idempotency_kit/settings.py b/idempotency_kit/settings.py index 320c65f..bc01d44 100644 --- a/idempotency_kit/settings.py +++ b/idempotency_kit/settings.py @@ -2,16 +2,23 @@ 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.""" - 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") - 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_adapter_round_trip.py b/tests/unit/core/test_adapter_round_trip.py index 50c1044..847c11f 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,45 @@ 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"),), + ] + + +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_coordinator.py b/tests/unit/core/test_coordinator.py index db2d514..b516dd3 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,62 @@ 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") + + +@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/core/test_idempotent.py b/tests/unit/core/test_idempotent.py index db65a77..a30e192 100644 --- a/tests/unit/core/test_idempotent.py +++ b/tests/unit/core/test_idempotent.py @@ -1,6 +1,8 @@ """Unit tests for async idempotent decorator.""" -from unittest.mock import ANY, MagicMock +import logging +from typing import Any +from unittest.mock import ANY, MagicMock, patch import pytest @@ -123,3 +125,123 @@ 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] + + +@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() 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) 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() diff --git a/tests/unit/storage/redis/aio/test_repository.py b/tests/unit/storage/redis/aio/test_repository.py index 1ec8d53..e58ca7d 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 @@ -8,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 @@ -283,6 +287,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.""" @@ -420,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 ( @@ -463,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: