From 01433d8bd1f5b14169dc4e4cf29872c52e728cc5 Mon Sep 17 00:00:00 2001 From: Alexey Shalaev <75322386+AlexeyShalaev@users.noreply.github.com> Date: Mon, 14 Sep 2026 13:52:51 +0300 Subject: [PATCH] fix!: request Redis | RedisCluster in the Dishka provider and read per slot on a cluster AsyncRedisIdempotencyProvider asked for redis.asyncio.Redis while redis_client_kit.providers.AsyncRedisProvider provides the union alias AsyncRedisClient = Redis | RedisCluster. Dishka resolves by exact key, so a container holding both never built. The provider now requests the union, which is structurally the same key, so no shared alias or dependency is needed; the test group gains redis-client-kit so the wiring from the issue is tested against the real provider. The repository takes the union too, and get_many uses mget_nonatomic on a RedisCluster: a single MGET cannot span hash slots there, so the documented cluster compatibility was false for that one method. BREAKING CHANGE: AsyncRedisIdempotencyProvider requests redis.asyncio.Redis | RedisCluster, the key redis_client_kit.AsyncRedisClient names. A provider of your own annotated `-> Redis` no longer matches and the container fails at construction with GraphMissingFactoryError; annotate it `-> Redis | RedisCluster`. Closes #32 --- docs/agents.md | 24 ++++++--- docs/api_reference.md | 4 +- docs/architecture.md | 2 + docs/user_guide.md | 4 +- idempotency_kit/dishka/aio/redis.py | 12 +++-- .../infra/storage/redis/aio/repository.py | 13 +++-- pyproject.toml | 1 + tests/unit/dishka/test_providers.py | 49 +++++++++++++++++-- tests/unit/storage/redis/aio/test_bulk.py | 20 ++++++++ uv.lock | 39 +++++++++++++++ 10 files changed, 148 insertions(+), 20 deletions(-) diff --git a/docs/agents.md b/docs/agents.md index 5a3735a..3865f90 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -213,11 +213,13 @@ result = await coordinator.coordinate( | `delete_many(operation, idempotency_keys)` | `int` | as `delete` | `RedisAsyncIdempotencyRepository(redis, *, key_prefix="idempotency:", metrics=None)` is the -implementation: `SET key value EX ceil(record.ttl_seconds) NX` for `save`, the same without -`NX` for `replace`, `GET` for `get`, `MGET` for `get_many`, and a non-transactional pipeline -for `save_many` so it works on Redis Cluster. It deletes any record it reads back expired and -reports that as a miss. A repository of your own needs `replace` too: the coordinator raises -`TypeError` at construction without it, unless `in_flight="run"`. +implementation, and `redis` is a `redis.asyncio.Redis` or a `RedisCluster`: `SET key value EX +ceil(record.ttl_seconds) NX` for `save`, the same without `NX` for `replace`, `GET` for `get`, +`MGET` for `get_many` — one per hash slot, via redis-py's `mget_nonatomic`, on a cluster — +and a non-transactional pipeline for `save_many`, so all of it works on Redis Cluster. It +deletes any record it reads back expired and reports that as a miss. A repository of your own +needs `replace` too: the coordinator raises `TypeError` at construction without it, unless +`in_flight="run"`. ### Metrics @@ -259,7 +261,7 @@ from idempotency_kit.dishka import ( ) container = make_async_container( - MyRedisProvider(), # provides redis.asyncio.Redis + MyRedisProvider(), # provides redis.asyncio.Redis | RedisCluster MySettingsProvider(), # provides IdempotencySettingsProtocol IdempotencyProvider(), # IdempotencyDomainService + IdempotencyMetricsProtocol AsyncRedisIdempotencyProvider(), @@ -267,6 +269,11 @@ container = make_async_container( ) ``` +`AsyncRedisIdempotencyProvider` asks for `redis.asyncio.Redis | RedisCluster`, exactly — the +key `redis_client_kit.AsyncRedisClient` names, so `redis_client_kit.providers.AsyncRedisProvider` +takes the place of `MyRedisProvider` as is. Dishka matches keys, not subclasses: a provider of +your own annotated `-> Redis` is a different key, and the container fails to build. + 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. @@ -391,6 +398,11 @@ fields existed is read as enabled, `"wait"` and 30 seconds. `in_flight="run"` mode the fingerprint is checked on the read before the action; a mismatch discovered on the collision after the action is logged and counted as `key_reuse`, and the caller keeps its own result, because the side effect has happened. +25. **The Dishka key for the client is `redis.asyncio.Redis | RedisCluster`, exactly.** + `AsyncRedisIdempotencyProvider` requests that union — the key + `redis_client_kit.AsyncRedisClient` names — and Dishka matches keys, not subclasses. A + provider annotated `-> Redis` fails the container at construction with + `GraphMissingFactoryError`; change the annotation, not the client. ## Common mistakes diff --git a/docs/api_reference.md b/docs/api_reference.md index e01373d..5a82cc3 100644 --- a/docs/api_reference.md +++ b/docs/api_reference.md @@ -105,10 +105,10 @@ Interface for metrics collection. Redis implementation of the repository protocol. - **Constructor**: - - `redis` (`redis.asyncio.Redis`): Any async Redis client, including a subclass such as an instrumented or fake one. + - `redis` (`redis.asyncio.Redis | RedisCluster`): Any async Redis client, single-node or cluster, 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**) -- **Storage**: `SET key value EX ttl NX` for `save`, the same without `NX` for `replace`, `GET` for `get`. +- **Storage**: `SET key value EX ttl NX` for `save`, the same without `NX` for `replace`, `GET` for `get`, `MGET` for `get_many` — one per hash slot, via `mget_nonatomic`, on a `RedisCluster`. ## Exceptions diff --git a/docs/architecture.md b/docs/architecture.md index 0003c87..4a50656 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -147,6 +147,8 @@ This distinction allows developers to decide whether to fail the request or proc Bulk operations (`get_many`, `save_many`, `delete_many`) are designed to be efficient by using Redis `MGET` and non-transactional pipelines. +**`MGET` per hash slot**: on a cluster a single `MGET` may not span hash slots — redis-py raises `RedisClusterException` before sending — so with a `RedisCluster` client `get_many` calls `mget_nonatomic`, which issues one `MGET` per slot and returns the values in input order. `delete_many` and the expired-key cleanup rely on the cluster client's own per-slot split of `DEL`. + **Non-transactional pipelines** (`transaction=False`) are used for `save_many` to ensure compatibility with Redis Cluster. In a cluster environment, different keys can map to different hash slots, making standard `MULTI/EXEC` transactions impossible for arbitrary keys. By using a non-transactional pipeline, we send all commands in a single network round-trip while allowing them to be processed independently across different cluster nodes. ## Metrics and Observability diff --git a/docs/user_guide.md b/docs/user_guide.md index f0f2e9d..1b9410d 100644 --- a/docs/user_guide.md +++ b/docs/user_guide.md @@ -327,7 +327,7 @@ except IdempotencyValidationError as e: ## Redis Cluster Compatibility -The `RedisAsyncIdempotencyRepository` is fully compatible with Redis Cluster. It uses non-transactional pipelines (`transaction=False`) for bulk operations, allowing keys to be distributed across different hash slots. +The `RedisAsyncIdempotencyRepository` takes a `redis.asyncio.RedisCluster` as well as a `Redis` and is fully compatible with Redis Cluster. Every write is single-key. `get_many` sends one `MGET` per hash slot on a cluster (redis-py's `mget_nonatomic`), because a single `MGET` cannot span slots. `save_many` uses a non-transactional pipeline (`transaction=False`), allowing keys to be distributed across different hash slots. **Note on Atomicity**: Since non-transactional pipelines are used, `save_many` is **not atomic** by default. If an error occurs, some records might remain in Redis. Use `rollback_on_error=True` if you need to ensure that either all records are saved or none (the library will manually delete successfully saved records if a failure occurs). @@ -742,7 +742,7 @@ 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()`. +Your own providers supply the Redis client and the settings object. The client is requested as `redis.asyncio.Redis | RedisCluster` — the key `redis_client_kit.AsyncRedisClient` names, so `redis_client_kit.providers.AsyncRedisProvider` stands in for `RedisProvider()` above with no adapter in between. A provider of your own must use that exact annotation: Dishka matches keys exactly, and `Redis` alone is a different one. 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. `in_flight` and `in_flight_lease_seconds` reach the coordinator the same way. A settings object that predates these fields is read as enabled, `"wait"` and 30 seconds. diff --git a/idempotency_kit/dishka/aio/redis.py b/idempotency_kit/dishka/aio/redis.py index 1a6404f..b785c89 100644 --- a/idempotency_kit/dishka/aio/redis.py +++ b/idempotency_kit/dishka/aio/redis.py @@ -1,7 +1,7 @@ """Dishka provider for Redis-backed async idempotency repository.""" from dishka import Provider, Scope, provide -from redis.asyncio import Redis as AsyncRedisClient +from redis.asyncio import Redis, RedisCluster from idempotency_kit import AsyncIdempotencyRepository, IdempotencyMetricsProtocol from idempotency_kit.infra.storage.redis.aio import RedisAsyncIdempotencyRepository @@ -10,14 +10,20 @@ class AsyncRedisIdempotencyProvider(Provider): - """Provider for async Redis-backed idempotency repository.""" + """Provider for async Redis-backed idempotency repository. + + The client is requested as ``Redis | RedisCluster`` -- the key + ``redis_client_kit.AsyncRedisClient`` names, so redis-client-kit's + ``AsyncRedisProvider`` satisfies it as is. Dishka matches keys exactly: + a provider of your own has to use that annotation, not ``Redis`` alone. + """ scope = Scope.APP @provide def get_repository( self, - redis: AsyncRedisClient, + redis: Redis | RedisCluster, settings: IdempotencySettingsProtocol, metrics: IdempotencyMetricsProtocol, ) -> AsyncIdempotencyRepository: diff --git a/idempotency_kit/infra/storage/redis/aio/repository.py b/idempotency_kit/infra/storage/redis/aio/repository.py index d1fd6ae..79b410d 100644 --- a/idempotency_kit/infra/storage/redis/aio/repository.py +++ b/idempotency_kit/infra/storage/redis/aio/repository.py @@ -17,9 +17,10 @@ _HAS_ORJSON = False try: - from redis.asyncio import Redis as AsyncRedisClient + from redis.asyncio import Redis, RedisCluster _HAS_REDIS = True + AsyncRedisClient = Redis | RedisCluster except ImportError: _HAS_REDIS = False AsyncRedisClient = Any # type: ignore[assignment,misc,valid-type] @@ -61,7 +62,7 @@ def __init__( """Initialize repository. Args: - redis: Redis client instance + redis: Redis client instance, single-node or cluster key_prefix: Prefix for all Redis keys (default: "idempotency:") metrics: Metrics collector instance (default: NoOp) """ @@ -316,7 +317,7 @@ async def delete(self, operation: str, idempotency_key: str) -> bool: self._metrics.record_latency(operation, "delete", time.perf_counter() - start) async def get_many(self, operation: str, idempotency_keys: list[str]) -> dict[str, IdempotencyRecord]: - """Retrieve multiple records from Redis using MGET. + """Retrieve multiple records from Redis using MGET, one per hash slot on a cluster. Args: operation: Name of the operation @@ -340,7 +341,11 @@ async def get_many(self, operation: str, idempotency_keys: list[str]) -> dict[st keys = [self._make_key(operation, key) for key in idempotency_keys] try: - values = await self._redis.mget(keys) + # One MGET cannot span hash slots on a cluster; mget_nonatomic sends one per slot. + if isinstance(self._redis, RedisCluster): + values = await self._redis.mget_nonatomic(keys) + else: + values = await self._redis.mget(keys) except Exception as e: self._metrics.record_error(operation, type(e).__name__) logger.exception( diff --git a/pyproject.toml b/pyproject.toml index c8b2708..41550cc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,6 +52,7 @@ test = [ "orjson>=3.11.7,<4.0.0", "dishka>=1.0.0", "prometheus-client>=0.15.0", + "redis-client-kit[providers,settings]>=0.2.0", ] docs = [ "zensical>=0.0.37", diff --git a/tests/unit/dishka/test_providers.py b/tests/unit/dishka/test_providers.py index 52b19aa..e857b24 100644 --- a/tests/unit/dishka/test_providers.py +++ b/tests/unit/dishka/test_providers.py @@ -4,7 +4,10 @@ from dishka import Provider, Scope, make_async_container, provide from fakeredis.aioredis import FakeRedis from prometheus_client import REGISTRY -from redis.asyncio import Redis as AsyncRedisClient +from redis.asyncio import Redis, RedisCluster +from redis_client_kit.config import RedisSettingsProtocol +from redis_client_kit.providers import AsyncRedisProvider +from redis_client_kit.settings import BaseRedisSettings from idempotency_kit import ( AsyncIdempotencyCoordinator, @@ -21,6 +24,7 @@ IdempotencySettingsProtocol, ) from idempotency_kit.infra.metrics.prometheus import PrometheusIdempotencyMetrics +from idempotency_kit.infra.storage.redis.aio import RedisAsyncIdempotencyRepository from idempotency_kit.settings import BaseIdempotencySettings @@ -55,11 +59,27 @@ def settings(self) -> IdempotencySettingsProtocol: ) @provide - def redis(self) -> AsyncRedisClient: - """Provide a fake Redis client (FakeRedis subclasses redis.asyncio.Redis).""" + def redis(self) -> Redis | RedisCluster: + """Provide a fake Redis client (FakeRedis subclasses redis.asyncio.Redis) under the shipped provider's key.""" return FakeRedis() +class _RedisClientKitAppProvider(Provider): + """Application-side provider for the redis-client-kit wiring: its settings object and ours.""" + + scope = Scope.APP + + @provide + def redis_settings(self) -> RedisSettingsProtocol: + """Provide redis-client-kit settings; nothing connects while the startup health check is off.""" + return BaseRedisSettings(key_prefix="probe") + + @provide + def settings(self) -> IdempotencySettingsProtocol: + """Provide idempotency settings.""" + return BaseIdempotencySettings(key_prefix="probe:", metrics_enabled=False) + + @pytest.mark.asyncio async def test__shipped_providers__metrics_disabled__container_resolves_coordinator() -> None: """Test that the shipped providers alone can build a container and resolve the coordinator.""" @@ -221,3 +241,26 @@ async def test__shipped_providers__in_flight_settings__reach_the_coordinator() - assert coordinator._in_flight_lease_seconds == 5 finally: await container.close() + + +@pytest.mark.asyncio +async def test__shipped_providers__redis_client_kit_client__container_builds_and_resolves_repository() -> None: + """The client redis-client-kit provides is the one the Redis provider asks for: no adapter in between (#32).""" + # Arrange + container = make_async_container( + AsyncRedisProvider(check_health_on_startup=False), + _RedisClientKitAppProvider(), + IdempotencyProvider(), + AsyncRedisIdempotencyProvider(), + AsyncIdempotencyCoordinatorProvider(), + ) + + # Act + try: + repository = await container.get(AsyncIdempotencyRepository) + + # Assert + assert isinstance(repository, RedisAsyncIdempotencyRepository) + assert isinstance(repository._redis, Redis) + finally: + await container.close() diff --git a/tests/unit/storage/redis/aio/test_bulk.py b/tests/unit/storage/redis/aio/test_bulk.py index 49bfc72..d15b866 100644 --- a/tests/unit/storage/redis/aio/test_bulk.py +++ b/tests/unit/storage/redis/aio/test_bulk.py @@ -6,6 +6,7 @@ import orjson import pytest from fakeredis import FakeAsyncRedis as AsyncRedisClient +from redis.asyncio import RedisCluster from idempotency_kit import ( IdempotencyDomainService, @@ -364,3 +365,22 @@ async def test_save_many_mixed_validation_errors(fake_redis: AsyncRedisClient) - with pytest.raises(IdempotencyValidationError, match="Bulk validation failed"): await repo.save_many([invalid_op, invalid_key]) + + +@pytest.mark.asyncio +async def test__get_many__cluster_client__reads_with_one_mget_per_slot() -> None: + """A single MGET cannot span hash slots on a cluster, so the repository asks redis-py for one per slot.""" + # Arrange + record = IdempotencyDomainService().create_record(operation="bulk", idempotency_key="key1", result={"n": 1}) + cluster = AsyncMock(spec=RedisCluster) + cluster.mget_nonatomic.return_value = [orjson.dumps(record.model_dump(mode="json")), None] + repo = RedisAsyncIdempotencyRepository(cluster) + + # Act + results = await repo.get_many("bulk", ["key1", "key2"]) + + # Assert + assert set(results) == {"key1"} + assert results["key1"].result == {"n": 1} + cluster.mget_nonatomic.assert_awaited_once_with(["idempotency:bulk:key1", "idempotency:bulk:key2"]) + cluster.mget.assert_not_called() diff --git a/uv.lock b/uv.lock index 3a8c4de..b160c2b 100644 --- a/uv.lock +++ b/uv.lock @@ -435,6 +435,7 @@ dev = [ { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "redis" }, + { name = "redis-client-kit", extra = ["providers", "settings"] }, { name = "ruff" }, { name = "testcontainers" }, ] @@ -451,6 +452,7 @@ test = [ { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "redis" }, + { name = "redis-client-kit", extra = ["providers", "settings"] }, { name = "testcontainers" }, ] @@ -476,6 +478,7 @@ dev = [ { name = "pytest-asyncio", specifier = ">=1.3.0" }, { name = "pytest-cov", specifier = ">=7.0.0" }, { name = "redis", specifier = ">=5.0.0" }, + { name = "redis-client-kit", extras = ["providers", "settings"], specifier = ">=0.2.0" }, { name = "ruff", specifier = ">=0.15.1" }, { name = "testcontainers", specifier = ">=4.14.1" }, ] @@ -492,6 +495,7 @@ test = [ { name = "pytest-asyncio", specifier = ">=1.3.0" }, { name = "pytest-cov", specifier = ">=7.0.0" }, { name = "redis", specifier = ">=5.0.0" }, + { name = "redis-client-kit", extras = ["providers", "settings"], specifier = ">=0.2.0" }, { name = "testcontainers", specifier = ">=4.14.1" }, ] @@ -1144,6 +1148,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/db/1d/068464f23075f66a8f1b806935e9cd9363ee446636ea70d2c22ee8659dbf/pydantic_core-2.46.5-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:d0a24b40877af2de4950252be9d21eaf7fb07660f3c2cae1f56c6b599ada5266", size = 2140686, upload-time = "2026-08-28T10:01:28.947Z" }, ] +[[package]] +name = "pydantic-settings" +version = "2.15.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pydantic" }, + { name = "python-dotenv" }, + { name = "typing-inspection" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/68/ca/31c57507b13119d7d3cfa1576dad2911a4861e3be07b579395f4e9d393f9/pydantic_settings-2.15.0.tar.gz", hash = "sha256:694b793e84f766ba76a90ebdefc01d0a9a045dab0382bee70393da93712ad117", size = 261253, upload-time = "2026-08-07T09:24:57.419Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/30/a4/2bffa9f8e804325a09867f0e9d30795c80ea9f8d62560bd1b6ad6220eb2f/pydantic_settings-2.15.0-py3-none-any.whl", hash = "sha256:0ba092c291c94baceb5eff768aa0d56400a457585bc0175925a5a5510303da42", size = 69413, upload-time = "2026-08-07T09:24:55.839Z" }, +] + [[package]] name = "pygments" version = "2.20.0" @@ -1341,6 +1359,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/66/9d/c5731f6e3608663d4d3656fd8d3aecee8b509c3082818f5a13eae925baea/redis-8.1.0-py3-none-any.whl", hash = "sha256:a4fe1aac3d3b3cc791d4b3d5931c5a956045dc951ee74d1c913ee3ac4d2ee9fb", size = 560618, upload-time = "2026-07-30T08:50:58.497Z" }, ] +[[package]] +name = "redis-client-kit" +version = "0.2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "redis" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/de/1c/4d6b4a063724afe1cfc27dea01342b6ee752242b088275959db954616aaf/redis_client_kit-0.2.0.tar.gz", hash = "sha256:c4eed2444b86952ff34b8b13862e3e3d595d49a1b74b997a98a74aede8b0f5df", size = 24031, upload-time = "2026-09-07T10:53:01.863Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2d/94/286de5e76e2aae1b49d4fa9690ed74cc587f1ae733813bd3df151634fd84/redis_client_kit-0.2.0-py3-none-any.whl", hash = "sha256:b02a3b20025cd748afda54febc27f291ce9090347d1ac24238f8fc0be194f167", size = 33787, upload-time = "2026-09-07T10:53:00.859Z" }, +] + +[package.optional-dependencies] +providers = [ + { name = "dishka" }, +] +settings = [ + { name = "pydantic" }, + { name = "pydantic-settings" }, +] + [[package]] name = "requests" version = "2.34.0"